Redis发布订阅

目录
一 、redis发布订阅模式：
二、通道与模式两者订阅模式原理实现：
三 、redis发布订阅PubSub与Stream模式区分
1）Pub-Sub
2）Stream消息队列
四、Stream与其他中间件的区别
1.数据结构
2.持久化
3.消息确认
4.重试机制
5.业务利用场景
五、SubPub与stream实际应用中的利用场景
1）利用发布/订阅（pub_sub）：
2）利用流（stream）：
六、Pub_Sub代码实现
1）引入依靠
 2）RedisConfig设置
 3）发布订阅通过channel标识对应业务场景
 4）生产者消息发送
 5）消费者订阅生产者对应的发送消息通道
七、stream流代码实现
1） 依靠（redis5.0以上版本以及spring data redis版本为2.2.0版本以上）*
2）定义一个监听
3）定义抽象类AbstractMsgService
4）创建生产者
5）创建消费者
6）创建容器
八、利用消息队列会遇到什么样的题目以及有哪办理措施
1.对于生产者/来说(网络抖动，服务器宕机)
1.1 网络抖动导致发送失败
1.2 由于服务器宕机了导致发送失败：
2.对于消费者来说(网络抖动，重复消费)
2.1 网络抖动导致确认失败
2.2 重复消费
3.对于消息队列来说（重复消费、消息积存）
3.1 消息积存
3.2 消息顺序性
3.3 消息耽误
九、利用redis有哪些实用场景
一 、redis发布订阅模式：

      

• 队列：基于 list 结构的消息队列，是一种 Publisher 与 Consumer 点对点的强关联关系
  
• 通道channel：可实现消费者订阅多个通道
  
• 模式Pattern：与通道的区别是 1、模式无需完整的通道名称订阅，用模糊订阅（通过？,*来代替）
  

   二、通道与模式两者订阅模式原理实现：

   

通道：内部实现是通过（PubSub_Channel）键对值，key（数组元素）为channel，value就是某个client

   

   

Pattern：内部实现是通过（PubSub_patterns)链表，程序创建包含客户端信息和被订阅模式的 pubsubPattern 结构，并将该结构添加到 redisServer.pubsub_patterns 链表中

   

​

   三 、redis发布订阅PubSub与Stream模式区分

   1）Pub-Sub

   

在于PubSub 的生产者传递过来一个消息，Redis会直接找到相应的消费者传递已往。如果一个消费者都没有，那么消息直接丢弃。如果开始有三个消费者，一个消费者忽然挂掉了，生产者会继续发送消息，另外两个消费者可以持续收到消息。但是挂掉的消费者重新连上的时候，这断连期间生产者发送的消息，对于这个消费者来说就是彻底丢失了

   

​

   2）Stream消息队列

   

能提供持久性、顺序的队列功能

   

实现顺序队列实现的内部结构

   

​消息ID：是按照毫秒数-序列号

   

消息内容：平常的键对值

   

消费者组下有last_deliveres_id记载着消费到那个id下的消息，

   

pendin_id表示还没有确认的消息的id，基于这个实现消息重试机制 s

   

   四、Stream与其他中间件的区别

   1.数据结构

   

Stream 是一个有序的消息日志，支持以时间顺序追加消息。来保证了数据的顺序性，每条消息都有一个唯一的 ID（包括时间戳和序列号），并且可以附加多个字段及其值。

   

RocketMQ 利用主题（Topic）和队列（Queue）模子，消息通过主题举行发布，消费者从队列中拉取消息。

   

Kafka 利用主题（Topic）和分区（Partion）模子，消息通过主题举行发布，消费者从队列中拉取消息。

   2.持久化

   

Stream数据持久化由 Redis 的内存存储消息 并通过持久化机制（如 RDB 和 AOF）举行实现持久化

   

rocketmq持久化是通过commit log举行存储消息体文件 后通过文件体系举行实现持久化

   

kafka数据持久化是通过日志文件举行实现持久化机制

   3.消息确认

   

Stream支持消息确认机制，消费者可以确认消息是否处理乐成

   

rocketmq支持多种消息确认机制，包括同步确认和异步确认。

   

kafka支持多种消息确认机制，包括同步确认和异步确认。

   4.重试机制

   

