redis_stream__0">1.redis stream 特点

1.支持消息持久化
2.消费者组模式
3.消息确认机制
4. 消息重试机制
5. 死信队列

2. 消息生产者服务

2.1 如下代码

java">@Service
@Slf4j
public class StreamMessageProducer {@Autowiredprivate StringRedisTemplate redisTemplate;private static final String STREAM_KEY = "message:stream";/*** 发送消息*/public String sendMessage(String topic, Object message) {try {StringRecord record = StreamRecords.string(Collections.singletonMap("message", JSON.toJSONString(message))).withStreamKey(STREAM_KEY + ":" + topic);RecordId recordId = redisTemplate.opsForStream().add(record);log.info("消息发送成功: topic={}, messageId={}", topic, recordId);return recordId.getValue();} catch (Exception e) {log.error("消息发送失败: topic={}, message={}", topic, message, e);throw new RuntimeException("消息发送失败", e);}}/*** 批量发送消息*/public List<String> sendMessages(String topic, List<Object> messages) {try {List<MapRecord<String, String, String>> records = messages.stream().map(msg -> StreamRecords.string(Collections.singletonMap("message", JSON.toJSONString(msg))).withStreamKey(STREAM_KEY + ":" + topic)).collect(Collectors.toList());List<String> messageIds = new ArrayList<>();for (MapRecord<String, String, String> record : records) {RecordId recordId = redisTemplate.opsForStream().add(record);messageIds.add(recordId.getValue());}log.info("批量消息发送成功: topic={}, count={}", topic, messageIds.size());return messageIds;} catch (Exception e) {log.error("批量消息发送失败: topic={}", topic, e);throw new RuntimeException("批量消息发送失败", e);}}
}

3.消息消费者服务（多线程消费）

3.1 代码如下

