当前位置：首页 > news >正文

【redis】springboot 用redis stream实现MQ消息队列考虑异常ack重试场景

news 来源：原创 2024/9/18 5:21:01

redis stream是redis5引入的特性，一定程度上借鉴了kafka等MQ的设计，部署的redis版本必须 >= 5

本文主要讲的是思路，结合简单的源码分析（放心，无需深入大量源码）；讲述在redis stream文档缺乏，网上资料欠缺，gpt回答不上来的情况下，博主是如何用两三天的时间从没接触过redis stream 到分析完成了redis stream mq功能。博主始终认为有明确的思路才能知道什么代码是正确的能复制拿来用，什么代码只是单纯跑起来demo的绝对达不到生产级别。
本文源自csdn博主：孟秋与你，博主虽才疏学浅却也是在资料极少的情况下，辛苦研究源码、整理思路撰写的本文，转载请声明出处。

文章目录

redisTemplate API的熟悉
配置
- redis mq config
- 监听器：
- 定时器
优化方向

(本文基于springboot3.3 jdk17 redis6环境,
理论上springboot2 redis5也是通用教程可能会有细微的api差异稍微分析一下源码方法都能处理)

redisTemplate API的熟悉

我们在操作redis的时候通常是使用spring-data-redis提供的redisTemplate或者jedis 本文以redisTemplate为例。
（实际业务场景可能需要考虑用jedis替换因为mq通常在数据量、并发量都大的场景；redisTemplate的优势在于和springboot的完美集成，且不需要考虑通过连接池来管理线程安全问题）

用过redisTemplate的同学应该都会自己封装一下工具类，因为redisTemplate封装的不够好，不管怎么样我们都需要先看看这个类

redisTemplate.opsForHash()，redisTemplate.opsForValue()

各位应该很熟悉了, stream是一种新引入的格式，那么我们直接在RedisTemplate类里面搜stream就好了,正常都会有对应API
（没对应API那就是spring版本太老了 spring那个老版本出来的时候 redis还没出到5 ）

搜到了opsForStream()方法在这里插入图片描述继续查看方法如下图：

这里说明一下，redis的streamKey就类似mq的topic, group是消费者组,cousumer是消费者，acknowledge即ack 应答机制告诉mq已经成功消费了，claim是强制将消息转至其它消费者通常用于消费失败/多次消费失败的场景，pending存放的是未ack的消息就比如消费某个消息时出现了异常没能执行到ack 这些消息就会放在pending list 确保消息不丢失。

通过api，加上我们掌握的mq基本知识，大概就能理解是怎么一回事了。demo搭建不难，但是代码要上生产，我们就必须考虑消息消费失败了怎么办该如何重试，也就是说重点的api在acknowledge和pending上面。

一个简单的封装

	@Componentpublic class RedisStreamUtil {@Autowiredprivate RedisTemplate<String, Object> redisTemplate;/*** 创建消费组** @param key   键名称* @param group 组名称* @return {@link String}*/public String createGroup(String key, String group) {return redisTemplate.opsForStream().createGroup(key, group);}/*** 获取消费者信息** @param key   键名称* @param group 组名称* @return {@link StreamInfo.XInfoConsumers}*/public StreamInfo.XInfoConsumers queryConsumers(String key, String group) {return redisTemplate.opsForStream().consumers(key, group);}/*** 查询组信息** @param key 键名称* @return*/public StreamInfo.XInfoGroups queryGroups(String key) {return redisTemplate.opsForStream().groups(key);}/*** 添加Map消息* @param key* @param value*/public String addMap(String key, Map<String, Object> value) {return redisTemplate.opsForStream().add(key, value).getValue();}/*** 读取消息* @param key*/public List<MapRecord<String, Object, Object>> read(String key) {return redisTemplate.opsForStream().read(StreamOffset.fromStart(key));}/*** 确认消费* @param key* @param group* @param recordIds*/public Long ack(String key, String group, String... recordIds) {return redisTemplate.opsForStream().acknowledge(key, group, recordIds);}/*** 删除消息* 当一个节点的所有消息都被删除，那么该节点会自动销毁* @param key* @param recordIds*/public Long del(String key, String... recordIds) {return redisTemplate.opsForStream().delete(key, recordIds);}/***  判断是否存在key* @param key*/public boolean hasKey(String key) {Boolean flag= redisTemplate.hasKey(key);return flag != null && flag;}}