Redis Streams自己不提供内建的重试机制，但可以通过客户端逻辑实现。你需要在消费者那里捕获处理失败的消息

   

rocketmq可以实现主动重试16次 还失败就会进入死信队列内里

   

Kafka 自己不提供内建的重试机制，但可以通过客户端逻辑实现。你需要在消费者那里捕获处理失败的消息

   5.业务利用场景

   

Stream用于需要高性能、低耽误的，能够快读响应和高并发能力场景，如及时数据处理和日志体系、及时消息处理等等

   

rocketmq用户高级消息传递特性，如事件消息、定时消息、顺序消息，适用于需要高吞吐量、可靠性和分布式特性的场景，如大规模数据传输，分布式体系的事件驱动架构等。

   

kafka是一个分布式流处理平台，提供了高吞吐量的体系 实现了日志收集与分析以及是大数据集成

   五、SubPub与stream实际应用中的利用场景

   1）利用发布/订阅（pub_sub）：

   

及时谈天应用程序：用于用户之间的及时消息传递。

   

及时新闻推送：用于向订阅者发送最新新闻。

   

事件处理体系：用于触发和处理事件。

   2）利用流（stream）：

   

任务队列：用于分发和处理任务，如后台任务处理。

   

数据同步：用于数据更新和同步，确保多个体系之间的划一性。

   

时间序列数据库：用于存储和查询时间序列数据，如监控和日志记载。

   

   

选择发布/订阅或流取决于你的具体需求和应用场景。

   六、Pub_Sub代码实现

   1）引入依靠

  

1. <strong><dependency>
2.     <groupId>org.springframework.boot</groupId>
3.     <artifactId>spring-boot-starter-data-redis</artifactId>
4. </dependency></strong>

复制代码

   2）RedisConfig设置

  

1. <strong>@Configuration
2. /*@EnableCaching*/
3. public class RedisConfig {
5.     @Autowired
6.     private RedisConnectionFactory redisConnectionFactory;
8.     @Autowired
9.     private RedisMessageListener redisMessageListener;
11.     /*配置Redis消息监听器容器：在Java Spring的配置类中配置Redis消息监听器容器，用于注册并管理消息监听器。*/
12.     @Bean
13.     public RedisMessageListenerContainer redisMessageListenerContainer() {
14.         RedisMessageListenerContainer container = new RedisMessageListenerContainer();
15.         container.setConnectionFactory(redisConnectionFactory);
16.         Arrays.stream(RedisMessageChannelEnum.values()).forEach(item ->container.addMessageListener(redisMessageListener, new ChannelTopic(item.getCode())));
17.         return container;
18.     }
19.     }</strong>

复制代码

   3）发布订阅通过channel标识对应业务场景

  

1. <strong>public enum RedisMessageChannelEnum {
2. ...
3.         }</strong>

复制代码

   4）生产者消息发送

  

1. <strong>@Log4j2
2. @Component
3. public class MessagePublisher {
4.     @Autowired
5.     private RedisTemplate<Object, Object> redisTemplate;
7.     /**
8.      * 服务端发布消息
9.      *
10.      * @param channel 通道名
11.      * @param message 待发送的消息
12.      */
13.     public void sendMessage(String channel, String message) {
14.         redisTemplate.convertAndSend(channel, message);
15.         log.info("消息发送成功...  channel={}, message={}", channel, message);
16.         redisTemplate.opsForList().leftPush("topic1",message);
17.     }
18. }</strong>

复制代码

   5）消费者订阅生产者对应的发送消息通道

  