java">@Service
@Slf4j
public class StreamMessageConsumer {@Autowiredprivate StringRedisTemplate redisTemplate;private static final String STREAM_KEY = "message:stream";private final Map<String, StreamMessageHandler<?>> handlers = new ConcurrentHashMap<>();/*** 注册消息处理器*/public <T> void registerHandler(String topic, Class<T> messageType, Consumer<T> handler) {handlers.put(topic, new StreamMessageHandler<>(messageType, handler));}/*** 启动消费*/@PostConstructpublic void startConsuming() {for (String topic : handlers.keySet()) {String streamKey = STREAM_KEY + ":" + topic;String consumerGroup = "group:" + topic;String consumerName = "consumer:" + UUID.randomUUID().toString();try {// 创建消费者组（如果不存在）createConsumerGroupIfNotExists(streamKey, consumerGroup);// 启动消费线程Thread consumerThread = new Thread(() -> consumeMessages(streamKey, consumerGroup, consumerName, topic));consumerThread.setName("stream-consumer-" + topic);consumerThread.start();} catch (Exception e) {log.error("启动消费者失败: topic={}", topic, e);}}}private void createConsumerGroupIfNotExists(String streamKey, String groupName) {try {redisTemplate.opsForStream().createGroup(streamKey, groupName);} catch (Exception e) {// 组已存在，忽略异常log.debug("Consumer group already exists: {}", groupName);}}private void consumeMessages(String streamKey, String group, String consumer, String topic) {StreamMessageHandler<?> handler = handlers.get(topic);while (!Thread.currentThread().isInterrupted()) {try {// 读取消息List<MapRecord<String, String, String>> records = redisTemplate.opsForStream().read(Consumer.from(group, consumer),StreamReadOptions.empty().count(10).block(Duration.ofSeconds(1)),StreamOffset.create(streamKey, ReadOffset.lastConsumed()));if (records != null && !records.isEmpty()) {for (MapRecord<String, String, String> record : records) {try {// 处理消息processMessage(record, handler);// 确认消息redisTemplate.opsForStream().acknowledge(streamKey, group, record.getId());} catch (Exception e) {log.error("消息处理失败: messageId={}", record.getId(), e);}}}} catch (Exception e) {log.error("消息消费异常: topic={}", topic, e);try {Thread.sleep(1000);} catch (InterruptedException ie) {Thread.currentThread().interrupt();break;}}}}private <T> void processMessage(MapRecord<String, String, String> record, StreamMessageHandler<T> handler) {try {String messageJson = record.getValue().get("message");T message = JSON.parseObject(messageJson, handler.getMessageType());handler.getHandler().accept(message);} catch (Exception e) {log.error("消息处理失败: {}", record, e);throw e;}}
}@Data
@AllArgsConstructor
class StreamMessageHandler<T> {private Class<T> messageType;private Consumer<T> handler;
}

4.消息重试服务

4.1 代码如下

java">@Service
@Slf4j
public class StreamMessageRetryService {@Autowiredprivate StringRedisTemplate redisTemplate;private static final String STREAM_KEY = "message:stream";private static final int MAX_RETRY_COUNT = 3;/*** 处理待处理的消息*/@Scheduled(fixedDelay = 60000) // 每分钟执行一次public void processPendingMessages() {for (String topic : getTopics()) {String streamKey = STREAM_KEY + ":" + topic;String groupName = "group:" + topic;try {// 获取待处理的消息PendingMessages pending = redisTemplate.opsForStream().pending(streamKey, groupName, Range.unbounded(), 100);if (pending != null) {for (PendingMessage message : pending.getPendingMessages()) {processRetry(streamKey, groupName, message);}}} catch (Exception e) {log.error("处理待处理消息失败: topic={}", topic, e);}}}private void processRetry(String streamKey, String groupName, PendingMessage message) {try {if (message.getTotalDeliveryCount() > MAX_RETRY_COUNT) {// 超过重试次数，移动到死信队列moveToDeadLetter(streamKey, groupName, message.getIdAsString());} else {// 重新投递消息redisTemplate.opsForStream().claim(streamKey, groupName, "retry-consumer", Duration.ofMinutes(1), message.getIdAsString());}} catch (Exception e) {log.error("处理重试消息失败: messageId={}", message.getIdAsString(), e);}}private void moveToDeadLetter(String streamKey, String groupName, String messageId) {try {// 读取消息内容List<MapRecord<String, String, String>> messages = redisTemplate.opsForStream().range(streamKey, Range.closed(messageId, messageId));if (messages != null && !messages.isEmpty()) {MapRecord<String, String, String> message = messages.get(0);// 存储到死信队列redisTemplate.opsForStream().add(streamKey + ":dead", message.getValue());// 确认原消息redisTemplate.opsForStream().acknowledge(streamKey, groupName, messageId);}} catch (Exception e) {log.error("移动消息到死信队列失败: messageId={}", messageId, e);}}
}

5.使用示例

5.1代码如下

java">@Service
@Slf4j
public class MessageService {@Autowiredprivate StreamMessageProducer producer;@Autowiredprivate StreamMessageConsumer consumer;@PostConstructpublic void init() {// 注册订单消息处理器consumer.registerHandler("order", OrderMessage.class, this::processOrderMessage);// 注册支付消息处理器consumer.registerHandler("payment", PaymentMessage.class, this::processPaymentMessage);}/*** 发送订单消息*/public String sendOrderMessage(OrderMessage message) {return producer.sendMessage("order", message);}/*** 处理订单消息*/private void processOrderMessage(OrderMessage message) {try {log.info("处理订单消息: {}", message);// 处理订单逻辑} catch (Exception e) {log.error("订单消息处理失败", e);throw e;}}/*** 处理支付消息*/private void processPaymentMessage(PaymentMessage message) {try {log.info("处理支付消息: {}", message);// 处理支付逻辑} catch (Exception e) {log.error("支付消息处理失败", e);throw e;}}
}@Data
@AllArgsConstructor
class OrderMessage {private String orderId;private String status;private BigDecimal amount;
}@Data
@AllArgsConstructor
class PaymentMessage {private String paymentId;private String orderId;private BigDecimal amount;private String status;
}

6.监控和管理服务

6.1 代码如下

java">@Service
@Slf4j
public class StreamMonitorService {@Autowiredprivate StringRedisTemplate redisTemplate;/*** 获取Stream信息*/public StreamInfo getStreamInfo(String topic) {try {String streamKey = STREAM_KEY + ":" + topic;StreamInfo.StreamInfoBuilder builder = StreamInfo.builder();// 获取消息总数Long length = redisTemplate.opsForStream().size(streamKey);builder.messageCount(length != null ? length : 0);// 获取消费者组信息StreamInfo.GroupInfo groupInfo = getGroupInfo(streamKey);builder.groupInfo(groupInfo);// 获取最新消息IDbuilder.lastMessageId(getLastMessageId(streamKey));return builder.build();} catch (Exception e) {log.error("获取Stream信息失败: topic={}", topic, e);throw new RuntimeException("获取Stream信息失败", e);}}/*** 清理过期消息*/@Scheduled(cron = "0 0 * * * *") // 每小时执行public void cleanupOldMessages() {try {for (String topic : getTopics()) {String streamKey = STREAM_KEY + ":" + topic;// 保留最近24小时的消息long maxLength = 1000000; // 最大消息数redisTemplate.opsForStream().trim(streamKey, maxLength, true);}} catch (Exception e) {log.error("清理过期消息失败", e);}}
}@Data
@Builder
class StreamInfo {private long messageCount;private String lastMessageId;private GroupInfo groupInfo;@Data@Builderstatic class GroupInfo {private String name;private int consumerCount;private long pendingMessageCount;}
}