Redis stream 是 Redis 5 引入的一种新的数据结构,它是一个高性能、高可靠性的消息队列,主要用于异步消息处理和流式数据处理。在此之前,想要使用 Redis 实现消息队列,通常可以使用例如:列表,有序集合、发布与订阅 3 种数据结构。但是 stream 相比它们具有以下的优势:
- 支持范围查找:内置的索引功能,可以通过索引来对消息进行范围查找
- 支持阻塞操作:避免低效的反复轮询查找消息
- 支持 ACK:可以通过确认机制来告知已经成功处理了消息,保证可靠性
- 支持多个消费者:多个消费者可以同时消费同一个流,Redis 会确保每个消费者都可以独立地消费流中的消息
情况
当前项目,是在window server 上 部署 rocketmq (ps 单体)使用 消息队列,使用三方exe实现 注册到 window 服务中(实现开机自启-自行搜索实现吧)。
问题
当前发现一个问题,window 被关机(主动、断电等),mq的broker无法启动,原因是 delayOffset.json文件损坏造成,文件内容变成如下图。
修改为,重新启动没问题了
每次重新都需要处理,虽然可以考虑使用脚本 处理,但是徒增成本。就想着 使用redis实现 进行替换(项目规模较小)。redis-stream 需要将 版本升级为 5(当前是4)。
Redis for Windows
redis下载 直接替换就行。
死信问题
网上看到 处理的方式有好几种。定时调度,处理pending信息(直接ack、重新消费、转移消费组等);人工运维处理 等。看了很多,结合 spring 提供的函数。
StringRedisTemplate.redisTemplate.opsForStream()
返回的接口类StreamOperations
参考spring文档,可以用的方法有 acknowledge
、 add
、 delete
、 createGroup
、 deleteConsumer
、 destroyGroup
、 consumers
、 groups
、 info
、 pending
、 size
、 range
、 read
、 reverseRange
、 trim
等。
没有包含命令:转移消费 xClaim
,实际在RedisStreamCommands
接口中,通过代码也能解决;
redisTemplate.execute((RedisCallback<List<ByteRecord>>) connection -> connection.streamCommands().xClaim("stream".getBytes(), "group", "consumer", Duration.ofSeconds(10), RecordId.of("streamId")))
也没有包含命令:设置消费组的起始消息 ID xgroupSetid
,实际在RedisStreamAsyncCommands
接口中,代码在比较底层,不能拿来就用。
RedisFuture<String> xgroupSetid(StreamOffset<K> streamOffset, K group);
当前方式
当前,实现的方式。在项目 重启 或 定时调度(暂无代码demo),通过 pending
获取消息 streamId
,先复制消息add
,在确认旧消息acknowledge
,最后删除旧消息delete
。
ps:自己感觉还行吧(迷之自信O(∩_∩)O哈哈~),不要拿来就用啊(⊙o⊙)…
redis桌面管理
下载redis可视化管理工具:AnotherRedisDesktopManager
参考
理解 Redis 新特性:Stream
springboot + redis stream做轻量级消息队列
Redis Stream实现消息队列
redis Stream消息队列 redision redis stream消息队列pending
使用redis流和spring数据获取挂起的消息
在SpringBoot中使用RedisTemplate重新消费Redis Stream中未ACK的消息
说明
redis-stream命令
- XADD:向流中添加新的消息。
- XREAD:从流中读取消息。
- XREADGROUP:从消费组中读取消息。
- XRANGE:根据消息 ID 范围读取流中的消息。
- XREVRANGE:与 XRANGE 类似,但以相反顺序返回结果。
- XDEL:从流中删除消息。
- XTRIM:根据 MAXLEN 参数修剪流的长度。
- XLEN:获取流的长度。
- XGROUP:管理消费组,包括创建、删除和修改。
- XACK:确认消费组中的消息已被处理。
- XPENDING:查询消费组中挂起(未确认)的消息。
- XCLAIM:将挂起的消息从一个消费者转移到另一个消费者。
- XINFO:获取流、消费组或消费者的详细信息。
pom
spring-boot 加 redis依赖,简单项目,测试用
<parent>
<artifactId>spring-boot-starter-parent</artifactId>
<groupId>org.springframework.boot</groupId>
<version>2.7.0</version>
<relativePath/>
</parent>
<properties>
<java.version>1.8</java.version>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
<spring-boot.version>2.7.0</spring-boot.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<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>
</dependencies>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-dependencies</artifactId>
<version>${spring-boot.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
配置
server:
port: 8888
servlet:
context-path: /
spring:
redis:
msg:
listener: false
host: 127.0.0.1
port: 6379
database: 1
client-type: lettuce
lettuce:
pool:
max-active: 8
代码结构
引入spring boot的 StringRedisTemplate
private static StringRedisTemplate redisTemplate;
@Autowired
public void setRedisTemplate(StringRedisTemplate redisTemplate) {
RedisStreamConfig.redisTemplate = redisTemplate;
}
声明变量。redis 的 key键名称,消费者 组名,消费者 客户端名。
final static String STREAM = "tsp", GROUP = "tsp-g-1", CONSUMER = "tsp-c-1";
生产者-发送消息
这块比较简单,直接调用就行
/** 生产者-发送消息 */
public static void push(String msg){
// 创建消息记录, 以及指定stream
StringRecord record = StreamRecords.string(Collections.singletonMap("data", msg)).withStreamKey(STREAM);
// 将消息添加至消息队列中 XADD stream [MAXLEN len] id field value [field value ...]
redisTemplate.opsForStream().add(record);
log.info("redis-消息队列-stream, {} ,send msg: {}", STREAM,msg);
}
消费者
消费者-监听类,实际的业务逻辑处理内容。消息队列 创建 会有测试消息,需要跳过。实际业务逻辑需要考虑,挂起消息 重新消费情况(解决 死信问题的简单方法)
/** 消费者 监听 */
public static class TspStreamListener implements StreamListener<String, ObjectRecord<String,String>> {
@Override
public void onMessage(ObjectRecord<String, String> message) {
RecordId messageId = message.getId();
// 消息的key和value
String string = message.getValue();
if("T".equals(string)){
log.info("消费者-监听>>>测试消息-ack. msgId={}, stream={}, body={}", messageId, message.getStream(), string);
redisTemplate.opsForStream().acknowledge(GROUP, message);
return;
}
log.info("消费者-监听>>>get msg. msgId={}, stream={}, body={}", messageId, message.getStream(), string);
//业务逻辑,需要考虑 被挂起的消息 重新消费情况
try {
log.info("消费者-监听>>>睡眠10s,模拟耗时逻辑-start");
TimeUnit.SECONDS.sleep(10);
log.info("消费者-监听>>>睡眠10s,模拟耗时逻辑-end");
} catch (InterruptedException e) {
e.printStackTrace();
}
log.info("消费者-监听>>>手动确认消息");
redisTemplate.opsForStream().acknowledge(GROUP, message);
log.info("消费者-监听>>>end");
}
}
消费者-注册,关闭自动ack,重连自动消费上次处理(已经在pending队列里面了)的下一个消息,拉取消息超时时间50s,每批数量1,使用默认线程池,传递数据类型 String,异常处理 打印错误日志。最后 start 启动消费。
@Bean
@ConditionalOnProperty(name = "spring.redis.msg.listener",havingValue = "true")
public StreamMessageListenerContainer<String, ObjectRecord<String, String>> tspConsumerListener(
RedisConnectionFactory factory){
//spring-data-redis 2.3.1.RELEASE及更高版本中,createGroup如果不存在,则会自动使用创建流
return streamContainer(StreamOffset.create(STREAM, ReadOffset.lastConsumed()),Consumer.from(GROUP,CONSUMER)
,factory,new TspStreamListener());
}
private StreamMessageListenerContainer<String, ObjectRecord<String, String>> streamContainer(StreamOffset<String> offset
,Consumer consumer, RedisConnectionFactory factory, StreamListener<String, ObjectRecord<String, String>> listener) {
// pollTimeout 拉取消息超时时间,targetType 传递的数据类型, executor 线程池
StreamMessageListenerContainer<String, ObjectRecord<String, String>> container = StreamMessageListenerContainer
.create(factory, StreamMessageListenerContainer.StreamMessageListenerContainerOptions
.builder().pollTimeout(Duration.ofSeconds(50)).batchSize(1).targetType(String.class).build());
//指定消费最新的消息
try {
//创建消费者,接收上次处理未ACK消费的消息,指定消费者对象,autoAcknowledge 关闭自动ack确认
container.register(StreamMessageListenerContainer.StreamReadRequest
.builder(offset).errorHandler((error) -> log.error(error.getMessage()))
.cancelOnError(e -> false).consumer(consumer).autoAcknowledge(false)
.build(), listener);
} catch (Exception e) {
log.error(e.getMessage());
}
container.start();
return container;
}
预处理
类加载完毕,执行方法。预处理操作,可以使用 spring 启动后处理类实现。本demo就先这样了。
本方法 是为了 解决,项目启动报错(无消息队列、无消费组)和程序重启死信问题(꒦_꒦)
@PostConstruct
private void start(){
}
如果redis 没有 对应key 的 stream 消息队列,消费者启动后会频繁报错。如下代码会在 项目启动 发送测试消息,创建 消息队列,并限制 消息队列 长度(限制内存消耗)。
ps:我好菜呀,不知道spring是否允许消费者 自动创建 消息队列o(╥﹏╥)o
// 生产者 创建 消息队列,防止启动 消费者 报错
if(!(Boolean.TRUE.equals(redisTemplate.hasKey(STREAM)))){
log.info("没有key,测试发送(创建key)");
push("T");
//限制 消息队列 容量 XTRIM stream MAXLEN len
redisTemplate.opsForStream().trim(STREAM,1000L);
}
判断消息队列,是否存在消费组。没有创建消费组
// 创建 消息队列-消费者
if (redisTemplate.opsForStream().groups(STREAM).isEmpty()) {
log.info("redis-消息队列-stream,createGroup,{} {}",STREAM,GROUP);
//XGROUP CREATE stream group [id|$|0] [MKSTREAM],使用 $ 表示仅消费新消息,或者使用 0 表示消费流中的所有消息
redisTemplate.opsForStream().createGroup(STREAM,GROUP);
}
死信处理代码
private void handlerPending(String key,String group){
//判断是否存在挂起的消息 XPENDING stream group [start stop count] [consumer]
PendingMessagesSummary pending = redisTemplate.opsForStream().pending(key, group);
long size;
if(pending == null || (size = pending.getTotalPendingMessages()) <= 0 ) {
log.debug("redis-消息队列,{},{},暂无挂起消息pending", key, group);
return;
}
//---------------------
String minId = pending.minMessageId(),maxId = pending.maxMessageId(),id;
// 从挂起消息开始处理 [1],minId : 1706178903044-0,maxId : 1706178903044-0
log.info("redis-消息队列,{},{},pending-挂起消息[{}],minId : {},maxId : {}",key,group,size,minId,maxId);
// - + 4 ---------------------获取挂起所有信息 streamId
PendingMessages msgIds = redisTemplate.opsForStream().pending(key, group, Range.closed("-", "+"), size);
log.info("redis-消息队列,{},{},pending-挂起消息,{}",key,group,msgIds);
List<MapRecord<String, Object, Object>> list;
// ---------------------循环处理,可以考虑 异步执行
for (PendingMessage msgId : msgIds) {
id = msgId.getId().getValue();
//PendingMessage{id=1706178903044-0, consumer=tsp-g-1:tsp-c-1, elapsedTimeSinceLastDeliveryMS=66775, totalDeliveryCount=1}
log.info("redis-消息队列,{},{},pending-挂起消息>>,{}",key,group,msgId);
//XRANGE key start end [COUNT count]
list = redisTemplate.opsForStream().range(key, Range.just(msgId.getIdAsString()));
log.info("redis-消息队列,{},{},range-挂起消息>> {},{}",key,group,id,list);
if(list == null || list.isEmpty()){
continue;
}
// 开始 结束 id相同,只返回 一个消息结果
MapRecord<String, Object, Object> record = list.get(0);
//MapBackedRecord{recordId=1706233950802-0, kvMap={data=401}}
if("1706233940011-0".equals(id)){
MapRecord<String, Object, Object> copy = record.withId(RecordId.autoGenerate());
log.info("redis-消息队列,{},{},range-挂起消息>0>add+++++copy {},{}",key,group,copy.getStream(),copy);
//XADD stream-name id field value [field value]
RecordId add = redisTemplate.opsForStream().add(copy);
log.info("redis-消息队列,{},{},range-挂起消息>0>add+++++copy {},{}",key,group,add,copy);
log.info("redis-消息队列,{},{},range-挂起消息>0>ack^^^^^old {},{}",key,group,record.getStream(),record);
//XACK stream group id [id id ...]
Long siz = redisTemplate.opsForStream().acknowledge(GROUP, record);
log.info("redis-消息队列,{},{},range-挂起消息>0>ack^^^^^old [{}],{}",key,group,siz,record);
log.info("redis-消息队列,{},{},range-挂起消息>0>del-----old {},{}",key,group,record.getStream(),record);
//XDEL key ID [ID ...]
siz = redisTemplate.opsForStream().delete(record);
log.info("redis-消息队列,{},{},range-挂起消息>0>del-----old [{}],{}",key,group,siz,record);
}else{
log.info("redis-消息队列,{},{},range-挂起消息>0> {},{},{}",key,group,record.getStream(),id,record);
}
}
log.info("redis-消息队列,{},{},pending-挂起消息-end",key,group);
}
汇总
代码
package demo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.context.annotation.Lazy;
import org.springframework.data.domain.Range;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import javax.annotation.PostConstruct;
import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;
/**
* @author z.y.l
* @version v1.0
* @date 2024/1/25
*/
@Configuration
@Import (RedisAutoConfiguration.class)
public class RedisStreamConfig {
private static final Logger log = LoggerFactory.getLogger(RedisStreamConfig.class);
final static String STREAM = "tsp", GROUP = "tsp-g-1", CONSUMER = "tsp-c-1";
private static StringRedisTemplate redisTemplate;
@Autowired
public void setRedisTemplate(StringRedisTemplate redisTemplate) {
RedisStreamConfig.redisTemplate = redisTemplate;
}
/** 生产者-发送消息 */
public static void push(String msg){
// 创建消息记录, 以及指定stream
StringRecord record = StreamRecords.string(Collections.singletonMap("data", msg)).withStreamKey(STREAM);
// 将消息添加至消息队列中 XADD stream [MAXLEN len] id field value [field value ...]
redisTemplate.opsForStream().add(record);
log.info("redis-消息队列-stream, {} ,send msg: {}", STREAM,msg);
}
/** 项目启动 预处理,无消息队列(消费者-启动报错)、限制容量、手动创建消费组、处理异常挂起的消息 */
@PostConstruct
private void start(){
// 生产者 创建 消息队列,防止启动 消费者 报错
if(!(Boolean.TRUE.equals(redisTemplate.hasKey(STREAM)))){
log.info("没有key,测试发送(创建key)");
push("T");
//限制 消息队列 容量 XTRIM stream MAXLEN len
redisTemplate.opsForStream().trim(STREAM,1000L);
}
// 创建 消息队列-消费者
if (redisTemplate.opsForStream().groups(STREAM).isEmpty()) {
log.info("redis-消息队列-stream,createGroup,{} {}",STREAM,GROUP);
//XGROUP CREATE stream group [id|$|0] [MKSTREAM],使用 $ 表示仅消费新消息,或者使用 0 表示消费流中的所有消息
redisTemplate.opsForStream().createGroup(STREAM,GROUP);
}else{
// 处理 挂起消息,也可以 使用调度 定时处理
// 当前 挂起消息处理方式:复制新消息-ack旧消息-删除旧消息
// 也可以使用 转移 消费组
handlerPending(STREAM,GROUP);
}
}
private void handlerPending(String key,String group){
//判断是否存在挂起的消息 XPENDING stream group [start stop count] [consumer]
PendingMessagesSummary pending = redisTemplate.opsForStream().pending(key, group);
long size;
if(pending == null || (size = pending.getTotalPendingMessages()) <= 0 ) {
log.debug("redis-消息队列,{},{},暂无挂起消息pending", key, group);
return;
}
String minId = pending.minMessageId(),maxId = pending.maxMessageId(),id;
// 从挂起消息开始处理 [1],minId : 1706178903044-0,maxId : 1706178903044-0
log.info("redis-消息队列,{},{},pending-挂起消息[{}],minId : {},maxId : {}",key,group,size,minId,maxId);
// - + 4 所有
PendingMessages msgIds = redisTemplate.opsForStream().pending(key, group, Range.closed("-", "+"), size);
log.info("redis-消息队列,{},{},pending-挂起消息,{}",key,group,msgIds);
List<MapRecord<String, Object, Object>> list;
for (PendingMessage msgId : msgIds) {
id = msgId.getId().getValue();
//PendingMessage{id=1706178903044-0, consumer=tsp-g-1:tsp-c-1, elapsedTimeSinceLastDeliveryMS=66775, totalDeliveryCount=1}
log.info("redis-消息队列,{},{},pending-挂起消息>>,{}",key,group,msgId);
//XRANGE key start end [COUNT count]
list = redisTemplate.opsForStream().range(key, Range.just(msgId.getIdAsString()));
log.info("redis-消息队列,{},{},range-挂起消息>> {},{}",key,group,id,list);
if(list == null || list.isEmpty()){
continue;
}
// 开始 结束 id相同,只返回 一个消息结果
MapRecord<String, Object, Object> record = list.get(0);
//MapBackedRecord{recordId=1706233950802-0, kvMap={data=401}}
if("1706233940011-0".equals(id)){
MapRecord<String, Object, Object> copy = record.withId(RecordId.autoGenerate());
log.info("redis-消息队列,{},{},range-挂起消息>0>add+++++copy {},{}",key,group,copy.getStream(),copy);
//XADD stream-name id field value [field value]
RecordId add = redisTemplate.opsForStream().add(copy);
log.info("redis-消息队列,{},{},range-挂起消息>0>add+++++copy {},{}",key,group,add,copy);
log.info("redis-消息队列,{},{},range-挂起消息>0>ack^^^^^old {},{}",key,group,record.getStream(),record);
//XACK stream group id [id id ...]
Long siz = redisTemplate.opsForStream().acknowledge(GROUP, record);
log.info("redis-消息队列,{},{},range-挂起消息>0>ack^^^^^old [{}],{}",key,group,siz,record);
log.info("redis-消息队列,{},{},range-挂起消息>0>del-----old {},{}",key,group,record.getStream(),record);
//XDEL key ID [ID ...]
siz = redisTemplate.opsForStream().delete(record);
log.info("redis-消息队列,{},{},range-挂起消息>0>del-----old [{}],{}",key,group,siz,record);
}else{
log.info("redis-消息队列,{},{},range-挂起消息>0> {},{},{}",key,group,record.getStream(),id,record);
}
}
log.info("redis-消息队列,{},{},pending-挂起消息-end",key,group);
}
@Bean
@ConditionalOnProperty(name = "spring.redis.msg.listener",havingValue = "true")
public StreamMessageListenerContainer<String, ObjectRecord<String, String>> tspConsumerListener(
RedisConnectionFactory factory){
//spring-data-redis 2.3.1.RELEASE及更高版本中,createGroup如果不存在,则会自动使用创建流
return streamContainer(StreamOffset.create(STREAM, ReadOffset.lastConsumed()),Consumer.from(GROUP,CONSUMER)
,factory,new TspStreamListener());
}
private StreamMessageListenerContainer<String, ObjectRecord<String, String>> streamContainer(StreamOffset<String> offset
,Consumer consumer, RedisConnectionFactory factory, StreamListener<String, ObjectRecord<String, String>> listener) {
// pollTimeout 拉取消息超时时间,targetType 传递的数据类型, executor 线程池
StreamMessageListenerContainer<String, ObjectRecord<String, String>> container = StreamMessageListenerContainer
.create(factory, StreamMessageListenerContainer.StreamMessageListenerContainerOptions
.builder().pollTimeout(Duration.ofSeconds(50)).batchSize(1).targetType(String.class).build());
//指定消费最新的消息
try {
//创建消费者,接收上次处理未ACK消费的消息,指定消费者对象,autoAcknowledge 关闭自动ack确认
container.register(StreamMessageListenerContainer.StreamReadRequest
.builder(offset).errorHandler((error) -> log.error(error.getMessage()))
.cancelOnError(e -> false).consumer(consumer).autoAcknowledge(false)
.build(), listener);
} catch (Exception e) {
log.error(e.getMessage());
}
container.start();
return container;
}
/** 消费者 监听 */
public static class TspStreamListener implements StreamListener<String, ObjectRecord<String,String>> {
@Override
public void onMessage(ObjectRecord<String, String> message) {
RecordId messageId = message.getId();
// 消息的key和value
String string = message.getValue();
if("T".equals(string)){
log.info("消费者-监听>>>测试消息-ack. msgId={}, stream={}, body={}", messageId, message.getStream(), string);
redisTemplate.opsForStream().acknowledge(GROUP, message);
return;
}
log.info("消费者-监听>>>get msg. msgId={}, stream={}, body={}", messageId, message.getStream(), string);
//业务逻辑,需要考虑 被挂起的消息 重新消费情况
try {
log.info("消费者-监听>>>睡眠10s,模拟耗时逻辑-start");
TimeUnit.SECONDS.sleep(10);
log.info("消费者-监听>>>睡眠10s,模拟耗时逻辑-end");
} catch (InterruptedException e) {
e.printStackTrace();
}
log.info("消费者-监听>>>手动确认消息");
redisTemplate.opsForStream().acknowledge(GROUP, message);
log.info("消费者-监听>>>end");
}
}
}
redis
日志
2024-01-26 10:33:54.566 - [main] INFO o.s.b.w.s.c.ServletWebServerApplicationContext.prepareWebApplicationContext(292) - Root WebApplicationContext: initialization completed in 1162 ms
2024-01-26 10:33:58.388 - [main] INFO demo.RedisStreamConfig.handlerPending(85) - redis-消息队列,tsp,tsp-g-1,pending-挂起消息[3],minId : 1706178903044-0,maxId : 1706233950802-0
2024-01-26 10:33:58.396 - [main] INFO demo.RedisStreamConfig.handlerPending(88) - redis-消息队列,tsp,tsp-g-1,pending-挂起消息,PendingMessages{groupName='tsp-g-1', range=[--+], pendingMessages=[PendingMessage{id=1706178903044-0, consumer=tsp-g-1:tsp-c-1, elapsedTimeSinceLastDeliveryMS=57524430, totalDeliveryCount=1}, PendingMessage{id=1706233940011-0, consumer=tsp-g-1:tsp-c-1, elapsedTimeSinceLastDeliveryMS=2477103, totalDeliveryCount=1}, PendingMessage{id=1706233950802-0, consumer=tsp-g-1:tsp-c-1, elapsedTimeSinceLastDeliveryMS=1968963, totalDeliveryCount=1}]}
2024-01-26 10:33:58.397 - [main] INFO demo.RedisStreamConfig.handlerPending(93) - redis-消息队列,tsp,tsp-g-1,pending-挂起消息>>,PendingMessage{id=1706178903044-0, consumer=tsp-g-1:tsp-c-1, elapsedTimeSinceLastDeliveryMS=57524430, totalDeliveryCount=1}
2024-01-26 10:33:58.404 - [main] INFO demo.RedisStreamConfig.handlerPending(95) - redis-消息队列,tsp,tsp-g-1,range-挂起消息>> 1706178903044-0,[]
2024-01-26 10:33:58.405 - [main] INFO demo.RedisStreamConfig.handlerPending(93) - redis-消息队列,tsp,tsp-g-1,pending-挂起消息>>,PendingMessage{id=1706233940011-0, consumer=tsp-g-1:tsp-c-1, elapsedTimeSinceLastDeliveryMS=2477103, totalDeliveryCount=1}
2024-01-26 10:33:58.409 - [main] INFO demo.RedisStreamConfig.handlerPending(95) - redis-消息队列,tsp,tsp-g-1,range-挂起消息>> 1706233940011-0,[MapBackedRecord{recordId=1706233940011-0, kvMap={data=303}}]
2024-01-26 10:33:58.409 - [main] INFO demo.RedisStreamConfig.handlerPending(104) - redis-消息队列,tsp,tsp-g-1,range-挂起消息>0>add+++++copy tsp,MapBackedRecord{recordId=*, kvMap={data=303}}
2024-01-26 10:33:58.413 - [main] INFO demo.RedisStreamConfig.handlerPending(106) - redis-消息队列,tsp,tsp-g-1,range-挂起消息>0>add+++++copy 1706236438413-0,MapBackedRecord{recordId=*, kvMap={data=303}}
2024-01-26 10:33:58.413 - [main] INFO demo.RedisStreamConfig.handlerPending(107) - redis-消息队列,tsp,tsp-g-1,range-挂起消息>0>ack^^^^^old tsp,MapBackedRecord{recordId=1706233940011-0, kvMap={data=303}}
2024-01-26 10:33:58.418 - [main] INFO demo.RedisStreamConfig.handlerPending(109) - redis-消息队列,tsp,tsp-g-1,range-挂起消息>0>ack^^^^^old [1],MapBackedRecord{recordId=1706233940011-0, kvMap={data=303}}
2024-01-26 10:33:58.419 - [main] INFO demo.RedisStreamConfig.handlerPending(110) - redis-消息队列,tsp,tsp-g-1,range-挂起消息>0>del-----old tsp,MapBackedRecord{recordId=1706233940011-0, kvMap={data=303}}
2024-01-26 10:33:58.420 - [main] INFO demo.RedisStreamConfig.handlerPending(112) - redis-消息队列,tsp,tsp-g-1,range-挂起消息>0>del-----old [1],MapBackedRecord{recordId=1706233940011-0, kvMap={data=303}}
2024-01-26 10:33:58.421 - [main] INFO demo.RedisStreamConfig.handlerPending(93) - redis-消息队列,tsp,tsp-g-1,pending-挂起消息>>,PendingMessage{id=1706233950802-0, consumer=tsp-g-1:tsp-c-1, elapsedTimeSinceLastDeliveryMS=1968963, totalDeliveryCount=1}
2024-01-26 10:33:58.422 - [main] INFO demo.RedisStreamConfig.handlerPending(95) - redis-消息队列,tsp,tsp-g-1,range-挂起消息>> 1706233950802-0,[MapBackedRecord{recordId=1706233950802-0, kvMap={data=401}}]
2024-01-26 10:33:58.422 - [main] INFO demo.RedisStreamConfig.handlerPending(114) - redis-消息队列,tsp,tsp-g-1,range-挂起消息>0> tsp,1706233950802-0,MapBackedRecord{recordId=1706233950802-0, kvMap={data=401}}
2024-01-26 10:33:58.422 - [main] INFO demo.RedisStreamConfig.handlerPending(117) - redis-消息队列,tsp,tsp-g-1,pending-挂起消息-end