注意：会有循环依赖的问题，如果没有那就是springboot版本太低，低版本默认是开启允许循环依赖的，高版本默认不允许(2.7已经不允许了具体版本不记得了)

解决方法1：在yml配置里面允许循环依赖

server:port: 8586spring:application:name: springboot3-demodata:redis:port: 6579host: 192.168.1.1password: xxxxxxxdatabase: 1lettuce:pool:max-wait: 5000msmax-active: 1000datasource:driver-class-name: com.mysql.cj.jdbc.Driverurl: jdbc:mysql://localhost:3306/test?characterEncoding=utf8&serverTimezone=UTC&rewriteBatchedStatements=truetype: com.alibaba.druid.pool.DruidDataSourceusername: rootpassword: root
# 允许循环依赖main:allow-circular-references: true

解决方法2：该工具类不交给spring托管代码如下图所示
在spring bean初始化的时候把redisTemplate bean赋值到工具类即可，工具类方法变成静态方法
在这里插入图片描述

配置

redis mq config

以下代码展示了如何配置多个生产者，也是这个代码最难写。


package com.qiuhuanhen.springboot3demo.redis.config;import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.StrUtil;
import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.qiuhuanhen.springboot3demo.redis.RedisStreamUtil;
import com.qiuhuanhen.springboot3demo.redis.consumer.RedisConsumer;
import com.qiuhuanhen.springboot3demo.redis.consumer.listener.RedisConsumersListener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.RedisServerCommands;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.Subscription;import javax.annotation.Resource;
import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ThreadPoolExecutor;@Configuration
@Slf4j
public class RedisConfig{@Autowiredprivate RedisStreamUtil redisStreamUtil;@Autowiredprivate ThreadPoolExecutor threadPoolExecutor;@Autowiredprivate Map<String, RedisConsumer> redisConsumer;/*** redis序列化** @param redisConnectionFactory* @return {@code RedisTemplate<String, Object>}*/@Beanpublic RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {RedisTemplate<String, Object> template = new RedisTemplate<>();template.setConnectionFactory(redisConnectionFactory);ObjectMapper om = new ObjectMapper();om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);om.activateDefaultTyping(om.getPolymorphicTypeValidator(), ObjectMapper.DefaultTyping.NON_FINAL);Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(om,Object.class);StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();template.setKeySerializer(stringRedisSerializer);template.setHashKeySerializer(stringRedisSerializer);template.setValueSerializer(jackson2JsonRedisSerializer);template.setHashValueSerializer(jackson2JsonRedisSerializer);template.afterPropertiesSet();return template;}@Beanpublic List<Subscription> subscriptions(RedisConnectionFactory factory) {List<Subscription> subscriptions = new ArrayList<>();subscriptions.add( createSubscription(factory, "orderStream", "orderGroup", "orderConsumer"));subscriptions.add( createSubscription(factory, "productStream", "productGroup", "productConsumer"));return subscriptions;}/*** @param factory* @param streamName   类似 topic* @param groupName    消费组是 Redis Streams 中的一个重要特性，它允许多个消费者协作消费同一个流中的消息。每个消费组可以有多个消费者。* @param consumerName 这是消费组中的具体消费者名称。每个消费者会从消费组中领取消息进行处理。* @return*/private Subscription createSubscription(RedisConnectionFactory factory, String streamName, String groupName, String consumerName) {initStream(streamName, groupName);StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options =StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()// 每次从Redis Stream中读取消息的最大条数 (32为rocketmq的pullBatchSize默认数量).batchSize(32).executor(threadPoolExecutor)// 轮询拉取消息的时间 (如果流中没有消息，它会等待这么久的时间，然后再次检查。).pollTimeout(Duration.ofSeconds(1)).errorHandler(throwable -> {log.error("[redis MQ handler exception]", throwable);throwable.printStackTrace();}).build();var listenerContainer = StreamMessageListenerContainer.create(factory, options);// 手动ask消息
//        Subscription subscription = listenerContainer.receive(Consumer.from(groupName, consumerName),
//                //  创建一个流的偏移量实例。 含义: 指定从哪个偏移量开始读取消息。ReadOffset.lastConsumed()表示从上次消费的位置开始。
//                StreamOffset.create(streamName, ReadOffset.lastConsumed()), redisConsumersListener);// 自动ask消息
//            Subscription subscription = listenerContainer.receiveAutoAck(Consumer.from(groupName, consumerName),
//                    StreamOffset.create(streamName, ReadOffset.lastConsumed()), redisConsumersListener);// 手动创建 核心在于 cancelOnError(t -> false)  出现异常不退出StreamMessageListenerContainer.ConsumerStreamReadRequest<String> build = StreamMessageListenerContainer.StreamReadRequest.builder(StreamOffset.create(streamName, ReadOffset.lastConsumed())).consumer(Consumer.from(groupName, consumerName)).autoAcknowledge(false)// 重要！.cancelOnError(t -> false).build();Subscription subscription = listenerContainer.register(build, new RedisConsumersListener(redisStreamUtil));listenerContainer.start();return subscription;}/*** 初始化流 保证stream流程是正常的** @param key* @param group*/private void initStream(String key, String group) {boolean hasKey = redisStreamUtil.hasKey(key);if (!hasKey) {Map<String, Object> map = new HashMap<>(1);map.put("field", "value");//创建主题String result = redisStreamUtil.addMap(key, map);//创建消费组redisStreamUtil.createGroup(key, group);//将初始化的值删除掉redisStreamUtil.del(key, result);log.info("stream:{}-group:{} initialize success", key, group);}}/*** 校验 Redis 版本号，是否满足最低的版本号要求！*/private static void checkRedisVersion(RedisTemplate<String, ?> redisTemplate) {// 获得 Redis 版本Properties info = redisTemplate.execute((RedisCallback<Properties>) RedisServerCommands::info);String version = MapUtil.getStr(info, "redis_version");// 校验最低版本必须大于等于 5.0.0int majorVersion = Integer.parseInt(StrUtil.subBefore(version, '.', false));if (majorVersion < 5) {throw new IllegalStateException(StrUtil.format("您当前的 Redis 版本为 {}，小于最低要求的 5.0.0 版本！", version));}}
}

