使用 Redis Stream 结合 Redission DelayedQueue 实现延迟消息队列（已上线生产）

一、Redis Stream 介绍

Redis Stream 是 Redis 5.0 版本新增加的数据结构，作为一种轻量级的消息队列，Redis Stream 作为一种轻量级的消息队列，适合资源有限或对性能要求较高的场景。

Redis Stream 主要用于消息队列（MQ，Message Queue），Redis 本身是有一个 Redis 发布订阅 (pub/sub) 来实现消息队列的功能，但它有个缺点就是消息无法持久化，如果出现网络断开、Redis 宕机等，消息就会被丢弃。

简单来说发布订阅 (pub/sub) 可以分发消息，但无法记录历史消息。

而 Redis Stream 提供了消息的持久化和主备复制功能，可以让任何客户端访问任何时刻的数据，并且能记住每一个客户端的访问位置，还能保证消息不丢失。

Redis Stream 的结构如下所示，它有一个消息链表，将所有加入的消息都串起来，每个消息都有一个唯一的 ID 和对应的内容：

每个 Stream 都有唯一的名称，它就是 Redis 的 key，在我们首次使用 xadd 指令追加消息时自动创建。

上图解析：

• Consumer Group ：消费组，使用 XGROUP CREATE 命令创建，一个消费组有多个消费者(Consumer)。
• last_delivered_id ：游标，每个消费组会有个游标 last_delivered_id，任意一个消费者读取了消息都会使游标 last_delivered_id 往前移动。
• pending_ids ：消费者(Consumer)的状态变量，作用是维护消费者的未确认的 id。 pending_ids 记录了当前已经被客户端读取的消息，但是还没有 ack (Acknowledge character：确认字符）。

二、如何实现延迟消息队列

我的方案是使用Redis的阻塞队列和延迟队列将消息存进去，当延迟时间到达时，将消息发送至Redis Stream消息队列即可。

Redis延迟队列原理如下：

• Redisson延迟队列使用三个结构来存储，一个是queueName的list，值是添加的元素；一个是timeoutSetName的zset，值是添加的元素，score为timeout值；还有一个是getName()的blockingQueue，值是到期的元素。
• 

take 方法用于从延迟队列拿取消息，如果延迟时间未到是拿不到的，会一直阻塞线程，直到到达指定时间，便会返回消息数据。

    /*** 从目标队列中取出头部元素（为空时阻塞线程）** @param queueName 队列名* @return 队列头部元素*/public static String take(@NonNull String queueName) throws InterruptedException {RBlockingQueue<String> targetQueue = CLIENT.getBlockingQueue(queueName);CLIENT.getDelayedQueue(targetQueue);return targetQueue.take();}

监听延迟队列，时间到达时推送至Redis Stream消息队列：

    /*** 时间到期后将消息推到 stream 消息队列** @param streamKey 队列名称* @param content   消息内容*/@SuppressWarnings("unchecked")private static void pushToStreamAfterExpire(String streamKey, String content) {// 开启一个新线程处理延迟消息，防止阻塞主线程new Thread(() -> {while (true) {String messages = "";try {// 从延迟消息队列拿取消息messages = take(streamKey);} catch (InterruptedException e) {Thread.currentThread().interrupt();log.info("(线程被中断) Stream 键：{}", streamKey);}// 推送至streamMap<String, Object> messageMap = JSONUtil.toBean(messages, Map.class);RedisStreamUtil.push(RedisStreamUtil.BASE_STREAM_NAME + streamKey, messageMap);log.info("(发布延迟消息到 Redis Stream 成功) Stream 键：{}，消息值：{}", streamKey, content);}}, "redis_delay_queue").start();}

三、代码实战（Java）

1、对Redis Stream做配置，主要做了以下几件事：

• 初始化线程池
• 对监听器容器做初始化
• 监听器订阅消息主题和消费者组
• 对未成功消费的消息定时做重试补偿机制

package com.ricent.common.redis.stream.config;import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.text.CharSequenceUtil;
import cn.hutool.core.util.ObjectUtil;
import com.ricent.common.redis.stream.listener.BaseRedisStreamMessageListener;
import com.ricent.common.redis.stream.util.RedisStreamUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RStream;
import org.redisson.api.StreamGroup;
import org.redisson.api.StreamMessageId;
import org.redisson.api.stream.StreamAddArgs;
import org.redisson.api.stream.StreamCreateGroupArgs;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisConnectionFactory;
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.serializer.StringRedisSerializer;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.Subscription;import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;/*** redis stream消息队列配置** @author clq* @since 2024/11/11*/
@Configuration
@Slf4j
@RequiredArgsConstructor
public class RedisStreamConfig {private final List<StreamListener<String, MapRecord<String, String, String>>> listeners;private final List<BaseRedisStreamMessageListener> baseListeners;private final ConcurrentMap<String, Map<String, String>> streamGroups = new ConcurrentHashMap<>();private final RedisConnectionFactory redisConnectionFactory;private static final String BASE_STREAM_NAME = "redis_stream:";@Beanpublic StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer() {// 初始化线程池AtomicInteger index = new AtomicInteger(1);int processors = Runtime.getRuntime().availableProcessors();ThreadPoolExecutor executor = new ThreadPoolExecutor(processors, processors, 0, TimeUnit.SECONDS,new LinkedBlockingDeque<>(), r -> {Thread thread = new Thread(r);thread.setName("redis-stream-consumer-thread-" + index.getAndIncrement());thread.setDaemon(true);return thread;});// 初始化消息监听容器StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options =StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder().pollTimeout(Duration.ofSeconds(3)).batchSize(5).serializer(new StringRedisSerializer()).errorHandler(e -> log.error(e.getMessage(), e)).executor(executor).build();// 创建容器StreamMessageListenerContainer<String, MapRecord<String, String, String>> container =StreamMessageListenerContainer.create(redisConnectionFactory, options);container.start();log.info("Redis Stream Message Listener Container initialized and started");return container;}/*** 监听器配置* 消费者组名称为 stream名称_group* 消费者名称为 stream名称_consumer** @param container 监听容器* @return 所有监听者的订阅信息*/@Beanpublic List<Subscription> subscriptions(StreamMessageListenerContainer<String, MapRecord<String, String, String>> container) {return listeners.stream().map(listener -> {BaseRedisStreamMessageListener baseListener = (BaseRedisStreamMessageListener) listener;String streamName = CharSequenceUtil.toUnderlineCase(baseListener.getStreamName());String groupName = streamName + "_group";String consumerName = streamName + "_consumer";String finalStreamName = BASE_STREAM_NAME + streamName;// 初始化消息队列和消费者组initStreamAndGroup(finalStreamName, groupName);// 记录所有消费主题和消费者组及消费者名称streamGroups.computeIfAbsent(finalStreamName, k -> new ConcurrentHashMap<>()).put(groupName, consumerName);// 订阅消费者组log.info("为 stream: '{}' with group: '{}' and consumer: '{}' 创建订阅", finalStreamName, groupName, consumerName);return container.register(streamReadRequest(finalStreamName, groupName, consumerName), listener);}).toList();}/*** 自定义 stream 读取消息请求* (使用默认的 container.receive() 方法遇到异常或 redis 重启会取消订阅，导致监听器后续收不到消息)** @param streamName   主题名称* @param groupName    消费者组名称* @param consumerName 消费者名称* @return Request to read a Redis Stream.*/private StreamMessageListenerContainer.StreamReadRequest<String> streamReadRequest(String streamName, String groupName, String consumerName) {return StreamMessageListenerContainer.StreamReadRequest.builder(StreamOffset.create(streamName, ReadOffset.lastConsumed())).consumer(Consumer.from(groupName, consumerName)).cancelOnError(t -> {if (Boolean.FALSE.equals(RedisStreamUtil.hasKey(streamName))) {log.error("redis stream ：【消息主题：{} 被删除，已取消订阅！】", streamName);return true;}RedisConnection connection = redisConnectionFactory.getConnection();if (ObjectUtil.notEqual("PONG", connection.ping())) {log.error("----- redis 连接断开，请检查 redis 客户端状态！ -----");}return false;}).autoAcknowledge(false).build();}/*** 初始化 stream 和 创建消费者组** @param streamName 名称* @param groupName  消费者组名称*/private void initStreamAndGroup(String streamName, String groupName) {RStream<String, Object> stream = RedisStreamUtil.CLIENT.getStream(streamName);// 检查 stream 是否存在if (!stream.isExists()) {log.info("Creating stream: '{}'", streamName);// 创建主题StreamMessageId messageId = stream.add(StreamAddArgs.entries(Map.of("initialField", "initialValue")));// 创建消费组stream.createGroup(StreamCreateGroupArgs.name(groupName));// 删除初始化消息stream.remove(messageId);log.info("Stream: '{}' 和 group: '{}' 初始化完成.", streamName, groupName);}// 检查消费者组是否存在try {List<StreamGroup> existingGroups = stream.listGroups();boolean groupExists = existingGroups.stream().anyMatch(group -> group.getName().equals(groupName));if (!groupExists) {log.info("Creating consumer group: '{}' for stream: '{}'", groupName, streamName);stream.createGroup(StreamCreateGroupArgs.name(groupName));log.info("消费者组: '{}' for stream: '{}' 创建完成.", groupName, streamName);}} catch (Exception e) {log.error("Error while ensuring consumer group exists for stream: '{}'", streamName, e);}}/*** 每 15分钟进行一次心跳检测，并对 pending的任务进行处理*/@Bean(destroyMethod = "shutdown")public ScheduledExecutorService pendingMessagesChecker() {ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);Runnable pendingMessagesTask = () -> {try {for (Map.Entry<String, Map<String, String>> entry : streamGroups.entrySet()) {// 消息主题String streamName = entry.getKey();Map<String, String> groups = entry.getValue();for (Map.Entry<String, String> groupEntry : groups.entrySet()) {// 消费者组String groupName = groupEntry.getKey();// 消费者String consumerName = groupEntry.getValue();// 对 pending 的消息进行处理baseListeners.stream().filter(i -> streamName.equals(BASE_STREAM_NAME + i.getStreamName())).findFirst().ifPresent(messageListener -> {Map<StreamMessageId, Map<String, Object>> messageIdMap = RedisStreamUtil.fetchNotAckAndConsume(streamName, groupName, consumerName,60L, i -> messageListener.consume(streamName, i.getValue()));if (CollUtil.isNotEmpty(messageIdMap)) {log.info("主题 {} 已成功处理{}条 pending 消息！消息体为：【{}】", streamName, messageIdMap.size(), messageIdMap);}});}}} catch (Exception e) {log.error("Error while checking pending messages", e);}};// 第一次任务立即执行，之后每15分钟执行一次executorService.scheduleWithFixedDelay(pendingMessagesTask, 0, 15, TimeUnit.MINUTES);return executorService;}
}

2、配置基类监听器，基类监听器实现了 StreamListener 接口，并且定义了 consume 方法，该方法用于实现具体的业务逻辑，后续监听器只需要继承 BaseRedisStreamMessageListener 类，重写consume方法即可。

package com.ricent.common.redis.stream.listener;import com.ricent.common.redis.stream.util.RedisStreamUtil;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.StreamMessageId;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;import java.util.HashMap;
import java.util.Map;
import java.util.Objects;/*** redis stream 消息监听器基类** @author clq* @since 2024/11/11*/
@Data
@Slf4j
@Component
public abstract class BaseRedisStreamMessageListener implements StreamListener<String, MapRecord<String, String, String>> {protected String streamName;@Overridepublic void onMessage(MapRecord<String, String, String> message) {log.info("--- redis stream 接收到消息 ---  message :{}", message);try {// 消息消费Map<String, Object> objectMap = processMap(message.getValue());consume(streamName, objectMap);// 消息确认ackMessage(message);} catch (Exception e) {log.error("Error while consuming message from stream: '{}'. Message: {}", streamName, message, e);}}/*** 具体的消费逻辑** @param streamName 主题* @param message    消息体*/public abstract void consume(String streamName, Map<String, Object> message);/*** 消息确认机制** @param message 消息内容*/private void ackMessage(MapRecord<String, String, String> message) {RecordId id = message.getId();String finalStreamName = RedisStreamUtil.BASE_STREAM_NAME + streamName;String groupName = streamName + "_group";boolean isValidRecordId = Objects.nonNull(id.getTimestamp()) && Objects.nonNull(id.getSequence());if (isValidRecordId) {StreamMessageId messageId = new StreamMessageId(id.getTimestamp(), id.getSequence());Long ackCount = RedisStreamUtil.ack(finalStreamName, groupName, messageId);log.info("已确认 : {} 条消息，消息Id为 {}", ackCount, messageId);}}/*** 将 Map<String,String> 转换成 Map<String,Object>* 去除 value 反序列化后多余的引号** @param map 处理前消息体* @return 处理后的消息体*/public static Map<String, Object> processMap(Map<String, String> map) {Map<String, Object> resultMap = new HashMap<>();for (Map.Entry<String, String> entry : map.entrySet()) {String key = entry.getKey();String value = entry.getValue();// 检查value是否以两个引号开头和结尾if (value.startsWith("\"") && value.endsWith("\"")) {// 去除第一个和最后一个引号value = value.substring(1, value.length() - 1);}resultMap.put(key, value);}return resultMap;}
}

3、以下是两个工具类，分别是 RedisStreamUtil 和 RedisDelayQueueUtil

package com.ricent.common.redis.stream.util;import cn.hutool.core.text.CharSequenceUtil;
import cn.hutool.extra.spring.SpringUtil;
import lombok.experimental.UtilityClass;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.*;
import org.redisson.api.stream.StreamAddArgs;
import org.redisson.api.stream.StreamCreateGroupArgs;
import org.redisson.api.stream.StreamReadGroupArgs;
import org.redisson.client.codec.LongCodec;
import org.springframework.data.redis.connection.stream.MapRecord;import java.text.MessageFormat;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;/*** redis stream 消息队列工具类** @author clq* @since 2024/11/11*/
@Slf4j
@SuppressWarnings("unused")
@UtilityClass
public class RedisStreamUtil {public static final String BASE_STREAM_NAME = "redis_stream:";public static final RedissonClient CLIENT = SpringUtil.getBean(RedissonClient.class);/*** 向指定的 Redis Stream 中添加消息** @param streamName Stream 的名称* @param message    要添加的消息内容* @return 添加的消息的 ID*/public static StreamMessageId push(String streamName, Map<String, Object> message) {MapRecord<String, String, Object> streamRecord = MapRecord.create(streamName, message);RStream<String, Object> stream = CLIENT.getStream(streamName);return stream.add(StreamAddArgs.entries(streamRecord.getValue()));}/*** 向指定的 Redis Stream 中添加消息** @param message 要添加的消息内容* @return 添加的消息的 ID*/public static StreamMessageId push(MapRecord<String, String, Object> message) {RStream<String, Object> stream = CLIENT.getStream(message.getStream());return stream.add(StreamAddArgs.entries(message.getValue()));}/*** 从指定的 stream 中读取从未读取的消息* 使用默认消费者组：streamName_group* 使用默认消费者：streamName_consumer*/public static Map<StreamMessageId, Map<String, Object>> pull(String streamName) {String groupName = MessageFormat.format("{0}_group", CharSequenceUtil.toUnderlineCase(streamName));String consumerName = MessageFormat.format("{0}_consumer", CharSequenceUtil.toUnderlineCase(streamName));return pull(streamName, groupName, consumerName, null);}/*** 从指定的 stream 中读取从未读取的消息* 使用默认消费者组：streamName_group* 使用默认消费者：streamName_consumer*/public static Map<StreamMessageId, Map<String, Object>> pull(String streamName, Consumer<MapRecord<String, String, Object>> consumer) {String groupName = CharSequenceUtil.toUnderlineCase(streamName) + "_group";String consumerName = CharSequenceUtil.toUnderlineCase(streamName) + "_consumer";return pull(streamName, groupName, consumerName, consumer);}/*** 从指定的 stream 中读取从未读取的消息（可以指定消费者组和消费者）** @param streamName   Stream 的名称* @param groupName    消费者组名称* @param consumerName 消费者的名称* @return 轮询到的消息列表*/public static Map<StreamMessageId, Map<String, Object>> pull(String streamName, String groupName, String consumerName, Consumer<MapRecord<String, String, Object>> consumer) {// 获取 Redis 消息队列RStream<String, Object> redisMsgQueue = CLIENT.getStream(streamName);// 设置默认的消费者组和消费者名称groupName = CharSequenceUtil.isBlank(groupName) ? CharSequenceUtil.toUnderlineCase(streamName) + "_group" : groupName;consumerName = CharSequenceUtil.isBlank(consumerName) ? CharSequenceUtil.toUnderlineCase(streamName) + "_consumer" : consumerName;try {// 创建消费者组List<String> nameList = redisMsgQueue.listGroups().stream().map(StreamGroup::getName).toList();if (!nameList.contains(groupName)) {redisMsgQueue.createGroup(StreamCreateGroupArgs.name(groupName));}// 读取消息Map<StreamMessageId, Map<String, Object>> messages = redisMsgQueue.readGroup(groupName, consumerName, StreamReadGroupArgs.neverDelivered());// 处理消息for (Map.Entry<StreamMessageId, Map<String, Object>> e : messages.entrySet()) {StreamMessageId id = e.getKey();Map<String, Object> recordMap = e.getValue();MapRecord<String, String, Object> mapRecord = MapRecord.create(streamName, recordMap);if (consumer != null) {consumer.accept(mapRecord);redisMsgQueue.ack(groupName, id);}}return new HashMap<>(messages);} catch (Exception e) {log.error("Error while pulling messages from Redis stream", e);throw e;}}/*** 从指定的消费者组中获取未确认的消息并处理** @param streamName   Stream 的名称* @param groupName    消费者组名称* @param consumerName 消费者的名称* @param ackTimeout   确认超时时间（秒）* @param consumer     消息处理回调* @return 未确认的消息列表*/public static Map<StreamMessageId, Map<String, Object>> fetchNotAckAndConsume(String streamName, String groupName, String consumerName, long ackTimeout, Consumer<MapRecord<String, String, Object>> consumer) {Map<StreamMessageId, Map<String, Object>> output = Collections.emptyMap();// 获取消费者组的待处理消息信息RStream<String, Object> redisMsgQueue = CLIENT.getStream(streamName);PendingResult pr = redisMsgQueue.getPendingInfo(groupName);if (pr.getTotal() == 0) {return output;}// 从最低 ID 开始读取消息StreamMessageId curId = pr.getLowestId();StreamMessageId maxId = pr.getHighestId();while (true) {// 读取一批待处理消息List<PendingEntry> entries = redisMsgQueue.listPending(groupName, curId, maxId, 100);if (entries.isEmpty()) {return output;}// 更新当前读取的 IDcurId = makeExclusive(entries.get(entries.size() - 1).getId());// 声明超时的待处理消息StreamMessageId[] messageIds = entries.stream().map(PendingEntry::getId).toArray(StreamMessageId[]::new);Optional<Map<StreamMessageId, Map<String, Object>>> claimOptional = Optional.ofNullable(redisMsgQueue.claim(groupName, consumerName, ackTimeout, TimeUnit.SECONDS, messageIds));// 处理每条消息claimOptional.ifPresent(claim -> claim.forEach((id, message) -> {log.info("NotACKMessage ID: {}, Message: {}", id, message);MapRecord<String, String, Object> mapRecord = MapRecord.create(streamName, message);if (consumer != null) {consumer.accept(mapRecord);long ackCount = redisMsgQueue.ack(groupName, id);if (ackCount > 0) {log.info("Acked message ID: {}, Message: {}", id, message);}}}));}}/*** 判断 stream 是否存在** @param streamName Stream 的名称* @return 如果流存在则返回 true，否则 false*/public static Boolean hasKey(String streamName) {RBucket<String> bucket = CLIENT.getBucket(streamName);return bucket.isExists();}/*** 确认消费** @param streamName Stream 的名称* @param groupName  消费者组* @param recordIds  消息id* @return java.lang.Long*/public static Long ack(String streamName, String groupName, StreamMessageId... recordIds) {RStream<String, Object> stream = CLIENT.getStream(streamName);return stream.ack(groupName, recordIds);}/*** 删除消费队列，及其消费者组** @param streamName Stream 的名称* @return 是否成功删除*/public static <K, V> Boolean delete(String streamName) {RStream<K, V> stream = CLIENT.getStream(streamName);return stream.delete();}/*** 移除消费者组的消息id** @param streamName Stream 的名称* @param ids        消息 id 列表* @return 是否成功删除*/public static <K, V> Long remove(String streamName, StreamMessageId... ids) {RStream<K, V> stream = CLIENT.getStream(streamName);return stream.remove(ids);}/*** 生成一个唯一的 StreamMessageId** @param id 原始的 StreamMessageId* @return 生成的唯一的 StreamMessageId*/private static StreamMessageId makeExclusive(StreamMessageId id) {return new StreamMessageId(id.getId0(), id.getId1() + 1);}/*** 根据指定的 lastSize 缩减 Redis Stream 的大小** @param streamName       stream 的名称* @param lastSize         缩减后的大小* @param queueSizeMapName 用于记录缩减后的大小的哈希表名称* @return 缩减后的大小*/public static long shrinkByLastSize(String streamName, long lastSize, String queueSizeMapName) {// lua 脚本String shrinkScript = """local nowSz = redis.call('XLEN', KEYS[1]);redis.call('XTRIM', KEYS[1], 'MAXLEN', nowSz - tonumber(ARGV[1]));local leftSz = redis.call('XLEN', KEYS[1]);redis.call('HSET', KEYS[2], KEYS[1], leftSz);return leftSz;""";RScript rScript = CLIENT.getScript(LongCodec.INSTANCE);return rScript.eval(RScript.Mode.READ_WRITE, shrinkScript, RScript.ReturnType.INTEGER,Arrays.asList(streamName, queueSizeMapName), lastSize);}}

package com.ricent.common.redis.stream.util;import cn.hutool.core.collection.CollUtil;
import cn.hutool.extra.spring.SpringUtil;
import cn.hutool.json.JSONUtil;
import lombok.NonNull;
import lombok.experimental.UtilityClass;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;import java.time.Duration;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;/*** 延迟队列工具类** @author clq* @since 2024-11-11*/
@Slf4j
@UtilityClass
@SuppressWarnings("unused")
public class RedisDelayQueueUtil {public static final RedissonClient CLIENT = SpringUtil.getBean(RedissonClient.class);/*** 从目标队列中取出头部元素（为空时返回 null）** @param queueName 队列名称* @return 队列头部元素*/public static String poll(@NonNull String queueName) {RBlockingQueue<String> targetQueue = CLIENT.getBlockingQueue(queueName);CLIENT.getDelayedQueue(targetQueue);return targetQueue.poll();}/*** 从目标队列中取出头部元素（为空时阻塞线程）** @param queueName 队列名* @return 队列头部元素*/public static String take(@NonNull String queueName) throws InterruptedException {RBlockingQueue<String> targetQueue = CLIENT.getBlockingQueue(queueName);CLIENT.getDelayedQueue(targetQueue);return targetQueue.take();}/*** 从目标队列中获取头部元素（仅获取头部元素而不删除，为空时返回 null）** @param queueName 队列名* @return 队列头部元素*/public static String peek(@NonNull String queueName) {RBlockingQueue<String> targetQueue = CLIENT.getBlockingQueue(queueName);CLIENT.getDelayedQueue(targetQueue);return targetQueue.peek();}/*** 添加延迟队列** @param queueName 队列名称* @param element   队列元素* @param endTime   延迟至指定时间*/public static List<String> addDelayQueue(String queueName, String element, LocalDateTime endTime) {long seconds = Duration.between(LocalDateTime.now(), endTime).getSeconds();if (seconds > 0) {return addDelayQueue(queueName, element, seconds, TimeUnit.SECONDS);}throw new IllegalArgumentException("延迟时间必须大于当前时间！");}/*** 添加延迟队列** @param queueName 队列名称* @param element   队列元素* @param delay     延迟时间* @param timeUnit  时间单位*/public static List<String> addDelayQueue(String queueName, String element, long delay, TimeUnit timeUnit) {try {RBlockingQueue<String> blockingDeque = CLIENT.getBlockingQueue(queueName);RDelayedQueue<String> delayedQueue = CLIENT.getDelayedQueue(blockingDeque);delayedQueue.offer(element, delay, timeUnit);log.info("(添加延时队列成功) 队列名称：{}，队列元素：{}，延迟时间：{}", queueName, element, timeUnit.toSeconds(delay) + "秒");return delayedQueue.readAll();} catch (Exception e) {log.error("(添加延时队列失败) {}", e.getMessage());throw e;}}/*** 删除延迟队列中的消息** @param queueName 队列名称* @param element   队列元素* @return 是否删除成功*/public static boolean removeDelayedQueue(@NonNull String queueName, @NonNull String element) {if (StringUtils.isBlank(queueName)) {return false;}try {RBlockingDeque<String> blockingDeque = CLIENT.getBlockingDeque(queueName);RDelayedQueue<String> delayedQueue = CLIENT.getDelayedQueue(blockingDeque);boolean flag = delayedQueue.remove(element);if (flag) {log.info("(删除延时队列保证唯一性) 队列元素：{}，队列元素：{}", queueName, element);}delayedQueue.destroy();return flag;} catch (Exception e) {log.error("(删除延时队列异常) 队列名称：{}，队列元素：{},错误信息：{}", queueName, element, e.getMessage());throw e;}}/*** 将延迟队列中的消息发布到 Redis Stream** @param value     消息值* @param streamKey Stream 键* @param delay     延迟时间* @param timeUnit  时间单位*/public static void publishToStream(String streamKey, String value, long delay, TimeUnit timeUnit) {// 将消息添加至延迟队列List<String> list = addDelayQueue(streamKey, value, delay, timeUnit);// 到期后推送至 stream 队列if (CollUtil.isNotEmpty(list)) {pushToStreamAfterExpire(streamKey, value);}}/*** 将延迟队列中的消息发布到 Redis Stream** @param streamKey Stream 键* @param content   消息内容* @param endTime   延迟消费的时间*/public static void publishToStream(String streamKey, String content, LocalDateTime endTime) {// 将消息添加至延迟队列List<String> list = addDelayQueue(streamKey, content, endTime);if (CollUtil.isNotEmpty(list)) {// 到期后推送至 stream 队列pushToStreamAfterExpire(streamKey, content);}}/*** 时间到期后将消息推到 stream 消息队列** @param streamKey 队列名称* @param content   消息内容*/@SuppressWarnings("unchecked")private static void pushToStreamAfterExpire(String streamKey, String content) {// 开启一个新线程处理延迟消息，防止阻塞主线程new Thread(() -> {while (true) {String messages = "";try {// 从延迟消息队列拿取消息messages = take(streamKey);} catch (InterruptedException e) {Thread.currentThread().interrupt();log.info("(线程被中断) Stream 键：{}", streamKey);}// 推送至streamMap<String, Object> messageMap = JSONUtil.toBean(messages, Map.class);RedisStreamUtil.push(RedisStreamUtil.BASE_STREAM_NAME + streamKey, messageMap);log.info("(发布延迟消息到 Redis Stream 成功) Stream 键：{}，消息值：{}", streamKey, content);}}, "redis_delay_queue").start();}
}

4、最后在我们的消息模块定义 RedisEventBusUtils ，后续通过该工具类发布订阅消息。

package com.ricent.common.eventbus.utils;import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.text.CharSequenceUtil;
import cn.hutool.json.JSONUtil;
import com.ricent.common.core.exception.ServiceException;
import com.ricent.common.redis.stream.util.RedisDelayQueueUtil;
import com.ricent.common.redis.stream.util.RedisStreamUtil;
import lombok.experimental.UtilityClass;
import lombok.extern.slf4j.Slf4j;import java.time.LocalDateTime;
import java.util.Map;
import java.util.concurrent.TimeUnit;/*** Redis Stream 消息工具类** @author clq* @since 2024/11/11*/
@Slf4j
@UtilityClass
public class RedisEventBusUtils {/*** 发送 Redis 消息事件** @param streamName 消息主题* @param messages   消息体*/public static void publishEvent(String streamName, Map<String, Object> messages) {validateParam(streamName, messages);RedisStreamUtil.push(RedisStreamUtil.BASE_STREAM_NAME + streamName, messages);}/*** 发送 Redis 延迟消息事件** @param streamName 消息主题* @param messages   消息体* @param delay      延迟时间* @param timeUnit   时间单位*/public static void publishEvent(String streamName, Map<String, Object> messages, Long delay, TimeUnit timeUnit) {validateParam(streamName, messages);RedisDelayQueueUtil.publishToStream(streamName, JSONUtil.toJsonStr(messages), delay, timeUnit);}/*** 发送 Redis 延迟消息事件** @param streamName 消息主题* @param messages   消息体* @param endTime    到期时间*/public static void publishEvent(String streamName, Map<String, Object> messages, LocalDateTime endTime) {validateParam(streamName, messages);RedisDelayQueueUtil.publishToStream(streamName, JSONUtil.toJsonStr(messages), endTime);}/*** 校验参数** @param streamName 消息主题* @param messages   消息体*/private static void validateParam(String streamName, Map<String, Object> messages) {if (CharSequenceUtil.isBlank(streamName)) {throw new ServiceException("streamName is blank：消息主题不能为空！");}if (CollUtil.isEmpty(messages)) {throw new ServiceException("messages is empty：消息体不能为空！");}}
}

5、至此，所有的代码配置工作已完成，接下来编写消费者实现业务逻辑和然后在实际业务需要的地方发布消息即可。

以下是一个使用 demo

消费者（继承BaseRedisStreamMessageListener类，重写consume方法，编写实际业务逻辑即可）

package com.ricent.meetingtask.consume;import com.ricent.common.redis.stream.listener.BaseRedisStreamMessageListener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;import java.util.Map;/*** 消息监听器** @author clq* @since 2024/11/11*/
@Slf4j
@Component
public class MeetingTaskListener extends BaseRedisStreamMessageListener {public MeetingTaskListener() {this.streamName = "demo_task";}@Overridepublic void consume(String streamName, Map<String, Object> message) {log.info("--- 消费消息 --- stream :{} message :{}", streamName, message);// 实际业务逻辑}
}

生产者（在需要的地方发送订阅消息即可，注意Stream名称一定要一致）

RedisEventBusUtils.publishEvent("demo_task", message,30L, TimeUnit.MINUTES);

四、总结

1）Redis Stream相比于它原生的消息订阅方式，具有以下的优势：

• 支持范围查找：内置的索引功能，可以通过索引来对消息进行范围查找
• 支持阻塞操作：避免低效的反复轮询查找消息
• 支持 ACK：可以通过确认机制来告知已经成功处理了消息，保证可靠性
• 支持多个消费者：多个消费者可以同时消费同一个流，Redis 会确保每个消费者都可以独立地消费流中的消息
• 支持消息持久化： Redis Stream 提供了消息的持久化和主备复制功能，可以让任何客户端访问任何时刻的数据，并且能记住每一个客户端的访问位置，还能保证消息不丢失。

2）但是对比专门的MQ消息队列，功能还是相对简单

特性/功能	Redis Stream	专业 MQ 系统 (如 Kafka、RabbitMQ、RocketMQ)
持久化	支持 RDB 和 AOF	更强大的持久化和存储机制
主备复制	支持	更复杂的分布式复制和集群管理
消息确认（ACK）	支持	更完善的消息确认和处理机制
功能丰富度	相对简单	更多高级功能（如消息回溯、死信队列等）
集群管理和扩展性	简单	更强大的集群管理和扩展能力
资源消耗	适合小到中等规模	更适合大规模、高并发场景
社区支持和生态系统	相对较小	更丰富的社区支持和生态系统

 3）我们结合了 Redis 的延迟队列和 Redis Stream的消息队列实现了延迟消息队列，但是注意它具有以下优缺点：

优点：轻便，继承了Redis的高性能，并且占用资源低；

缺点：适用于延迟时间确定的场景，如果延迟时间一直变动，需要额外的逻辑来动态调整延迟时间，会增加系统的复杂性和维护难度。

4）适用场景

• 定时任务调度：适合需要定时执行的任务，定时时间在任务发布时就已经确定。
• 固定延迟的消息处理：适合延迟时间固定且不易变动的场景，如订单取消、优惠券过期等。
• 轻量级应用：适合资源有限、性能要求高的小到中等规模的应用场景。

ps：以下是我整理的java面试资料，感兴趣的可以看看。最后，创作不易，觉得写得不错的可以点点关注！

链接：https://www.yuque.com/u39298356/uu4hxh?# 《Java知识宝典》