文章目录
- 1. RedisStreamConfig
- 2. 消费者`MyMessageListener`
- 3. RedisStreamUtil
- 4. RedisStreamConstant
- 5. 测试
- 6. 处理消费者已读取未ack的消息
redis stream文档参考 https://zhuanlan.zhihu.com/p/60501638
1. RedisStreamConfig
package com.tophant.eventdemo.common.config;
import cn.hutool.core.collection.ListUtil;
import com.tophant.eventdemo.common.contstant.RedisStreamConstant;
import com.tophant.eventdemo.listener.MyMessageListener;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
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.stream.Consumer;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import java.util.List;
/**
* redis stream配置
*
* @author wanfei
* @date 2023/03/28
*/
@Slf4j
@Configuration
@RequiredArgsConstructor
public class RedisStreamConfig {
private final RedisStreamUtil redisStreamUtil;
/**
* 创建 Redis Stream 集群消费的容器（注册一个消费者类作为多个消费者）
* <p>
* Redis Stream 的 xreadgroup 命令：https://www.geek-book.com/src/docs/redis/redis/redis.io/commands/xreadgroup.html
*/
@Bean(initMethod = "start", destroyMethod = "stop")
public StreamMessageListenerContainer<String, ObjectRecord<String, String>> redisStreamMessageListenerContainer(
RedisConnectionFactory connectionFactory) {
// 第一步，创建 StreamMessageListenerContainer 容器
// 创建 options 配置
StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, ObjectRecord<String, String>> containerOptions =
StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()
.batchSize(10) // 一次性最多拉取多少条消息
// .executor(Executors.newFixedThreadPool(10)) // 线程池 ---没起作用
.targetType(String.class) // 目标类型。统一使用 String
.build();
// 创建 container 对象
StreamMessageListenerContainer<String, ObjectRecord<String, String>> container =
StreamMessageListenerContainer.create(connectionFactory, containerOptions);
MyMessageListener myMessageListener1 = new MyMessageListener(redisStreamUtil);
MyMessageListener myMessageListener2 = new MyMessageListener(redisStreamUtil);
MyMessageListener myMessageListener3 = new MyMessageListener(redisStreamUtil);
List<StreamListener> listeners = ListUtil.of(
myMessageListener1,
myMessageListener2,
myMessageListener3
);
for (int i = 0; i < listeners.size(); i++) {
StreamListener listener = listeners.get(i);
// 消费者名称
String consumeName = RedisStreamConstant.CONSUMER_NAME + i;
// 创建 listener 对应的消费者分组
try {
// 注意：key是stream的key
redisStreamUtil.createGroup(RedisStreamConstant.STREAM_NAME, RedisStreamConstant.GROUP_NAME);
} catch (Exception ignore) {
}
// 创建 Consumer 对象
Consumer consumer = Consumer.from(RedisStreamConstant.GROUP_NAME, consumeName);
// 设置 Consumer 消费进度，以最小消费进度为准
StreamOffset<String> streamOffset = StreamOffset.create(RedisStreamConstant.STREAM_NAME, ReadOffset.lastConsumed());
// 设置 Consumer 监听
StreamMessageListenerContainer.StreamReadRequestBuilder<String> builder = StreamMessageListenerContainer.StreamReadRequest
.builder(streamOffset).consumer(consumer)
.autoAcknowledge(false) // 不自动 ack
.cancelOnError(throwable -> false); // 默认配置，发生异常就取消消费，显然不符合预期；因此，我们设置为 false
container.register(builder.build(), listener);
log.info("[redisStreamMessageListenerContainer][完成注册 StreamKey({}) 对应的监听器({})]",
RedisStreamConstant.STREAM_NAME, listener.getClass().getName());
}
return container;
}
}
.executor()：设置线程池发现没有用，不能一个消费者多线程消费消息，后来试了new 多个消费者，发现可以达到一样的效果
2. 消费者MyMessageListener
package com.tophant.eventdemo.listener;
import cn.hutool.core.thread.ThreadUtil;
import com.tophant.eventdemo.common.config.RedisStreamUtil;
import com.tophant.eventdemo.contstant.RedisStreamConstant;
import lombok.RequiredArgsConstructor;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;
import java.util.concurrent.TimeUnit;
/**
* 我消息监听器
*
* @author wanfei
* @date 2023/03/28
*/
@Component
@RequiredArgsConstructor
public class MyMessageListener implements StreamListener<String, ObjectRecord<String, String>> {
private final RedisStreamUtil redisStreamUtil;
@Override
public void onMessage(ObjectRecord<String, String> message) {
System.out.println("111 Thread: " + Thread.currentThread().getName() + ", Message: " + message.getValue());
// 消息执行完毕，没有报错，ack确认
redisStreamUtil.ack(message.getStream(), RedisStreamConstant.GROUP_NAME, message.getId().getValue());
// ack之后删除消息
redisStreamUtil.del(message.getStream(), message.getId().getValue());
ThreadUtil.sleep(5, TimeUnit.SECONDS);
}
}
3. RedisStreamUtil
package com.tophant.eventdemo.common.config;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Range;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import java.util.List;
import java.util.stream.Collectors;
/**
* redis stream工具类
*
* @author wanfei
* @date 2023/03/22
*/
@Component
public class RedisStreamUtil {
@Autowired
private StringRedisTemplate stringRedisTemplate;
/**
* 创建消费者组
*
* @param key 关键
* @param group 集团
* @return {@link String}
*/
public String createGroup(String key, String group) {
return stringRedisTemplate.opsForStream().createGroup(key, group);
}
/**
* 查询消费者组信息
*
* @param key 关键
* @param group 集团
* @return {@link StreamInfo.XInfoConsumers}
*/
public StreamInfo.XInfoConsumers consumers(String key, String group) {
return stringRedisTemplate.opsForStream().consumers(key, group);
}
/**
* 确认已消费
*
* @param key 关键
* @param group 集团
* @param recordIds 记录id
* @return {@link Long}
*/
public Long ack(String key, String group, String... recordIds) {
return stringRedisTemplate.opsForStream().acknowledge(key, group, recordIds);
}
/**
* 发送消息
*
* @param record 记录
* @return {@link String}
*/
public String add(Record record) {
return stringRedisTemplate.opsForStream().add(record).getValue();
}
/**
* 删除消息，这里的删除仅仅是设置了标志位，不影响消息总长度
* 消息存储在stream的节点下，删除时仅对消息做删除标记，当一个节点下的所有条目都被标记为删除时，销毁节点
*
* @param key 关键
* @param recordIds 记录id
* @return {@link Long}
*/
public Long del(String key, String... recordIds) {
return stringRedisTemplate.opsForStream().delete(key, recordIds);
}
/**
* 查询队列中pending消息(已读取ack未确认)
*
* @param key stream key
* @param group 消费者组
* @return {@link List}<{@link PendingMessage}>
*/
public List<PendingMessage> pending(String key, String group) {
PendingMessages pending = stringRedisTemplate.opsForStream().pending(key, group, Range.unbounded(), Long.MAX_VALUE);
return pending.stream().collect(Collectors.toList());
}
/**
* 从消费者中读取消息
*
* @param key 关键
* @param group 集团
* @param name 名字
* @return {@link List}<{@link ObjectRecord}<{@link String}, {@link String}>>
*/
public List<ObjectRecord<String, String>> read(String key, String group, String name) {
return stringRedisTemplate.opsForStream().read(String.class, Consumer.from(group, name), StreamOffset.create(key, ReadOffset.from("0")));
}
}
4. RedisStreamConstant
package com.tophant.eventdemo.contstant;
/**
* redis stream常量
*
* @author wanfei
* @date 2023/03/28
*/
public interface RedisStreamConstant {
/**
* 消费者名称
*/
String CONSUMER_NAME = "my-consumer";
/**
* 消费者组名称
*/
String GROUP_NAME = "my-group";
/**
* 队列名称
*/
String STREAM_NAME = "my-stream";
}
5. 测试
package com.tophant.eventdemo.controller;
import cn.hutool.core.lang.Dict;
import com.tophant.eventdemo.common.config.RedisStreamUtil;
import com.tophant.eventdemo.contstant.RedisStreamConstant;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.stream.StreamRecords;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
/**
* @author Huhailong
* @Description
* @Date 2021/3/10.
*/
@RestController
@RequestMapping("/test")
@Api(tags = "redis-stream测试")
@Slf4j
@RequiredArgsConstructor
public class TestController {
private final RedisStreamUtil redisStreamUtil;
@GetMapping("/moreTest/{count}")
@ApiOperation("发送消息测试 count是循环次数")
public String moreAddTest(@PathVariable("count") Integer count){
for(int i=0; i<count; i++) {
redisStreamUtil.add(StreamRecords.newRecord()
.ofObject(new Dict().set("name", "张三" + i))
.withStreamKey(RedisStreamConstant.STREAM_NAME));
}
return "发送成功";
}
}
6. 处理消费者已读取未ack的消息
package com.tophant.eventdemo.common.job;
import com.tophant.eventdemo.common.config.RedisStreamUtil;
import com.tophant.eventdemo.common.contstant.RedisStreamConstant;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.PendingMessage;
import org.springframework.data.redis.connection.stream.StreamRecords;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;
/**
* 这个任务用于处理消费者已读取未ack的消息
*/
@Slf4j
@RequiredArgsConstructor
@Component
public class RedisPendingMessageResendJob {
private final RedisStreamUtil redisStreamUtil;
/**
* 一分钟执行一次,这里选择每分钟的35秒执行，是为了避免整点任务过多的问题
*/
@Scheduled(cron = "35 * * * * ?")
public void messageResend() {
try {
execute();
} catch (Exception ex) {
log.error("[messageResend][执行异常]", ex);
}
}
private void execute() {
// 直接从消费者组队列中查询未确认消息
Map<String, PendingMessage> pendingMessageMap = redisStreamUtil.pending(RedisStreamConstant.STREAM_NAME, RedisStreamConstant.GROUP_NAME)
.stream()
// 当未确认消息时间超过2分钟才重新投递消息，防止正在处理的消息被更新投递
.filter(e -> e.getElapsedTimeSinceLastDelivery().getSeconds() > 120)
.collect(Collectors.toMap(e -> e.getId().getValue(), Function.identity()));
// 一个消费者组可能有多个消费者
List<String> consumerNameList = pendingMessageMap.values().stream()
.map(PendingMessage::getConsumerName).distinct().collect(Collectors.toList());
// 查询多个消费者的所有消息-过滤-只返回未确认消息
List<ObjectRecord<String, String>> pendingRecords = consumerNameList.stream()
.flatMap(consumerName ->
redisStreamUtil.read(RedisStreamConstant.STREAM_NAME, RedisStreamConstant.GROUP_NAME, consumerName).stream()
// 只取 pending 消息
.filter(e -> pendingMessageMap.containsKey(e.getId().getValue())))
.collect(Collectors.toList());
pendingRecords.forEach(record -> {
// 重新投递消息
redisStreamUtil.add(StreamRecords.newRecord()
// 设置内容
.ofObject(record.getValue())
// 设置队列
.withStreamKey(RedisStreamConstant.STREAM_NAME));
// 老的消息ack确认
redisStreamUtil.ack(RedisStreamConstant.STREAM_NAME, RedisStreamConstant.GROUP_NAME, record.getId().getValue());
// 老的消息删除
redisStreamUtil.del(RedisStreamConstant.STREAM_NAME, record.getId().getValue());
// 已读取次数重新投递不会 +1 ,这里都是1
log.info("【RedisPendingMessageResendJob】 消息 recordId: {} 已读取未ack时间: {} 秒 已读取次数: {} 重新投递", record.getId().getValue(),
pendingMessageMap.get(record.getId().getValue()).getElapsedTimeSinceLastDelivery().getSeconds(),
pendingMessageMap.get(record.getId().getValue()).getTotalDeliveryCount());
});
}
}
