spring-boot redis stream消息队列demo-及死信简单处理

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: 8888servlet:context-path: /
spring:redis:msg:listener: falsehost: 127.0.0.1port: 6379database: 1client-type: lettucelettuce:pool:max-active: 8

代码结构

引入spring boot的 StringRedisTemplate

    private static StringRedisTemplate redisTemplate;@Autowiredpublic 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){// 创建消息记录, 以及指定streamStringRecord 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>> {@Overridepublic void onMessage(ObjectRecord<String, String> message) {RecordId messageId = message.getId();// 消息的key和valueString 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就先这样了。

本方法 是为了 解决，项目启动报错（无消息队列、无消费组）和程序重启死信问题(꒦_꒦)

    @PostConstructprivate 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 lenredisTemplate.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-0log.info("redis-消息队列,{},{},pending-挂起消息[{}],minId : {},maxId : {}",key,group,size,minId,maxId);// - + 4 ---------------------获取挂起所有信息 streamIdPendingMessages 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;@Autowiredpublic void setRedisTemplate(StringRedisTemplate redisTemplate) {RedisStreamConfig.redisTemplate = redisTemplate;}/** 生产者-发送消息 */public static void push(String msg){// 创建消息记录, 以及指定streamStringRecord 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);}/** 项目启动 预处理，无消息队列（消费者-启动报错）、限制容量、手动创建消费组、处理异常挂起的消息 */@PostConstructprivate void start(){// 生产者 创建 消息队列，防止启动 消费者 报错if(!(Boolean.TRUE.equals(redisTemplate.hasKey(STREAM)))){log.info("没有key,测试发送(创建key)");push("T");//限制 消息队列 容量 XTRIM stream MAXLEN lenredisTemplate.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-0log.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>> {@Overridepublic void onMessage(ObjectRecord<String, String> message) {RecordId messageId = message.getId();// 消息的key和valueString 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

仅供参考-请结合实际情况使用

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hqwc.cn/news/433976.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

Vue3中ElementPlus组件二次封装，实现原组件属性、插槽、事件监听、方法的透传

Vue3中ElementPlus组件二次封装，实现原组件属性、插槽、事件监听、方法的透传

本文以el-input组件为例&#xff0c;其它组件类似用法。 一、解决数据绑定问题 封装组件的第一步&#xff0c;要解决的就是数据绑定的问题&#xff0c;由于prop数据流是单向传递的&#xff0c;数据只能从父流向子&#xff0c;子想改父只能通过提交emit事件通知父修改。 父&a…
阅读更多...
scrapy的概念作用和工作流程

scrapy的概念作用和工作流程

1. scrapy的概念 Scrapy是一个Python编写的开源网络爬虫框架。它是一个被设计用于爬取网络数据、提取结构性数据的框架。 Scrapy 使用了Twisted[twɪstɪd]异步网络框架&#xff0c;可以加快我们的下载速度。 Scrapy文档地址&#xff1a;http://scrapy-chs.readthedocs.io/zh_…
阅读更多...
Linux中断 -- 中断路由、优先级、数据和标识

Linux中断 -- 中断路由、优先级、数据和标识

目录 1.中断路由 2.中断优先级 3.中断平衡 4.Linux内核中重要的数据结构 5.中断标识 承前文&#xff0c;本文从中断路由、优先级、数据结构和标识意义等方面对Linux内核中断进行一步的解析。 1.中断路由 Aset affinity flow GIC文中有提到SPI类型中断的路由控制器寄存器为…
阅读更多...
Prometheus 架构全面解析

Prometheus 架构全面解析

在本指南中&#xff0c;我们将详细介绍 Prometheus 架构。 Prometheus 是一个用 Golang 编写的开源监控和告警系统&#xff0c;能够收集和处理来自各种目标的指标。您还可以查询、查看、分析指标&#xff0c;并根据阈值收到警报。 此外&#xff0c;在当今世界&#xff0c;可观…
阅读更多...
风口抓猪-借助亚马逊云科技EC2服务器即刻构建PalWorld(幻兽帕鲁)私服~~~持续更新中

风口抓猪-借助亚马逊云科技EC2服务器即刻构建PalWorld(幻兽帕鲁)私服~~~持续更新中

Pocketpair出品的生存类游戏《幻兽帕鲁》最近非常火&#xff0c;最高在线人数已逼近200万。官方服务器亚历山大&#xff0c;游戏开发商也提供了搭建私人专用服务器的方案&#xff0c;既可以保证稳定的游戏体验&#xff0c;也可以和朋友一起联机游戏&#xff0c;而且还能自定义经…
阅读更多...
【PyTest】玩转HTML报告：修改、汉化和优化

【PyTest】玩转HTML报告：修改、汉化和优化

前言 Pytest框架可以使用两种测试报告&#xff0c;其中一种就是使用pytest-html插件生成的测试报告&#xff0c;但是报告中有一些信息没有什么用途或者显示的不太好看&#xff0c;还有一些我们想要在报告中展示的信息却没有&#xff0c;最近又有人问我pytest-html生成的报告&a…
阅读更多...
Ubuntu findfont: Font family ‘SimHei‘ not found.

Ubuntu findfont: Font family ‘SimHei‘ not found.

matplotlib中文乱码显示 当我们遇到这样奇怪的问题时, 结果往往很搞笑 尝试1不行 Stopping Jupyter Installing font-manager: sudo apt install font-manager Cleaning the matplotlib cache directory: rm ~/.cache/matplotlib -fr Restarting Jupyter. 尝试2 This work fo…
阅读更多...
vue实现甘特图

vue实现甘特图

目录 实现效果 一、安装依赖 二、使用 二、绕过license 实现效果 一、安装依赖 npm i --save vue-gantt-schedule-timeline-calendar 实现甘特图需先安装上述依赖&#xff0c;安装依赖实际上是通过gantt-schedule-timeline-calendar来实现的。所以node_module中因包含以下…
阅读更多...
openssl3.2 - 测试程序的学习 - test\sanitytest.c

openssl3.2 - 测试程序的学习 - test\sanitytest.c

文章目录 openssl3.2 - 测试程序的学习 - test\sanitytest.c概述笔记添加好实现的工程效果END openssl3.2 - 测试程序的学习 - test\sanitytest.c 概述 openssl3.2 - 测试程序的学习 重建工程参考 makefile, 将该加的实现加入vs2019工程, 不能无脑添加, 否则编译不过. 这个工…
阅读更多...
掌握可视化大屏：提升数据分析和决策能力的关键(上)

掌握可视化大屏：提升数据分析和决策能力的关键(上)

&#x1f90d; 前端开发工程师、技术日更博主、已过CET6 &#x1f368; 阿珊和她的猫_CSDN博客专家、23年度博客之星前端领域TOP1 &#x1f560; 牛客高级专题作者、打造专栏《前端面试必备》 、《2024面试高频手撕题》 &#x1f35a; 蓝桥云课签约作者、上架课程《Vue.js 和 E…
阅读更多...
Nginx，安装及基础配置

Nginx，安装及基础配置

目录 下载 Nginx Nginx 目录 配置文件 主要配置 events server 启动与关闭 方法一 方法二 方法三 查看是否启动成功 下载 Nginx 官网中文网 Nginx 目录 conf&#xff1a; 存放 核心配置文件 contrib&#xff1a; 存放 语法支持脚本 docs&#xff1a; …
阅读更多...
浪花 - 响应拦截器（强制登录）

浪花 - 响应拦截器（强制登录）

1. 配置响应拦截器 import axios from axios;const myAxios axios.create({baseURL: http://localhost:8080/api/, });myAxios.defaults.withCredentials true;// 请求拦截器 myAxios.interceptors.request.use(function (config) {// Do something before request is sentc…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023