监听器：

（核心是实现StreamListener接口）

@Slf4j
public class RedisConsumersListener implements StreamListener<String, MapRecord<String, String, String>> {private RedisStreamUtil redisStreamUtil;public RedisConsumersListener(RedisStreamUtil redisStreamUtil) {this.redisStreamUtil = redisStreamUtil;}/*** 监听器** @param message*/@Overridepublic void onMessage(MapRecord<String, String, String> message) {// stream的key值String streamName = message.getStream();//消息IDRecordId recordId = message.getId();//消息内容Map<String, String> msg = message.getValue();// do something 处理 (这里一般是通过设计模式获取实现类方法 统一处理)//逻辑处理完成后，ack消息，删除消息，group为消费组名称StreamInfo.XInfoGroups xInfoGroups = redisStreamUtil.queryGroups(streamName);xInfoGroups.forEach(xInfoGroup -> redisStreamUtil.ack(streamName, xInfoGroup.groupName(), recordId.getValue()));redisStreamUtil.del(streamName, recordId.getValue());}log.info("【streamName】= " + streamName + ",【recordId】= " + recordId + ",【msg】=" + msg);}
}

感兴趣可以看博主踩到的坑, 看完思路才能自行判断代码是否能直接复制使用（个人感觉这才是分析技术最精彩的地方有正确的思路才能在使用新技术时披荆斩棘）; 不感兴趣可以直接跳到下一目录

===== ====== ====== 踩坑start ===== ==== ===== =====
一开始使用的是receive方法（被注释的部分）

