您好,登录后才能下订单哦!
密码登录
登录注册
点击 登录注册 即表示同意《亿速云用户服务条款》
这篇文章主要介绍了SpringBoot2如何整合Redis哨兵集群 实现消息队列场景,具有一定借鉴价值,感兴趣的朋友可以参考下,希望大家阅读完这篇文章之后大有收获,下面让小编带着大家一起了解一下。
Redis的分布式解决方案,在3.0版本后推出的方案,有效地解决了Redis分布式的需求,当一个服务宕机可以快速的切换到另外一个服务。redis cluster主要是针对海量数据+高并发+高可用的场景。
<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>
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 (){ 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(); } }
@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 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(); } } } }
@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哨兵集群 实现消息队列场景”这篇文章对大家有帮助,同时也希望大家多多支持亿速云,关注亿速云行业资讯频道,更多相关知识等着你来学习!
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。