Redis5.0 Stream实现轻量化消息队列（一文实现Java实战完整版）
Posted AdobePeng
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Redis5.0 Stream实现轻量化消息队列（一文实现Java实战完整版）相关的知识，希望对你有一定的参考价值。
背景
本人近期在搞一个轻量化部署，需要用到消息队列，但是感觉kafka相对较重，所以最终选择了一个相对轻量化消息队列“Redis Stream”。感觉网上的Java实现不是很好，经过一段时间摸索，决定将完整的可运行的使用Java实现Redis消息队列写出来，供大家参考。代码已上传至gitee，文末可下载。
还是希望大家能够耐心的看完，相信看完本文可以帮助您快速了解Redis作为消息中间件在Java环境中的开发。
撸代码之前需要掌握
首先需要了解一下Stream的基础知识，这里给个链接，里面是针对官网的翻译版，可以先行参照了解一下每个命令的使用
本文用到的命令如下：
> XADD mystream * hello world 创建一个mystream 流
> XGROUP CREATE mystream group-1 $ 创建消费组group-1
> XGROUP CREATE mystream group-2 $ 创建消费组group-2
> XRANGE mystream - + 查询流中的消息
> XPENDING mystream group-1 没有组中没有ACK的消息
> XPENDING mystream group-1 0 + 10 consumer-1 查看消费中consumer-1中没有消费的消息
还包括XCLAIM 转组命令、XTRIM 定时清理流数据命令等在java中实现
环境准备
我使用的是redis5.0.2，这里直接为大家送上redis在linux上安装的源文件。
链接：https://pan.baidu.com/s/1dYrf6vC8mNS-6O_D88j7Lw 提取码：dwh8
准备三个Spring boot工程，一个生产者producer，两个消费者consumer1、consumer2
开撸~撸~
首先需要创建我们的流stream，以及相应的组group，这个可以手动在redis中创建，也可以代码自动创建。备注：这里再啰嗦一句，对于这些流以及组的概念本文就不进行重述了，还望大家先了解一下基本的操作。
我们这里创建一个流：mystream；两个组：group-1、group-2
127.0.0.1:6379> XADD mystream * hello world
"1617952839936-0"
127.0.0.1:6379> XGROUP CREATE mystream group-1 $
OK
127.0.0.1:6379> XGROUP CREATE mystream group-2 $
OK
127.0.0.1:6379> 下面我们开始通过Java代码来实现生产者和消费者，我使用的Spring Boot版本是2.4.3
生产者
代码结构如下
pom中需要引入如下
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-pool2</artifactId>
</dependency>配置文件application.yml，链接redis设置流的名称
server:
port: 8080
servlet:
context-path: /
spring:
redis:
database: 0
host: 192.168.44.129
port: 6379
password:
timeout: 0
lettuce:
pool:
max-active: 8
max-wait: -1
max-idle: 8
min-idle: 0
redisstream:
stream: mystreamRedisStreamConfig只是做了读取配置中流的名称
@Data
@Component
@ConfigurationProperties(prefix = "redisstream")
public class RedisStreamConfig
private String stream;
PublishService类中做了简单的发送操作，这里我们通过调用该test方法可以将数据发送到相应的流中
@Service
public class PublishService
@Autowired
private StringRedisTemplate stringRedisTemplate;
@Autowired
private RedisStreamConfig redisStreamConfig;
public void test(String msg)
// 创建消息记录, 以及指定stream
StringRecord stringRecord = StreamRecords.string(Collections.singletonMap("name", msg)).withStreamKey(redisStreamConfig.getStream1());
// 将消息添加至消息队列中
this.stringRedisTemplate.opsForStream().add(stringRecord);
RedisStreamController类里面做了一个简单的调用
@RestController
public class RedisStreamController
@Autowired
private PublishService publishService;
@GetMapping("produceMsg")
public void produceMsg(@Param("msg")String msg)
publishService.test(msg);
至此一个简单的Redis生产者就已经完成了，我们来测试一下
打开浏览器输入: localhost:8080/produceMsg?msg=nihao
执行完成，通过命令>XRANGE mystream - + 我们可以看到刚输入的“nihao”已经存入该流中。
127.0.0.1:6379> XRANGE mystream - +
1) 1) "1617952839936-0"
2) 1) "hello"
2) "world"
2) 1) "1617955133254-0"
2) 1) "name"
2) "nihao"
127.0.0.1:6379>
消费者
代码结构如下，只需要关注红框内的文件就好，pom文件同生产者
application.yml文件如下，设置当前工程从group-1中消费，当前消费者名称为consumer-1
server:
port: 8081
servlet:
context-path: /
spring:
redis:
database: 0
host: 192.168.44.129
port: 6379
password:
timeout: 0
lettuce:
pool:
max-active: 8
max-wait: -1
max-idle: 8
min-idle: 0
redisstream:
stream: mystream
group: group-1
consumer: consumer-1
RedisStreamConfig类文件如下，简单的获取相应的配置
@Data
@Component
@ConfigurationProperties(prefix = "redisstream")
public class RedisStreamConfig
private String stream;
private String group;
private String consumer;
RedisStreamConsumerConfig类如下，重点都在这里，具体的功能写的还算详细，这里面只要包含了将消费者监听类绑定到响应的流上，以及拉取消息的一些配置。
这里面我们关闭了ACK自动消费，我们在消息监听类里面进行手动消费
@Configuration
public class RedisStreamConsumerConfig
@Autowired
ThreadPoolTaskExecutor threadPoolTaskExecutor;
@Autowired
RedisStreamConfig redisStreamConfig;
/**
* 主要做的是将OrderStreamListener监听绑定消费者，用于接收消息
*
* @param connectionFactory
* @param streamListener
* @return
*/
@Bean
public StreamMessageListenerContainer<String, ObjectRecord<String, String>> consumerListener1(
RedisConnectionFactory connectionFactory,
OrderStreamListener streamListener)
StreamMessageListenerContainer<String, ObjectRecord<String, String>> container =
streamContainer(redisStreamConfig.getStream(), connectionFactory, streamListener);
container.start();
return container;
/**
* @param mystream 从哪个流接收数据
* @param connectionFactory
* @param streamListener 绑定的监听类
* @return
*/
private StreamMessageListenerContainer<String, ObjectRecord<String, String>> streamContainer(String mystream, RedisConnectionFactory connectionFactory, StreamListener<String, ObjectRecord<String, String>> streamListener)
StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, ObjectRecord<String, String>> options =
StreamMessageListenerContainer.StreamMessageListenerContainerOptions
.builder()
.pollTimeout(Duration.ofSeconds(5)) // 拉取消息超时时间
.batchSize(10) // 批量抓取消息
.targetType(String.class) // 传递的数据类型
.executor(threadPoolTaskExecutor)
.build();
StreamMessageListenerContainer<String, ObjectRecord<String, String>> container = StreamMessageListenerContainer
.create(connectionFactory, options);
//指定消费最新的消息
StreamOffset<String> offset = StreamOffset.create(mystream, ReadOffset.lastConsumed());
//创建消费者
Consumer consumer = Consumer.from(redisStreamConfig.getGroup(), redisStreamConfig.getConsumer());
StreamMessageListenerContainer.StreamReadRequest<String> streamReadRequest = StreamMessageListenerContainer.StreamReadRequest.builder(offset)
.errorHandler((error) ->
)
.cancelOnError(e -> false)
.consumer(consumer)
//关闭自动ack确认
.autoAcknowledge(false)
.build();
//指定消费者对象
container.register(streamReadRequest, streamListener);
return container;
@Bean
public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory)
RedisTemplate<String, Object> template = new RedisTemplate<String, Object>();
template.setConnectionFactory(factory);
Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
ObjectMapper om = new ObjectMapper();
om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
jackson2JsonRedisSerializer.setObjectMapper(om);
StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
// key采用String的序列化方式
template.setKeySerializer(stringRedisSerializer);
// hash的key也采用String的序列化方式
template.setHashKeySerializer(stringRedisSerializer);
// value序列化方式采用jackson
template.setValueSerializer(jackson2JsonRedisSerializer);
// hash的value序列化方式采用jackson
template.setHashValueSerializer(jackson2JsonRedisSerializer);
template.afterPropertiesSet();
return template;
OrderStreamListener类是具体的监听类，用于拿取接收到的消息进行逻辑处理。
如果处理成功的话，我们进行手动ACK；
如果异常的话，如果是单击部署的话，我们可以针对业务类异常直接记录到DB或者文件中，然后手动ACK，如果是网络中断，超时等异常我们可以记录进行尝试重新消费
如果是分布式部署的话，加入一个消费组下面有多个消费者，其中一个消费失败了，如果是业务异常，直接ACK，如果是非业务性异常（即网络中断，超时等异常），我们将对其进行转组操作（后面会详细讲到）
@Component
public class OrderStreamListener implements StreamListener<String, ObjectRecord<String, String>>
static final Logger LOGGER = LoggerFactory.getLogger(OrderStreamListener.class);
@Autowired
StringRedisTemplate stringRedisTemplate;
@Autowired
RedisStreamConfig redisStreamConfig;
@Override
public void onMessage(ObjectRecord<String, String> message)
try
// 消息ID
RecordId messageId = message.getId();
// 消息的key和value
String string = message.getValue();
LOGGER.info("StreamMessageListener stream message。messageId=, stream=, body=", messageId, message.getStream(), string);
// 通过RedisTemplate手动确认消息
this.stringRedisTemplate.opsForStream().acknowledge(redisStreamConfig.getGroup(), message);
catch (Exception e)
// 处理异常
e.printStackTrace();
至此消费者1已经完成，我们可以先跑起来看下效果。刚刚我们已经往mystream中扔了一条name：nihao的数据，我们启动消费者看看能够消费到
我们可以看到，启动消费者1已经成功消费到数据。
还记得我们刚开始为该stream创建了两个组（group-1,group-2），我们刚刚创建的是消费者1，消费的组是group-1。那么我们现在再来创建一个消费者2，绑定组group-2。
这里我们直接复制一下上面的消费者工程如下：这里我们主要修改两个地方，一个是配置文件application.yml(改了端口号，改了组的配置信息)、一个是OrderStreamListener类（只是加了日志，用于和消费者1区分）
server:
port: 8082
servlet:
context-path: /
spring:
redis:
database: 0
host: 192.168.44.129
port: 6379
password:
timeout: 0
lettuce:
pool:
max-active: 8
max-wait: -1
max-idle: 8
min-idle: 0
redisstream:
stream: mystream
group: group-2
consumer: consumer-2
@Component
public class OrderStreamListener implements StreamListener<String, ObjectRecord<String, String>>
static final Logger LOGGER = LoggerFactory.getLogger(OrderStreamListener.class);
@Autowired
StringRedisTemplate stringRedisTemplate;
@Autowired
RedisStreamConfig redisStreamConfig;
@Override
public void onMessage(ObjectRecord<String, String> message)
try
// 消息ID
RecordId messageId = message.getId();
// 消息的key和value
String string = message.getValue();
LOGGER.info("StreamMessageListener stream message。messageId=, stream=,group=, body=", messageId, message.getStream(),redisStreamConfig.getGroup(), string);
// 通过RedisTemplate手动确认消息
this.stringRedisTemplate.opsForStream().acknowledge(redisStreamConfig.getGroup(), message);
catch (Exception e)
// 处理异常
e.printStackTrace();
此时我们启动消费者2，我们看一下nihao这条消息会不会被消费到
我们可以看到，同样被消费到了。大家是不是对Stream有那么点了解了呢。
结论：一条消息发送到一个Stream流中，那么每个group都存在这条消息。这就像kafka中的消费组的概念，一条消息发出去，每个消费组中都存在这一份。
到这里一个简单的消息队列就完成了，但是离真正使用还远，这里有几个问题：
- 能不能多个节点去消费同一份数据，就像kafka中一个消费组可以有多个消费者去消费同一份数据
- 消费端消费失败后去哪里找到这条记录
- 一个消费端如何绑定多个流
- 如何处理死信（如何判断、如何处理）
- 流中的数据一直存在内存中，时间长了内存不够怎么处理
上面都是我近期在开发的时候遇到的问题。当然这些也是使用消息队列必须要考虑的问题，接下来我们一一处理。
问题1：多个节点去消费同一份数据
那么我们需要对上面的代码进行一些改造，我们将工程consumer2中的组改成group-1，如下：
server:
port: 8082
servlet:
context-path: /
spring:
redis:
database: 0
host: 192.168.44.129
port: 6379
password:
timeout: 0
lettuce:
pool:
max-active: 8
max-wait: -1
max-idle: 8
min-idle: 0
redisstream:
stream: mystream
group: group-1
consumer: consumer-2
现在我们就是一个流(mystream) =》一个组(group-1) =》 两个消费者(consumer-1,consumer-2)
启动consumer2，通过生产者发送消息查看执行情况
我们连续调用发送消息接口
localhost:8080/produceMsg?msg=111
localhost:8080/produceMsg?msg=222
localhost:8080/produceMsg?msg=333
localhost:8080/produceMsg?msg=444
localhost:8080/produceMsg?msg=555
localhost:8080/produceMsg?msg=666
查看消费情况：我们可以看到消费者1消费了111、222、555, 消费者2消费了333、444、666。这就说明我们如果是分布式部署的话，多个节点消费同一份数据，连负载均衡都自动帮忙搞好了。
问题2：消费端消费失败后去哪里找到这条记录
此时我们停掉consumer2工程，保留producer工程和consumer工程，我们现在将consumer消费的地方ACK注释掉，我们来看一下这条消息去哪里了。
这时发送一条消息，localhost:8080/produceMsg?msg=777
我们可以看到消费者1消费到了这条消息，但是没有进行ACK确认。
如果看过Redis Stream基本操作的话应该知道，这条消息存在group-1的pending里面。
我们可以通过命令> XPENDING mystream group-1 查看组中有多少没有被确认消费的数据
或者> XPENDING mystream group-1 0 + 10 consumer-1 查看具体的那个组那个消费者没有消费的数据
可以看到有在组group-1中，消费者consumer-1存在一条消息没有被ACK
127.0.0.1:6379> XPENDING mystream group-1
1) (integer) 1
2) "1617962550864-0"
3) "1617962550864-0"
4) 1) 1) "consumer-1"
2) "1"
127.0.0.1:6379> XPENDING mystream group-1 0 + 10 consumer-1
1) 1) "1617962550864-0"
2) "consumer-1"
3) (integer) 74328
4) (integer) 1
127.0.0.1:6379>
问题3：一个消费端如何绑定多个流
这个目前我的做法是这样的
首先我们再创建一个流：mystream2；一个组：group-1
127.0.0.1:6379> XADD mystream2 * hello world
"1617970973509-0"
127.0.0.1:6379> XGROUP CREATE mystream2 group-1 $
OK
127.0.0.1:6379>
配置生产者
修改工程producer中的相应文件如下
application.yml 文件 增加了stream2
redisstream:
stream: mystream
stream2: mystream2
---------------------------------------------
RedisStreamConfig 类 增加了stream2
@Data
@Component
@ConfigurationProperties(prefix = "redisstream")
public class RedisStreamConfig
private String stream;
private String stream2;
---------------------------------------------
PublishService 类，往mystream发完之后继续调用私有方法，发往mystream2
@Service
public class PublishService
@Autowired
private StringRedisTemplate stringRedisTemplate;
@Autowired
private RedisStreamConfig redisStreamConfig;
public void test(String msg)
// 创建消息记录, 以及指定stream
StringRecord stringRecord = StreamRecords.string(Collections.singletonMap("name", msg)).withStreamKey(redisStreamConfig.getStream());
// 将消息添加至消息队列中
this.stringRedisTemplate.opsForStream().add(stringRecord);
// 发往流mystream2
sendToStream2(msg);
private void sendToStream2(String msg)
// 创建消息记录, 以及指定stream
StringRecord stringRecord = StreamRecords.string(Collections.singletonMap("name", msg)).withStreamKey(redisStreamConfig.getStream2());
// 将消息添加至消息队列中
this.stringRedisTemplate.opsForStream().add(stringRecord);
配置消费者
修改工程consumer中相应的配置文件，
application.yml 文件 增加了stream2
redisstream:
stream: mystream
stream2: mystream2
group: group-1
consumer: consumer-1
---------------------------------------------
RedisStreamConfig 类 增加了stream2
@Data
@Component
@ConfigurationProperties(prefix = "redisstream")
public class RedisStreamConfig
private String stream;
private String stream2;
private String group;
private String consumer;
新增监听类OrderStreamListener2 ，不需要改任何东西。这里我们需要将ACK都放开，用于接收mystream2中的消息，如下
修改类RedisStreamConsumerConfig，将监听类OrderStreamListener2绑定如下：同之前监听类绑定相同，注意修改红色标注的三个地方
这样一个消费端绑定多个流已经实现，我们现在来测试一下
调用 localhost:8080/produceMsg?msg=888
我们预期看到的是调用一次，消费端打印两条数据，流mystream、mystream2各一条。执行结果如下：
结果如我们预期，至此一个消费端绑定多个流已经完成；
问题4：如何处理死信（如何判断、如何处理）
何为死信？即消费端消费不了的消息。
环境：一个生产者，两个消费节点。生产者往流mystream中扔消息，两个消费节点consumer-1、consumer-2都从组group-1中消费消息
逻辑：如果consumer-1消费失败，没有ACK确认消费，那么由生产者去定时扫描group-1中没有被ack的消息（同上面XPENDING操作），此时可以获取到此条消息“从消费组中获取到此刻的时间”和“转组的次数”，如果消息超过20秒（该时间可根据系统需求自定义）没有被消费掉并且转组次数为1的情况下，我们就将其进行转组，转到消费者consumer-2中。如果获取到的转组次数为2，说明已经被转过组，这是还没有被消费掉，我们就默认这条消息有问题，我们就将其手动ACK掉。
话说的有点多，上代码：
配置生产者
修改配置文件：
application.yml 文件 修改如下：
redisstream:
stream: mystream
stream2: mystream2
group: group-1
consumer1: consumer-1
consumer2: consumer-2
----------------------------------------------------
RedisStreamConfig 类 修改如下：
@Data
@Component
@ConfigurationProperties(prefix = "redisstream")
public class RedisStreamConfig
private String stream;
private String stream2;
private String group;
private String consumer1;
private String consumer2;
增加定时器扫描没有被ACK的消息，具体逻辑代码注释写的还算详细。
@Component
public class RedisStreamScheduled
private static final Logger LOGGER = LoggerFactory.getLogger(RedisStreamScheduled.class);
@Autowired
private StringRedisTemplate stringRedisTemplate;
@Autowired
private RedisStreamConfig redisStreamConfig;
/**
* 每隔5秒钟，扫描一下有没有等待自己消费的
* 处理死信队列，如果发送给消费者1超过1分钟还没有ack，则转发给消费者2，如果超过20秒，并且转发次数为2，进行手动ack。并且记录异常信息
*/
@Scheduled(cron="0/5 * * * * ?")
public void scanPendingMsg()
StreamOperations<String, String, String> streamOperations = this.stringRedisTemplate.opsForStream();
// 获取group中的pending消息信息，本质上就是执行XPENDING指令
PendingMessagesSummary pendingMessagesSummary = streamOperations.pending(redisStreamConfig.getStream(), redisStreamConfig.getGroup());
// 所有pending消息的数量
long totalPendingMessages = pendingMessagesSummary.getTotalPendingMessages();
if(totalPendingMessages == 0)
return;
// 消费组名称
String groupName= pendingMessagesSummary.getGroupName();
// pending队列中的最小ID
String minMessageId = pendingMessagesSummary.minMessageId();
// pending队列中的最大ID
String maxMessageId = pendingMessagesSummary.maxMessageId();
LOGGER.info("流：,消费组：，一共有条pending消息，最大ID=，最小ID=", redisStreamConfig.getStream(),groupName, totalPendingMessages, minMessageId, maxMessageId);
// 获取每个消费者的pending消息数量
Map<String, Long> pendingMessagesPerConsumer = pendingMessagesSummary.getPendingMessagesPerConsumer();
Map<String,List<RecordId>> consumerRecordIdMap = new HashMap<>();
// 遍历每个消费者中的pending消息
pendingMessagesPerConsumer.entrySet().forEach(entry ->
// 待转组的 RecordId
List<RecordId> list = new ArrayList<>();
// 消费者
String consumer = entry.getKey();
// 消费者的pending消息数量
long consumerTotalPendingMessages = entry.getValue();
LOGGER.info("消费者：，一共有条pending消息", consumer, consumerTotalPendingMessages);
if (consumerTotalPendingMessages > 0)
// 读取消费者pending队列的前10条记录，从ID=0的记录开始，一直到ID最大值
PendingMessages pendingMessages = streamOperations.pending(redisStreamConfig.getStream(), Consumer.from(redisStreamConfig.getGroup(), consumer), Range.closed("0", "+"), 10);
// 遍历所有Opending消息的详情
pendingMessages.forEach(message ->
// 消息的ID
RecordId recordId = message.getId();
// 消息从消费组中获取，到此刻的时间
Duration elapsedTimeSinceLastDelivery = message.getElapsedTimeSinceLastDelivery();
// 消息被获取的次数
long deliveryCount = message.getTotalDeliveryCount();
// 判断是否超过60秒没有消费
if(elapsedTimeSinceLastDelivery.getSeconds()>20)
// 如果消息被消费的次数为1，则进行一次转组，否则手动消费
if( 1 == deliveryCount )
list.add(recordId);
else
LOGGER.info("手动ACK消息,并记录异常，id=, elapsedTimeSinceLastDelivery=, deliveryCount=", recordId, elapsedTimeSinceLastDelivery, deliveryCount);
streamOperations.acknowledge(redisStreamConfig.getStream(),redisStreamConfig.getGroup(),recordId);
);
if(list.size()>0)
consumerRecordIdMap.put(consumer,list);
);
// 最后将待转组的消息进行转组
if(!consumerRecordIdMap.isEmpty())
this.changeConsumer(consumerRecordIdMap);
/**
* 将消息进行转组
* @param consumerRecordIdMap
*/
private void changeConsumer(Map<String,List<RecordId>> consumerRecordIdMap)
consumerRecordIdMap.entrySet().forEach(entry ->
// 根据当前consumer去获取另外一个consumer
String oldComsumer = entry.getKey();
String newConsumer = redisStreamConfig.getConsumers().stream().filter(s -> !s.equals(oldComsumer)).collect(Collectors.toList()).get(0);
List<RecordId> recordIds = entry.getValue();
List<ByteRecord> retVal = this.stringRedisTemplate.execute(new RedisCallback<List<ByteRecord>>()
@Override
public List<ByteRecord> doInRedis(RedisConnection redisConnection) throws DataAccessException
// 相当于执行XCLAIM操作，批量将某一个consumer中的消息转到另外一个consumer中
return redisConnection.streamCommands().xClaim(redisStreamConfig.getStream().getBytes(),
redisStreamConfig.getGroup(), newConsumer, minIdle(Duration.ofSeconds(10)).ids(recordIds));
);
for (ByteRecord byteRecord : retVal)
LOGGER.info("改了消息的消费者：id=, value=,newConsumer=", byteRecord.getId(), byteRecord.getValue(),newConsumer);
);
不知道大家还记不记得我们上面有一条777的数据发送到消费者1，但是没有ACK的数据，此刻具体接收时间已经远超过20秒钟，那么我们现在运行程序，看看能够将其转组到消费者2中。
执行结果如下：
我们可以从日志上面看到，已经转组成功，那么我们现在到redis中执行命令>XPENDING mystream group-1 0 + 10 consumer-2 看看是否已经转到了consumer-2中
127.0.0.1:6379> XPENDING mystream group-1 0 + 10 consumer-2
1) 1) "1617962550864-0"
2) "consumer-2"
3) (integer) 64710
4) (integer) 2
127.0.0.1:6379>
显然这条消息已经转到了 consumer-2 中，大家可以看到，3）和4），分别代表着接收到消息的时间和转组的次数。
但是此时我们的consumer2工程并不能收到这条转组的消息，因为这条消息只是从consumer-1的pending中转移到了consumer-2的pending中，想要消费必须使用定时器定时扫秒消费。
下面我们来看看如何让consumer2消费到这条消息
配置消费者consumer2：
修改配置文件
application.yml 文件修改如下：
redisstream:
stream: mystream
group: group-1
consumer: consumer-2
---------------------------------------------
RedisStreamConfig 类修改如下：
@Data
@Component
@ConfigurationProperties(prefix = "redisstream")
public class RedisStreamConfig
private String stream;
private String group;
private String consumer;
新增定时器，定时扫描pending中的消息。这里面我们要注意，这个if（totalDeliveryCount > 1）判断。我们只去消费转组次数大于1的，避免新传递过来的消息重复消费的情况
@Component
public class ScheduleJob
static final Logger LOGGER = LoggerFactory.getLogger(ScheduleJob.class);
@Autowired
StringRedisTemplate stringRedisTemplate;
@Autowired
RedisStreamConfig redisStreamConfig;
/**
* 每隔5秒钟，扫描一下有没有等待自己消费的
* 主要消费那些转组过来的消息，如果转组次数大于1，则进行尝试消费
*/
@Scheduled(cron="0/5 * * * * ?")
public void reportCurrentTime()
StreamOperations<String, String, String> streamOperations = this.stringRedisTemplate.opsForStream();
/*从消费者的pending队列中读取消息，能够进到这里面的，一定是非业务异常，例如接口超时、服务器宕机等。
对于业务异常，例如字段解析失败等，丢进异常表或者redis*/
PendingMessages pendingMessages = streamOperations.pending(redisStreamConfig.getStream(), Consumer.from(redisStreamConfig.getGroup(), redisStreamConfig.getConsumer()));
if(pendingMessages.size() > 0)
pendingMessages.forEach( pendingMessage ->
// 最后一次消费到现在的间隔
Duration elapsedTimeSinceLastDelivery = pendingMessage.getElapsedTimeSinceLastDelivery();
// 转组次数
long totalDeliveryCount = pendingMessage.getTotalDeliveryCount();
// 只消费转组次数大于1次的
if(totalDeliveryCount > 1)
try
RecordId id = pendingMessage.getId();
List<MapRecord<String, String, String>> result = streamOperations.range(redisStreamConfig.getStream(), Range.rightOpen(id.toString(),id.toString()));
MapRecord<String, String, String> entries = result.get(0);
// 消费消息
LOGGER.info("获取到转组的消息，消费了该消息id=, 消息value=, 消费者=", entries.getId(), entries.getValue(),redisStreamConfig.getConsumer());
// 手动ack消息
streamOperations.acknowledge(redisStreamConfig.getGroup(), entries);
catch (Exception e)
// 异常处理
e.printStackTrace();
);
运行程序，查看结果：已经成功的消费到了这条消息，并且手动ACK掉。
此时我们查看redis中已经没有待消费的消息了：
127.0.0.1:6379> XPENDING mystream group-1 0 + 10 consumer-2
(empty list or set)
127.0.0.1:6379> XPENDING mystream group-1 0 + 10 consumer-1
(empty list or set)
127.0.0.1:6379>
问题5：流中的数据一直存在内存中，时间长了内存不够怎么处理
这个当然Redis Stream也帮我们想到了这个问题，并且看到过官网介绍的也应该知道主要有两个命令来控制。分别是XTRIM和MAXLEN
相比之下，使用XTRIM在java中更为合理，在producer中新增一个定时器，定时清理数据
@Component
public class CleanStreamJob
@Autowired
private StringRedisTemplate stringRedisTemplate;
@Autowired
private RedisStreamConfig redisStreamConfig;
@Scheduled(cron="0/5 * * * * ?")
public void reportCurrentTime()
// 定时的清理stream中的数据，保留3条
this.stringRedisTemplate.opsForStream().trim(redisStreamConfig.getStream(),3L);
// // 定时的清理stream中的数据，保留3条左右，不少于3条
// this.stringRedisTemplate.opsForStream().trim(redisStreamConfig.getStream(),3L,true);
具体定时器的执行时间以及保留条数大家可自行根据业务进行修改。
启动执行，之前我们的stream中已经存在多条记录，执行完应该还剩最后三条。通过> XINFO STREAM mystream命令查看：
127.0.0.1:6379> XINFO STREAM mystream
1) "length"
2) (integer) 3
3) "radix-tree-keys"
4) (integer) 1
5) "radix-tree-nodes"
6) (integer) 2
7) "groups"
8) (integer) 2
9) "last-generated-id"
10) "1617972388360-0"
11) "first-entry"
12) 1) "1617960242592-0"
2) 1) "name"
2) "666"
13) "last-entry"
14) 1) "1617972388360-0"
2) 1) "name"
2) "888"
127.0.0.1:6379>
至此，整套关于Redis Stream分布式消息队列Java开发已经完成。
创作不易，还望大家点赞支持。
附上git地址：AdobePeng/RedisStream
以上是关于Redis5.0 Stream实现轻量化消息队列（一文实现Java实战完整版）的主要内容，如果未能解决你的问题，请参考以下文章
别再用 Redis List 实现消息队列了，Stream 专为队列而生
Redis Stream 流的深度解析与实现高级消息队列一万字
Redis Stream 流的深度解析与实现高级消息队列一万字