        // 手动ask消息
//        Subscription subscription = listenerContainer.receive(Consumer.from(groupName, consumerName),
//                //  创建一个流的偏移量实例。 含义: 指定从哪个偏移量开始读取消息。ReadOffset.lastConsumed()表示从上次消费的位置开始。
//                StreamOffset.create(streamName, ReadOffset.lastConsumed()), redisConsumersListener);

这也是网上使用最多的方法，通过方法名我们可以判断出 receiveAutoAck是会自动ack的，不出异常还好，那如果出现异常呢如何ack？所以我们肯定是要手动控制的。
在这里插入图片描述
我们可以看看源码它们的差异：

是的，就是一个是否自动ack的差别。

既然引入了消息队列，那说明数据量是比较大的，所以肯定是需要考虑异常情况下消息不能丢失的，于是博主在消费时，故意编写了异常模拟不触发ack的场景. 结果发现一旦消费出现异常没有ack时，pending list不再新增数据，在项目重启后数据又增加了，但是再次消息异常时 pending list又阻塞了，这种现象非常奇怪！难道一个消息没ack redis stream就阻塞吗？这显然不符合设计。反复思考后，看起来像是出现异常后就停止了轮询，这个mq就像极了是一次性的。
但是和轮询相关的也就一个pollTimeout参数，它能掀起多大的火花呢？

于是继续看代码配置redis mq时，都有哪些api. 使用receive方法后返回的是一个Subscription ，Subscription类有isActive()方法，于是在定时器中打印subsciption.isActive() 发现它竟然为false

于是我们追踪这个方法：
在这里插入图片描述
追踪到了StreamPollTask类

如果是task类那么应该会有run方法，我们直接在里面搜run()

在这里插入图片描述
run方法里面主要就这两个方法
this.pollState.running();
this.doLoop();
第一个running方法一眼看到头，没什么东西；我们看doLoop() 这个方法看起来是循环执行，如果任务中断了说明是loop出问题了

里面有行代码：

    if (this.cancelSubscriptionOnError.test(ex)) {this.cancel();}

也就是说在cancelSubscriptionOnError.test为true的时候会取消执行
在这里插入图片描述

还记得isActive()方法吗它正是去判断该状态的.

通过构造方法可以看出该参数是StreamMessageListenerContainer.StreamReadRequest streamRequest 传进来的
在这里插入图片描述

StreamMessageListenerContainer.StreamReadRequest在我们查看listenerContainer.receive源码时有过一面之缘：
在这里插入图片描述

我们再看看StreamReadRequest.builder出来的StreamReadRequestBuilder类：
在这里插入图片描述
至此，分析完成了闭环，因为receive方法创建出来默认是遇到异常就取消执行这明显不符合实际使用，这个设计个人感觉非常欠佳。

这便是为什么使用以下代码来创建的原因

