diff --git a/README.md b/README.md index 1b08aaa..567c9d4 100644 --- a/README.md +++ b/README.md @@ -15,6 +15,11 @@ 5、SpringBoot2.0 整合 QuartJob ,实现定时器实时管理
+ +6、SpringBoot2.0 整合 Redis集群 ,实现消息队列场景
+ +持续更新中... + ## 项目简介 SpringBoot 集成常用中间件 diff --git a/pom.xml b/pom.xml index e66d22a..4ccb44c 100644 --- a/pom.xml +++ b/pom.xml @@ -23,12 +23,12 @@ ware-rocket-queue ware-shard-jdbc - - - + ware-email-send ware-swagger-two + + ware-redis-cluster diff --git a/ware-redis-cluster/pom.xml b/ware-redis-cluster/pom.xml new file mode 100644 index 0000000..fb90253 --- /dev/null +++ b/ware-redis-cluster/pom.xml @@ -0,0 +1,88 @@ + + + + middle-ware-parent + com.boot.parent + 1.0-SNAPSHOT + + 4.0.0 + + com.redis.cluster + ware-redis-cluster + jar + + + + org.springframework.boot + spring-boot-starter-data-redis + ${spring-boot.version} + + + redis.clients + jedis + ${redis-client.version} + + + + org.springframework.boot + spring-boot-starter-web + ${spring-boot.version} + + + org.springframework.boot + spring-boot-starter-test + ${spring-boot.version} + test + + + org.springframework.boot + spring-boot-starter-aop + ${spring-boot.version} + + + org.springframework.boot + spring-boot-configuration-processor + ${spring-boot.version} + true + + + org.springframework + spring-context-support + ${spring.version} + + + joda-time + joda-time + ${joda.time.version} + + + com.alibaba + fastjson + ${fastjson.version} + + + + + + ${project.artifactId} + + + src/main/resources + true + + + + + org.apache.maven.plugins + maven-compiler-plugin + 2.3.2 + + 1.8 + 1.8 + + + + + \ No newline at end of file diff --git a/ware-redis-cluster/src/main/java/com/redis/cluster/RedisApplication.java b/ware-redis-cluster/src/main/java/com/redis/cluster/RedisApplication.java new file mode 100644 index 0000000..84e592d --- /dev/null +++ b/ware-redis-cluster/src/main/java/com/redis/cluster/RedisApplication.java @@ -0,0 +1,12 @@ +package com.redis.cluster; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration; + +@SpringBootApplication(exclude= {DataSourceAutoConfiguration.class}) +public class RedisApplication { + public static void main(String[] args) { + SpringApplication.run(RedisApplication.class,args) ; + } +} diff --git a/ware-redis-cluster/src/main/java/com/redis/cluster/config/RedisConfig.java b/ware-redis-cluster/src/main/java/com/redis/cluster/config/RedisConfig.java new file mode 100644 index 0000000..a9cb62a --- /dev/null +++ b/ware-redis-cluster/src/main/java/com/redis/cluster/config/RedisConfig.java @@ -0,0 +1,17 @@ +package com.redis.cluster.config; + +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.core.StringRedisTemplate; + +@Configuration +public class RedisConfig { + + @Bean + public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory factory) { + StringRedisTemplate stringRedisTemplate = new StringRedisTemplate(); + stringRedisTemplate.setConnectionFactory(factory); + return stringRedisTemplate; + } +} diff --git a/ware-redis-cluster/src/main/java/com/redis/cluster/config/RedisLock.java b/ware-redis-cluster/src/main/java/com/redis/cluster/config/RedisLock.java new file mode 100644 index 0000000..3cd3987 --- /dev/null +++ b/ware-redis-cluster/src/main/java/com/redis/cluster/config/RedisLock.java @@ -0,0 +1,41 @@ +package com.redis.cluster.config; + +import org.springframework.stereotype.Component; +import redis.clients.jedis.Jedis; +import redis.clients.jedis.JedisSentinelPool; +import javax.annotation.Resource; + +@Component +public class RedisLock { + private static String keyPrefix = "RedisLock:"; + @Resource + private JedisSentinelPool jedisSentinelPool; + + public boolean addLock(String key, long expire) { + Jedis jedis = null; + try { + jedis = jedisSentinelPool.getResource(); + /* + * nxxx的值只能取NX或者XX，如果取NX，则只有当key不存在是才进行set，如果取XX，则只有当key已经存在时才进行set + * expx的值只能取EX或者PX，代表数据过期时间的单位，EX代表秒，PX代表毫秒。 + */ + String value = jedis.set(keyPrefix + key, "1", "nx", "ex", expire); + return value != null; + } catch (Exception e){ + e.printStackTrace(); + }finally { + if (jedis != null) jedis.close(); + } + return false; + } + + public void removeLock(String key) { + Jedis jedis = null; + try { + jedis = jedisSentinelPool.getResource(); + jedis.del(keyPrefix + key); + } finally { + if (jedis != null) jedis.close(); + } + } +} diff --git a/ware-redis-cluster/src/main/java/com/redis/cluster/config/RedisParam.java b/ware-redis-cluster/src/main/java/com/redis/cluster/config/RedisParam.java new file mode 100644 index 0000000..e1e6c74 --- /dev/null +++ b/ware-redis-cluster/src/main/java/com/redis/cluster/config/RedisParam.java @@ -0,0 +1,86 @@ +package com.redis.cluster.config; + +import org.springframework.boot.context.properties.ConfigurationProperties; + +@ConfigurationProperties(prefix = "spring.redis.sentinel") +public class RedisParam { + + private String nodes ; + private String master ; + private Integer maxTotal ; + private Integer minIdle ; + private Integer maxWaitMillis ; + private Integer timeBetweenEvictionRunsMillis ; + private boolean testWhileIdle ; + private boolean testOnBorrow ; + private boolean testOnReturn ; + + public String getNodes() { + return nodes; + } + public void setNodes(String nodes) { + this.nodes = nodes; + } + public String getMaster() { + return master; + } + public void setMaster(String master) { + this.master = master; + } + + public Integer getMaxTotal() { + return maxTotal; + } + + public void setMaxTotal(Integer maxTotal) { + this.maxTotal = maxTotal; + } + + public Integer getMinIdle() { + return minIdle; + } + + public void setMinIdle(Integer minIdle) { + this.minIdle = minIdle; + } + + public Integer getMaxWaitMillis() { + return maxWaitMillis; + } + + public void setMaxWaitMillis(Integer maxWaitMillis) { + this.maxWaitMillis = maxWaitMillis; + } + + public Integer getTimeBetweenEvictionRunsMillis() { + return timeBetweenEvictionRunsMillis; + } + + public void setTimeBetweenEvictionRunsMillis(Integer timeBetweenEvictionRunsMillis) { + this.timeBetweenEvictionRunsMillis = timeBetweenEvictionRunsMillis; + } + + public boolean isTestWhileIdle() { + return testWhileIdle; + } + + public void setTestWhileIdle(boolean testWhileIdle) { + this.testWhileIdle = testWhileIdle; + } + + public boolean isTestOnBorrow() { + return testOnBorrow; + } + + public void setTestOnBorrow(boolean testOnBorrow) { + this.testOnBorrow = testOnBorrow; + } + + public boolean isTestOnReturn() { + return testOnReturn; + } + + public void setTestOnReturn(boolean testOnReturn) { + this.testOnReturn = testOnReturn; + } +} diff --git a/ware-redis-cluster/src/main/java/com/redis/cluster/config/RedisPool.java b/ware-redis-cluster/src/main/java/com/redis/cluster/config/RedisPool.java new file mode 100644 index 0000000..8d6788a --- /dev/null +++ b/ware-redis-cluster/src/main/java/com/redis/cluster/config/RedisPool.java @@ -0,0 +1,45 @@ +package com.redis.cluster.config; + +import com.redis.cluster.listen.RedisListener; +import com.redis.cluster.utils.SpringUtil; +import org.apache.commons.pool2.impl.GenericObjectPoolConfig; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import redis.clients.jedis.JedisSentinelPool; +import javax.annotation.Resource; +import java.util.Arrays; +import java.util.HashSet; +import java.util.Set; + +@Configuration +@EnableConfigurationProperties(RedisParam.class) +public class RedisPool { + + @Resource + private RedisParam redisParam ; + + @Bean("jedisSentinelPool") + public JedisSentinelPool getRedisPool (){ + Set sentinels = new HashSet<>(); + sentinels.addAll(Arrays.asList(redisParam.getNodes().split(","))); + GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig(); + poolConfig.setMaxTotal(redisParam.getMaxTotal()); + poolConfig.setMinIdle(redisParam.getMinIdle()); + poolConfig.setMaxWaitMillis(redisParam.getMaxWaitMillis()); + poolConfig.setTestWhileIdle(redisParam.isTestWhileIdle()); + poolConfig.setTestOnBorrow(redisParam.isTestOnBorrow()); + poolConfig.setTestOnReturn(redisParam.isTestOnReturn()); + poolConfig.setTimeBetweenEvictionRunsMillis(redisParam.getTimeBetweenEvictionRunsMillis()); + JedisSentinelPool redisPool = new JedisSentinelPool(redisParam.getMaster(), sentinels, poolConfig); + return redisPool; + } + @Bean + SpringUtil springUtil() { + return new SpringUtil(); + } + @Bean + RedisListener redisListener() { + return new RedisListener(); + } +} diff --git a/ware-redis-cluster/src/main/java/com/redis/cluster/controller/MsgBody.java b/ware-redis-cluster/src/main/java/com/redis/cluster/controller/MsgBody.java new file mode 100644 index 0000000..1da3ef2 --- /dev/null +++ b/ware-redis-cluster/src/main/java/com/redis/cluster/controller/MsgBody.java @@ -0,0 +1,36 @@ +package com.redis.cluster.controller; + +import java.util.Date; + +public class MsgBody { + + private String name ; + + private String desc ; + + private Date createTime ; + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + public String getDesc() { + return desc; + } + + public void setDesc(String desc) { + this.desc = desc; + } + + public Date getCreateTime() { + return createTime; + } + + public void setCreateTime(Date createTime) { + this.createTime = createTime; + } +} diff --git a/ware-redis-cluster/src/main/java/com/redis/cluster/controller/RedisController.java b/ware-redis-cluster/src/main/java/com/redis/cluster/controller/RedisController.java new file mode 100644 index 0000000..62d9421 --- /dev/null +++ b/ware-redis-cluster/src/main/java/com/redis/cluster/controller/RedisController.java @@ -0,0 +1,48 @@ +package com.redis.cluster.controller; + +import com.alibaba.fastjson.JSONObject; +import com.redis.cluster.service.RedisService; +import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; +import redis.clients.jedis.Jedis; +import redis.clients.jedis.JedisSentinelPool; +import javax.annotation.Resource; +import java.util.Date; + +@RestController +public class RedisController { + + @Resource + private RedisService redisService ; + @Resource + private StringRedisTemplate stringRedisTemplate; + @Resource + private JedisSentinelPool jedisSentinelPool; + + @RequestMapping("/increment") + public String increment (){ + stringRedisTemplate.opsForValue().increment("cicada-key",10L) ; + return "success" ; + } + + @RequestMapping("/sentinel") + public String sentinel (){ + Jedis jedis = jedisSentinelPool.getResource() ; + jedis.set("sentinel","sentinel") ; + return "success" ; + } + + /** + * 队列推消息 + */ + @RequestMapping("/saveQueue") + public String saveQueue (){ + MsgBody msgBody = new MsgBody() ; + msgBody.setName("LogAModel"); + msgBody.setDesc("描述"); + msgBody.setCreateTime(new Date()); + redisService.saveQueue("LogA-key", JSONObject.toJSONString(msgBody)); + return "success" ; + } +} diff --git a/ware-redis-cluster/src/main/java/com/redis/cluster/listen/LogAListen.java b/ware-redis-cluster/src/main/java/com/redis/cluster/listen/LogAListen.java new file mode 100644 index 0000000..a22e14a --- /dev/null +++ b/ware-redis-cluster/src/main/java/com/redis/cluster/listen/LogAListen.java @@ -0,0 +1,39 @@ +package com.redis.cluster.listen; + +import com.redis.cluster.config.RedisLock; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.stereotype.Component; +import javax.annotation.Resource; + +@Component +public class LogAListen implements RedisHandler { + + private static final Logger LOG = LoggerFactory.getLogger(LogAListen.class) ; + @Resource + private RedisLock redisLock; + @Override + public String queueName() { + return "LogA-key"; + } + @Override + public String consume(String msgBody) { + // 加锁，防止消息重复投递 + String lockKey = "lock-order-uuid-A"; + boolean lock = false; + try { + lock = redisLock.addLock(lockKey, 60); + if (!lock) { + return "success"; + } + LOG.info("LogA-key == >>" + msgBody); + } catch (Exception e){ + e.printStackTrace(); + } finally { + if (lock) { + redisLock.removeLock(lockKey); + } + } + return "success"; + } +} diff --git a/ware-redis-cluster/src/main/java/com/redis/cluster/listen/RedisHandler.java b/ware-redis-cluster/src/main/java/com/redis/cluster/listen/RedisHandler.java new file mode 100644 index 0000000..cffcaaf --- /dev/null +++ b/ware-redis-cluster/src/main/java/com/redis/cluster/listen/RedisHandler.java @@ -0,0 +1,13 @@ +package com.redis.cluster.listen; + +public interface RedisHandler { + /** + * 队列名称 + */ + String queueName(); + + /** + * 队列消息内容 + */ + String consume (String msgBody); +} diff --git a/ware-redis-cluster/src/main/java/com/redis/cluster/listen/RedisListener.java b/ware-redis-cluster/src/main/java/com/redis/cluster/listen/RedisListener.java new file mode 100644 index 0000000..5ab8df1 --- /dev/null +++ b/ware-redis-cluster/src/main/java/com/redis/cluster/listen/RedisListener.java @@ -0,0 +1,60 @@ +package com.redis.cluster.listen; + +import com.redis.cluster.utils.SpringUtil; +import org.springframework.beans.factory.InitializingBean; +import redis.clients.jedis.Jedis; +import redis.clients.jedis.JedisSentinelPool; +import javax.annotation.Resource; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.SynchronousQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +public class RedisListener implements InitializingBean { + /** + * Redis 集群 + */ + @Resource + private JedisSentinelPool jedisSentinelPool; + private List handlers = null; + private ExecutorService product = null; + private ExecutorService consumer = null; + /** + * 初始化配置 + */ + @Override + public void afterPropertiesSet() { + handlers = SpringUtil.getBeans(RedisHandler.class) ; + product = new ThreadPoolExecutor(10,15,60 * 3, + TimeUnit.SECONDS,new SynchronousQueue<>()); + consumer = new ThreadPoolExecutor(10,15,60 * 3, + TimeUnit.SECONDS,new SynchronousQueue<>()); + for (RedisHandler redisHandler : handlers){ + product.execute(() -> { + redisTask(redisHandler); + }); + } + } + /** + * 队列监听 + */ + public void redisTask (RedisHandler redisHandler){ + Jedis jedis = null ; + while (true){ + try { + jedis = jedisSentinelPool.getResource() ; + List msgBodyList = jedis.brpop(0, redisHandler.queueName()); + if (msgBodyList != null && msgBodyList.size()>0){ + consumer.execute(() -> { + redisHandler.consume(msgBodyList.get(1)) ; + }); + } + } catch (Exception e){ + e.printStackTrace(); + } finally { + if (jedis != null) jedis.close(); + } + } + } +} diff --git a/ware-redis-cluster/src/main/java/com/redis/cluster/service/RedisService.java b/ware-redis-cluster/src/main/java/com/redis/cluster/service/RedisService.java new file mode 100644 index 0000000..d1f49e8 --- /dev/null +++ b/ware-redis-cluster/src/main/java/com/redis/cluster/service/RedisService.java @@ -0,0 +1,5 @@ +package com.redis.cluster.service; + +public interface RedisService { + void saveQueue (String queueKey ,String msgBody) ; +} diff --git a/ware-redis-cluster/src/main/java/com/redis/cluster/service/impl/RedisServiceImpl.java b/ware-redis-cluster/src/main/java/com/redis/cluster/service/impl/RedisServiceImpl.java new file mode 100644 index 0000000..7ed1453 --- /dev/null +++ b/ware-redis-cluster/src/main/java/com/redis/cluster/service/impl/RedisServiceImpl.java @@ -0,0 +1,27 @@ +package com.redis.cluster.service.impl; + +import com.redis.cluster.service.RedisService; +import org.springframework.stereotype.Service; +import redis.clients.jedis.Jedis; +import redis.clients.jedis.JedisSentinelPool; +import javax.annotation.Resource; + +@Service +public class RedisServiceImpl implements RedisService { + + @Resource + private JedisSentinelPool jedisSentinelPool; + + @Override + public void saveQueue(String queueKey, String msgBody) { + Jedis jedis = null; + try { + jedis = jedisSentinelPool.getResource(); + jedis.lpush(queueKey,msgBody) ; + } catch (Exception e){ + e.printStackTrace(); + } finally { + if (jedis != null) jedis.close(); + } + } +} diff --git a/ware-redis-cluster/src/main/java/com/redis/cluster/utils/SpringUtil.java b/ware-redis-cluster/src/main/java/com/redis/cluster/utils/SpringUtil.java new file mode 100644 index 0000000..c10ba5e --- /dev/null +++ b/ware-redis-cluster/src/main/java/com/redis/cluster/utils/SpringUtil.java @@ -0,0 +1,31 @@ +package com.redis.cluster.utils; + +import org.springframework.beans.BeansException; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +public class SpringUtil implements ApplicationContextAware { + + private static ApplicationContext applicationContext; + + @Override + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + SpringUtil.applicationContext = applicationContext; + } + public static T getBean(Class clazz) { + return applicationContext.getBean(clazz); + } + + public static T getBean(String name, Class clazz) { + return applicationContext.getBean(name, clazz); + } + + public static List getBeans(Class clz) { + Map map = applicationContext.getBeansOfType(clz); + return new ArrayList<>(map.values()); + } +} diff --git a/ware-redis-cluster/src/main/resources/application-dev.yml b/ware-redis-cluster/src/main/resources/application-dev.yml new file mode 100644 index 0000000..6164622 --- /dev/null +++ b/ware-redis-cluster/src/main/resources/application-dev.yml @@ -0,0 +1,14 @@ +spring: + # Redis 集群 + redis: + sentinel: + # sentinel 配置 + master: mymaster + nodes: 192.168.0.127:26379 + maxTotal: 60 + minIdle: 10 + maxWaitMillis: 10000 + testWhileIdle: true + testOnBorrow: true + testOnReturn: false + timeBetweenEvictionRunsMillis: 10000 diff --git a/ware-redis-cluster/src/main/resources/application.yml b/ware-redis-cluster/src/main/resources/application.yml new file mode 100644 index 0000000..b5aa42c --- /dev/null +++ b/ware-redis-cluster/src/main/resources/application.yml @@ -0,0 +1,14 @@ +# Tomcat +server: + tomcat: + uri-encoding: UTF-8 + max-threads: 1000 + min-spare-threads: 30 + port: 7006 + connection-timeout: 5000ms + +spring: + application: + name: ware-redis-cluster + profiles: + active: dev \ No newline at end of file