1. <strong>@Component
2. @Log4j2
3. public class RedisMessageListener implements MessageListener {
5.     @Autowired
6.     private RetrySendMessageServiceRouteAdaptor retrySendMessageServiceRouteAdaptor;
7.     @Autowired
8.     private RedisTemplate redisTemplate;
10.     @Autowired
11.     private ObjectMapper objectMapper;
12.     @Override
13.     public void onMessage(Message message, byte[] bytes) {
15.         // 获取消息
16.         byte[] messageBody = message.getBody();
18.         // 使用值序列化器转换
20.         Object msg = redisTemplate.getValueSerializer().deserialize(messageBody);
22.         // 获取监听的频道
24.         byte[] channelByte = message.getChannel();
26.         // 使用字符串序列化器转换
28.         Object channel = redisTemplate.getStringSerializer().deserialize(channelByte);
30.         log.info("---频道---: " + channel);
32.         log.info("---消息内容---: " + msg);
34.         String channelString = objectMapper.convertValue(channel, String.class);
35.                RedisMessageChannelEnum redisMessageChannelEnum =  RedisMessageChannelEnum.getEnumByCode(channelString);
37.           // 通过对应的消息通道匹配到对应的是适配器服务执行业务
38.         retrySendMessageServiceRouteAdaptor.<RetrySendMessageService>find(ServiceModeEnum.WX_SYSTEM, redisMessageChannelEnum).retrySendMassage(retrySendMassageDto);
39.         }</strong>

复制代码

  七、stream流代码实现

   1） 依靠（redis5.0以上版本以及spring data redis版本为2.2.0版本以上）*

  

1. <strong> <dependency>
2.             <groupId>org.springframework.data</groupId>
3.             <artifactId>spring-data-redis</artifactId>
4.             <version>2.2.0.RELEASE</version>
5.         </dependency></strong>

复制代码

  2）定义一个监听

  

1. <strong>@Target(ElementType.METHOD)
2. @Retention(value = RetentionPolicy.RUNTIME)
3. public @interface MsgStreamListener
4. {
5.     String stream();
7.     String group();
9.     String name();
10. }</strong>

复制代码

  3）定义抽象类AbstractMsgService

  

1. <strong>public abstract class AbstractMsgService {
2. }</strong>

复制代码

  4）创建生产者

  

1. <strong>@Service
2. @RequiredArgsConstructor
3. public class RedisMessageProducer
4. {
5.     private final RedisTemplate<String,String> redisTemplate;
7.     public void sendMsg(String streamKey,String msgKey,String msg){
8.         Map<String,String> msgMap = new HashMap<>();
10.         msgMap.put(msgKey,msg);
12.         RecordId recordId = redisTemplate.opsForStream().add(streamKey,msgMap);
14.         if(recordId == null){
15.             throw new RuntimeException("发送消息失败");
16.         }
17.     }
19. }</strong>

复制代码

  5）创建消费者

  

1. <strong>@Service
2. @Slf4j
3. public class MessageHandlerService extends AbstractMsgService
4. {
6.     @MsgStreamListener(group = "test1",name = "test1",stream = "test1")
7.     public void onMessage(ObjectRecord<String, String> message)
8.     {
9.         var stream = message.getStream();
10.         var msgId = message.getId().toString();
11.         var msgBody = message.getValue();
13.         log.info("receive test1 msg stream:{} msgId:{} msgBody:{}",stream,msgId,msgBody);
14.     }
15. }
17. </strong>

复制代码

  6）创建容器

  

1. <strong>@Configuration
2. @Slf4j
3. public class RedisStreamConfig
4. {
5.     private final RedisTemplate<String,String> redisTemplate;
7.     private final ApplicationContext applicationContext;
9.     public RedisStreamConfig(RedisTemplate<String, String> redisTemplate, ApplicationContext applicationContext) {
10.         this.redisTemplate = redisTemplate;
11.         this.applicationContext = applicationContext;
12.     }
14.     @Bean
15.     public StreamMessageListenerContainer<String, ObjectRecord<String,String>> streamMessageListenerContainer(RedisConnectionFactory connectionFactory)
16.     {
17.         StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, ObjectRecord<String,String>> options =
18.                 StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()
19.                         .pollTimeout(java.time.Duration.ofSeconds(1))
20.                         .targetType(String.class)
21.                         .build();
23.         //创建监听redis流的消息监听容器
24.         StreamMessageListenerContainer<String, ObjectRecord<String,String>> listenerContainer =
25.                 StreamMessageListenerContainer.create(connectionFactory, options);
27.         //找到所有继承AbstractMsgService的类
28.         var serviceArray = applicationContext.getBeansOfType(AbstractMsgService.class).values().toArray();
30.         for (var service : serviceArray) {
32.             for (Method method : service.getClass().getMethods()) {
34.                 if(method.isAnnotationPresent(MsgStreamListener.class)){
36.                     MsgStreamListener annotation = method.getAnnotation(MsgStreamListener.class);
38.                     String stream = annotation.stream();
39.                     String group = annotation.group();
40.                     String name = annotation.name();
42.                     StreamListener<String,ObjectRecord<String,String>> listener = (StreamListener<String, ObjectRecord<String,String>>) message -> {
43.                         try {
44.                             method.invoke(service,message);
45.                         }catch (Exception e){
46.                             log.warn(e.getMessage());
47.                         }
48.                     };
50.                     //创建redis流的消息监听器
51.                     listenerContainer.receive(Consumer.from(group,name),
52.                             StreamOffset.create(stream, ReadOffset.lastConsumed()),
53.                             listener);
55.                     initializeStream(stream,name);
56.                 }
58.             }
59.         }
61.         listenerContainer.start();
63.         return listenerContainer;
64.     }
66.     public void initializeStream(String stream,String group) {
68.         StreamOperations<String, Object, Object> streamOperations = redisTemplate.opsForStream();
70.         // 创建一个流
71.         try {
72.             streamOperations.createGroup(stream, ReadOffset.from("0"), group);
73.         } catch (Exception e) {
74.             // 流可能已存在，忽略异常
75.         }
76.     }
78. }
80. </strong>

复制代码

  八、利用消息队列会遇到什么样的题目以及有哪办理措施

   1.对于生产者/来说(网络抖动，服务器宕机)

   1.1 网络抖动导致发送失败

   

redis中stream没有重试机制 但是你可以通过捕获异常记载日志再举行处理

   

kafka、rocketmq会主动重试机制

   1.2 由于服务器宕机了导致发送失败：

   

可以通过集群的方式举行实现

   2.对于消费者来说(网络抖动，重复消费)

   2.1 网络抖动导致确认失败

   

可以通过手动ack举行确认

   2.2 重复消费

   

可以通过幂等性来办理

   3.对于消息队列来说（重复消费、消息积存）

   3.1 消息积存

   

看下是否是因为消费者过少 可以适当增长消费者

   

是否能优化消费者处理逻辑

   

进步服务器性能（增长硬件资源、优化设置或升级版本‌）

   

是否可扩大消息队列长度

   3.2 消息顺序性

   

rocketmq：利用同一个MQ队列，并且针对单个队列只开启一个消费者消费

   

kafka：发送同一个partiion分区，并且针对单个队列只开启一个消费者消费

   3.3 消息耽误

   

3.3.1 由于网络耽误：确保保消息队列服务器与消息生产者和消费者之间的网络毗连稳定，淘汰网络耽误的影响

   

3.3.2 服务器负载: 增长消息队列服务器的硬件资源，例如增长CPU、内存和磁盘容量等，以进步消息处理的速度

   

3.3.3 消息处理时间过长:

   

• 消息消费者的并行处理：将消息的处理逻辑举行拆分和并行化，可以通过增长消费者的数量或者利用多线程处理的方式来进步消息的处理速度
  
• 异步处理：对于一些耗时操作，可以利用异步处理的方式，将消息放入消息队列之后立刻返回，然后由后台异步处理完成，从而淘汰消息发送和接收的时间
  
• 检查业务逻辑：检查消息的处理逻辑是否存在瓶颈或者低效的操作，有可能通过重新设计业务逻辑来进步消息的处理速度。
  

   九、利用redis有哪些实用场景

   

• 查询缓存
  
• 分布式锁
  
• 短信验证码登录 或者举行缓存登任命户信息：
  
• 实现登录控制功能，要求用户输入电话号，获取验证码，后台天生验证码并将验证码存储在Redis中，有效期5分钟。当用户在5分钟内输入准确的验证码，验证通过。
  
• 获取附近地理信息: Redis提供了Geospatial数据范例，如有序聚集（Sorted Set）来存储地理位置数据。
  
• zset有序聚集：排行榜
  
• 获取分类树：
  
• incr的命令的原子性的自增属性各类计数器、统计网站访问量、点赞数等
  
• 限流：执行lua脚本，基于redis中incrby举行一个累加，通过传参是哪个接口，限流次数 失效时间，通过切面的前置通知去执行我们的脚本
  

   

   

   

   

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。