     StreamMessageListenerContainer.ConsumerStreamReadRequest<String> build = StreamMessageListenerContainer.StreamReadRequest.builder(StreamOffset.create(streamName, ReadOffset.lastConsumed())).consumer(Consumer.from(groupName, consumerName)).autoAcknowledge(false)// 重要！.cancelOnError(t -> false).build();

===== ====== ====== 踩坑end ===== ==== ===== =====

定时器

代码比较乱注释代码比较多的原因不是因为瞎写，而是那些api 在实际业务中可能会使用到，所以特地写在下面了

// 定期处理 pending list 中的消息@Scheduled(cron = "0/20 * * * * ?")public void processPendingMessages() {String streamKey = "orderStream"; // Redis Stream 的键String groupName = "orderGroup";  // 消费者组的名称String consumerName = "orderConsumer"; // 当前消费者的名称for (Subscription each : subscription) {System.out.println(each.isActive());}StreamOperations<String, String, String> streamOps = redisTemplate.opsForStream();// 获取 pending list 中未确认的消息概要PendingMessagesSummary pendingSummary = streamOps.pending(streamKey, groupName);// 所有pending消息的数量long totalPendingMessages = pendingSummary.getTotalPendingMessages();if (pendingSummary.getTotalPendingMessages() == 0L) {return;}// 消费组名称String groupName1 = pendingSummary.getGroupName();// pending队列中的最小IDString minMessageId = pendingSummary.minMessageId();// pending队列中的最大IDString maxMessageId = pendingSummary.maxMessageId();if (pendingSummary.getTotalPendingMessages() > 0) {// 读取消费者pending队列的前10条记录，从ID=0的记录开始，一直到ID最大值
//            PendingMessages pendingMessages = streamOps.pending(streamKey, Consumer.from(groupName, consumerName), Range.closed("0", "+"), 10);// 获取 pending list 中具体的消息PendingMessages pendingMessages = streamOps.pending(streamKey, groupName, Range.unbounded(), 10000);int size = pendingMessages.size();// 获取当前批次的消息PendingMessage currentBatchMin = pendingMessages.get(0);PendingMessage currentBatchMax = pendingMessages.get(size-1);pendingMessages.forEach(pendingMessage ->{// 消息被获取的次数 可以根据次数做不同业务 超过一定次数未消费 考虑是否要ack并dellong deliveryCount = pendingMessage.getTotalDeliveryCount();// 读取每个未确认的消息
//                List<MapRecord<String,String,String>> messages = streamOps.read(
//                        StreamReadOptions.empty(),
//                        StreamOffset.create(streamKey,ReadOffset.lastConsumed())
                        StreamOffset.create(streamKey,ReadOffset.from("0"))
//                );List<MapRecord<String, String, String>> messages = streamOps.range(streamKey, Range.closed(currentBatchMin.getId().toString(), currentBatchMax.getId().toString()), Limit.limit().count(10000));for (MapRecord<String, String, String> message : messages) {try {// 处理消息processMessage(message);// 成功处理后确认消息streamOps.acknowledge(streamKey, groupName, message.getId());streamOps.delete(streamKey, message.getId());} catch (Exception e) {// 处理异常情况e.printStackTrace();}}});}}

至于如何触发就比较简单了，往redis添加一个streamKey即可

   @GetMapping("/stream")public String testStream() {String mystream = "";for (int i = 0; i < 10; i++) {Oper oper = new Oper();oper.setTestId(11111111L);oper.setTestDesc("订单消息队列");oper.setVersion(i);oper.setTestXxx(LocalDateTime.now().toString());Map<String, Object> map = new HashMap<>();map.put("oper", oper);try {Thread.sleep(10);mystream = redisStreamUtil.addMap("orderStream", map);} catch (InterruptedException e) {throw new RuntimeException(e);}}return String.valueOf(mystream);}

优化方向

建立一个消费者抽象类，定义消费方法
建议一个降级处理抽象类，定义补偿方法（即消费失败时的处理）
定义spring的properties类把生产者消费者字段写到里面
redis需要部署集群，可在博主的主页搜索哨兵，有哨兵架构教程。
实际业务中，消费消息很可能是存入数据库，在入库完成之后 redis ack完成之前，如果这一瞬间突然宕机了，而数据量又非常大，可能会导致消费重复的情况，因为没有完成ack 下次还是会把该数据从pending list里面取出来。

解决方案1 ：考虑是加redisson锁
解决方案2：数据库存入消息id字段并建立唯一索引
(唯一索引的魅力体现出来了)

至此，一份生产级别的redis stream mq架构成立。