问题思考
如何保证消息的可靠性传输？
通过subscibe/publish处理的消息没有持久化的特性，一旦出现网络中断，Redis宕机这类异常的时候就会导致消息丢失，而且也没有较好的机制取支持消息重复消费的问题。因此可靠性方面较差。
基于Stream实现消息队列
Redis5.0中发布的Stream类型，也用来实现典型的消息队列。提供了消息的持久化和主备复制功能，可以让任何客户端访问任何时刻的数据，并且能记住每一个客户端的访问位置，还能保证消息不丢失。该Stream类型的出现，几乎满足了消息队列具备的全部内容，包括但不限于：
- 消息ID的序列化生成
- 消息遍历
- 消息的阻塞和非阻塞读取
- 消息的分组消费
- 未完成消息的处理
- 消息队列监控
关于Stream的一些基本入门篇章这里不做过多介绍，感兴趣的朋友可以去阅读下这篇文章：
下边的部分我们直接来进入关于Redis XStream相关的实战环节。
封装消息监听功能
首先是定义一个MQ相关的接口：
public interface RedisStreamListener { /** * 处理正常消息 */ HandlerResult handleMsg(StreamEntry streamEntry); }
接着是基于这套接口做消息发送的实现：
package org.idea.mq.redis.framework.listener; import com.alibaba.fastjson.JSON; import org.idea.mq.redis.framework.bean.HandlerResult; import org.idea.mq.redis.framework.config.StreamListener; import org.idea.mq.redis.framework.mq.xstream.RedisStreamMQListener; import org.idea.mq.redis.framework.redis.IRedisService; import org.idea.mq.redis.framework.utils.PayMsg; import redis.clients.jedis.StreamEntry; import javax.annotation.Resource; import java.util.Map; import static org.idea.mq.redis.framework.config.MQConstants.SUCCESS; /** * @Author linhao * @Date created in 10:07 下午 2022/2/9 */ @StreamListener(streamName = "order-service:order-payed-stream", groupName = "order-service-group", consumerName = "user-service-consumer") public class OrderPayedListener implements RedisStreamMQListener { @Resource private IRedisService iRedisService; @Override public HandlerResult handleMsg(StreamEntry streamEntry) { Map<String, String> map = streamEntry.getFields(); String json = map.get("json"); PayMsg payMsg = JSON.parseObject(json, PayMsg.class); System.out.println("pending payMsg is : " + payMsg); return SUCCESS; } }
自定义消息注解
package org.idea.mq.redis.framework.config; import org.springframework.stereotype.Component; import java.lang.annotation.*; /** * @Author linhao * @Date created in 10:04 下午 2022/2/9 */ @Target({ElementType.TYPE}) @Retention(RetentionPolicy.RUNTIME) @Documented @Component public @interface StreamListener { String streamName() default ""; String groupName() default ""; String consumerName() default ""; }
代码中有一个自定义的@StreamListener的注解，该注解的内部包含了一个@Component的注解，可以将使用了该注解的对象注入到Spring容器中。
为了能将这些个初始化类进行自动装配，还需要加入一个配置的对象，代码如下：
package org.idea.mq.redis.framework.config; import org.idea.mq.redis.framework.bean.HandlerResult; import org.idea.mq.redis.framework.mq.xstream.RedisStreamMQListener; import org.idea.mq.redis.framework.redis.IRedisService; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.boot.context.event.ApplicationReadyEvent; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationListener; import org.springframework.context.annotation.Configuration; import org.springframework.stereotype.Component; import org.springframework.util.CollectionUtils; import redis.clients.jedis.StreamEntry; import redis.clients.jedis.StreamEntryID; import redis.clients.jedis.StreamPendingEntry; import javax.annotation.Resource; import java.util.List; import java.util.Map; import static org.idea.mq.redis.framework.config.MQConstants.SUCCESS; /** * @Author linhao * @Date created in 3:25 下午 2022/2/7 */ @Configuration public class StreamListenerConfiguration implements ApplicationListener<ApplicationReadyEvent> { @Resource private ApplicationContext applicationContext; @Resource private IRedisService iRedisService; private static Logger logger = LoggerFactory.getLogger(StreamListenerConfiguration.class); @Override public void onApplicationEvent(ApplicationReadyEvent applicationReadyEvent) { Map<String, RedisStreamMQListener> beanMap = applicationContext.getBeansOfType(RedisStreamMQListener.class); beanMap.values().forEach(redisStreamMQListener -> { StreamListener StreamListener = redisStreamMQListener.getClass().getAnnotation(StreamListener.class); ListenerInitWrapper listenerInitWrapper = new ListenerInitWrapper(StreamListener.streamName(), StreamListener.groupName(), StreamListener.consumerName()); Thread handleThread = new Thread(new CoreMsgHandlerThread(listenerInitWrapper, redisStreamMQListener, iRedisService)); Thread pendingHandleThread = new Thread(new PendingMsgHandlerThread(listenerInitWrapper, redisStreamMQListener, iRedisService)); handleThread.start(); pendingHandleThread.start(); logger.info("{} load successed ", redisStreamMQListener); }); } class PendingMsgHandlerThread implements Runnable { private ListenerInitWrapper listenerInitWrapper; private RedisStreamMQListener redisStreamMQListener; private IRedisService iRedisService; public PendingMsgHandlerThread(ListenerInitWrapper listenerInitWrapper, RedisStreamMQListener redisStreamMQListener, IRedisService iRedisService) { this.redisStreamMQListener = redisStreamMQListener; this.listenerInitWrapper = listenerInitWrapper; this.iRedisService = iRedisService; } @Override public void run() { String startId = "0-0"; while (true) { List<StreamPendingEntry> streamConsumersInfos = iRedisService.xpending(listenerInitWrapper.getStreamName(), listenerInitWrapper.getGroupName(), new StreamEntryID(startId), 1); //如果该集合非空，则触发监听行为 if (!CollectionUtils.isEmpty(streamConsumersInfos)) { for (StreamPendingEntry streamConsumersInfo : streamConsumersInfos) { StreamEntryID streamEntryID = streamConsumersInfo.getID(); //比当前pending的streamId小1 String streamIdStr = streamEntryID.toString(); String[] items = streamIdStr.split("-"); Long timestamp = Long.valueOf(items[0]) - 1; String beforeId = timestamp + "-" + "0"; List<Map.Entry<String, List<StreamEntry>>> result = iRedisService.xreadGroup(listenerInitWrapper.getStreamName(), listenerInitWrapper.getGroupName(), new StreamEntryID(beforeId), 1, listenerInitWrapper.getConsumerName()); for (Map.Entry<String, List<StreamEntry>> streamInfo : result) { List<StreamEntry> streamEntries = streamInfo.getValue(); for (StreamEntry streamEntry : streamEntries) { try { //业务处理 HandlerResult handlerResult = redisStreamMQListener.handleMsg(streamEntry); if (SUCCESS.equals(handlerResult)) { startId = streamEntryID.toString(); iRedisService.xack(listenerInitWrapper.getStreamName(), listenerInitWrapper.getGroupName(), new StreamEntryID(startId)); } } catch (Exception e) { logger.error("[PendingMsgHandlerThread] e is ", e); } } } } } try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } } } } class CoreMsgHandlerThread implements Runnable { private ListenerInitWrapper listenerInitWrapper; private RedisStreamMQListener redisStreamMQListener; private IRedisService iRedisService; public CoreMsgHandlerThread(ListenerInitWrapper listenerInitWrapper, RedisStreamMQListener redisStreamMQListener, IRedisService iRedisService) { this.redisStreamMQListener = redisStreamMQListener; this.listenerInitWrapper = listenerInitWrapper; this.iRedisService = iRedisService; } @Override public void run() { while (true) { List<Map.Entry<String, List<StreamEntry>>> streamConsumersInfos = iRedisService.xreadGroup(listenerInitWrapper.getStreamName(), listenerInitWrapper.getGroupName(), StreamEntryID.UNRECEIVED_ENTRY, 1, listenerInitWrapper.getConsumerName()); for (Map.Entry<String, List<StreamEntry>> streamInfo : streamConsumersInfos) { List<StreamEntry> streamEntries = streamInfo.getValue(); for (StreamEntry streamEntry : streamEntries) { //业务处理 try { HandlerResult result = redisStreamMQListener.handleMsg(streamEntry); if (SUCCESS.equals(result)) { iRedisService.xack(listenerInitWrapper.getStreamName(), listenerInitWrapper.getGroupName(), streamEntry.getID()); } } catch (Exception e) { logger.error("[CoreMsgHandlerThread] e is ", e); } } } } } } }