十年网站开发经验 + 多家企业客户 + 靠谱的建站团队
量身定制 + 运营维护+专业推广+无忧售后,网站问题一站解决
这篇文章主要介绍了SpringBoot2如何整合redis哨兵集群 实现消息队列场景,具有一定借鉴价值,感兴趣的朋友可以参考下,希望大家阅读完这篇文章之后大有收获,下面让小编带着大家一起了解一下。
站在用户的角度思考问题,与客户深入沟通,找到剑川网站设计与剑川网站推广的解决方案,凭借多年的经验,让设计与互联网技术结合,创造个性化、用户体验好的作品,建站类型包括:成都网站设计、网站制作、外贸营销网站建设、企业官网、英文网站、手机端网站、网站推广、域名注册、网页空间、企业邮箱。业务覆盖剑川地区。
Redis的分布式解决方案,在3.0版本后推出的方案,有效地解决了Redis分布式的需求,当一个服务宕机可以快速的切换到另外一个服务。redis cluster主要是针对海量数据+高并发+高可用的场景。
org.springframework.boot spring-boot-starter-data-redis ${spring-boot.version} redis.clients jedis ${redis-client.version}
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
@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方法 }
@Configuration @EnableConfigurationProperties(RedisParam.class) public class RedisPool { @Resource private RedisParam redisParam ; @Bean("jedisSentinelPool") public JedisSentinelPool getRedisPool (){ Setsentinels = 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(); } }
@Configuration public class RedisConfig { @Bean public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory factory) { StringRedisTemplate stringRedisTemplate = new StringRedisTemplate(); stringRedisTemplate.setConnectionFactory(factory); return stringRedisTemplate; } }
生产者消费者模式:客户端监听消息队列,消息达到,消费者马上消费,如果消息队列里面没有消息,那么消费者就继续监听。基于Redis的LPUSH(BLPUSH)把消息入队,用 RPOP(BRPOP)获取消息的模式。
@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(); } } }
1)封装接口
public interface RedisHandler { /** * 队列名称 */ String queueName(); /** * 队列消息内容 */ String consume (String msgBody); }
2)接口实现
@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"; } }
public class RedisListener implements InitializingBean { /** * Redis 集群 */ @Resource private JedisSentinelPool jedisSentinelPool; private Listhandlers = 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(); } } } }
@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(); } } }
@RestController public 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哨兵集群 实现消息队列场景”这篇文章对大家有帮助,同时也希望大家多多支持创新互联,关注创新互联行业资讯频道,更多相关知识等着你来学习!