这篇文章主要介绍了SpringBoot2如何整合Redis哨兵集群 实现消息队列场景，具有一定借鉴价值，感兴趣的朋友可以参考下，希望大家阅读完这篇文章之后大有收获，下面让小编带着大家一起了解一下。
一、Redis集群简介
1、RedisCluster概念
Redis的分布式解决方案，在3.0版本后推出的方案，有效地解决了Redis分布式的需求，当一个服务宕机可以快速的切换到另外一个服务。redis cluster主要是针对海量数据+高并发+高可用的场景。
二、与SpringBoot2.0整合
1、核心依赖
<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-redis</artifactId> <version>${spring-boot.version}</version></dependency><dependency> <groupId>redis.clients</groupId> <artifactId>jedis</artifactId> <version>${redis-client.version}</version></dependency>2、核心配置
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: 100003、参数渲染类
@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 ; // 省略GET和SET方法}4、集群配置文件
@Configuration@EnableConfigurationProperties(RedisParam.class)public class RedisPool { @Resource private RedisParam redisParam ; @Bean("jedisSentinelPool") public JedisSentinelPool getRedisPool (){ Set<String> 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(); }}5、配置Redis模板类
@Configurationpublic class RedisConfig { @Bean public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory factory) { StringRedisTemplate stringRedisTemplate = new StringRedisTemplate(); stringRedisTemplate.setConnectionFactory(factory); return stringRedisTemplate; }}三、模拟队列场景案例
生产者消费者模式：客户端监听消息队列，消息达到，消费者马上消费，如果消息队列里面没有消息，那么消费者就继续监听。基于Redis的LPUSH（BLPUSH）把消息入队，用 RPOP（BRPOP）获取消息的模式。
1、加锁解锁工具
@Componentpublic 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(); 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(); } }}2、消息消费
1）封装接口
public interface RedisHandler { String queueName(); String consume (String msgBody);}2）接口实现
@Componentpublic 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"; }}3、消息监听器
public class RedisListener implements InitializingBean { @Resource private JedisSentinelPool jedisSentinelPool; private List<RedisHandler> 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<String> 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(); } } }}4、消息生产者
@Servicepublic 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(); } }}5、场景测试接口
@RestControllerpublic class RedisController { @Resource private RedisService redisService ; @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" ; }}感谢你能够认真阅读完这篇文章，希望小编分享的“SpringBoot2如何整合Redis哨兵集群 实现消息队列场景”这篇文章对大家有帮助，同时也希望大家多多支持编程网，关注编程网行业资讯频道，更多相关知识等着你来学习!
