• 周三. 2月 28th, 2024

5G编程聚合网

5G时代下一个聚合的编程学习网

热门标签

(七)整合 Redis集群 ,实现消息队列场景

[db:作者]

1月 15, 2022

整合 Redis集群 ,实现消息队列场景

  • 1、Redis集群简介
    • 1.1 RedisCluster概念
  • 2、SpringBoot整合Redis集群
    • 2.1 核心依赖
    • 2.2 核心配置
    • 2.3 参数渲染类
    • 2.4 集群配置文件
    • 2.5 配置Redis模板类
  • 3、模拟队列场景案例
    • 3.1 加锁解锁工具
    • 3.2 消息消费
    • 3.3 消息监听器
    • 3.4 消息生产者
    • 3.5 场景测试接口

1、Redis集群简介

1.1 RedisCluster概念

Redis的分布式解决方案,在3.0版本后推出的方案,有效地解决了Redis分布式的需求,当一个服务宕机可以快速的切换到另外一个服务。redis cluster主要是针对海量数据+高并发+高可用的场景。

2、SpringBoot整合Redis集群

2.1 核心依赖

<dependency>
<groupId>org.springframework.bootgroupId>
<artifactId>spring-boot-starter-data-redisartifactId>
<version>${spring-boot.version}version>dependency><dependency>
<groupId>redis.clientsgroupId>
<artifactId>jedisartifactId>
<version>${redis-client.version}version>dependency>
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.

2.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: 10000
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.

2.3 参数渲染类

@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方法}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.

2.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();
}}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.
  • 16.
  • 17.
  • 18.
  • 19.
  • 20.
  • 21.
  • 22.
  • 23.
  • 24.
  • 25.
  • 26.

2.5 配置Redis模板类

@Configurationpublic class RedisConfig {
@Bean
public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory factory) {
StringRedisTemplate stringRedisTemplate = new StringRedisTemplate();
stringRedisTemplate.setConnectionFactory(factory);
return stringRedisTemplate;
}}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.

3、模拟队列场景案例

生产者消费者模式:客户端监听消息队列,消息达到,消费者马上消费,如果消息队列里面没有消息,那么消费者就继续监听。基于Redis的LPUSH(BLPUSH)把消息入队,用 RPOP(BRPOP)获取消息的模式。

3.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();
/*
* 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.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.
  • 16.
  • 17.
  • 18.
  • 19.
  • 20.
  • 21.
  • 22.
  • 23.
  • 24.
  • 25.
  • 26.
  • 27.
  • 28.
  • 29.
  • 30.

3.2 消息消费

1)封装接口

public interface RedisHandler {
/**
* 队列名称
*/
String queueName();
/**
* 队列消息内容
*/
String consume (String msgBody);}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.

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";
}}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.
  • 16.
  • 17.
  • 18.
  • 19.
  • 20.
  • 21.
  • 22.
  • 23.
  • 24.
  • 25.
  • 26.
  • 27.
  • 28.

3.3 消息监听器

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();
}
}
}}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.
  • 16.
  • 17.
  • 18.
  • 19.
  • 20.
  • 21.
  • 22.
  • 23.
  • 24.
  • 25.
  • 26.
  • 27.
  • 28.
  • 29.
  • 30.
  • 31.
  • 32.
  • 33.
  • 34.
  • 35.
  • 36.
  • 37.
  • 38.
  • 39.
  • 40.
  • 41.
  • 42.
  • 43.
  • 44.
  • 45.
  • 46.

3.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();
}
}}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.

3.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" ;
}}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.

               

《(七)整合 Redis集群 ,实现消息队列场景》有17个想法
  1. Haѵe you evеr considered writing an ebook or guest autһoring on other sites?

    I have a blog cеntered on the same ideas you discuss and would love to have үou shɑre some stories/informatіon. I
    know my audience would appreciate your work.
    If you’re even remotely interested, feel free to
    send me an e maіl.

    My web ρage – oferta interneti ne fier

  2. Thаnks for ones marvelous posting! I гeally enjoyed
    reading іt, you could be a great author.Ӏ wіll mаke ϲertain t᧐
    bookmark your blog аnd may cоmе back from now on. I want to encourage ʏoᥙ
    to continue yoսr great job, have а nice afternoon!
    Shit dhe bli Reklama dhe Njoftime falas
    Shqiperi kosove Maqedoni
    shitdhebli.ϲom

  3. Зайдя на sladkiye-strasti.ru, я был впечатлён разнообразием и качеством товаров, которые предлагает этот сайт. Особое внимание я уделил [url=https://sladkiye-strasti.ru/]женские штучки 18[/url] , который представлен в широком ассортименте. Покупка на сайте оказалась не только выгодной, но и очень удобной благодаря быстрой доставке и анонимной упаковке товаров.

发表回复

您的电子邮箱地址不会被公开。 必填项已用*标注