实战干货:基于Redis6.0 部署迷你版本消息队列(中)

问题思考


如何保证消息的可靠性传输?


通过subscibe/publish处理的消息没有持久化的特性,一旦出现网络中断,Redis宕机这类异常的时候就会导致消息丢失,而且也没有较好的机制取支持消息重复消费的问题。因此可靠性方面较差。


基于Stream实现消息队列


Redis5.0中发布的Stream类型,也用来实现典型的消息队列。提供了消息的持久化和主备复制功能,可以让任何客户端访问任何时刻的数据,并且能记住每一个客户端的访问位置,还能保证消息不丢失。该Stream类型的出现,几乎满足了消息队列具备的全部内容,包括但不限于:


  • 消息ID的序列化生成
  • 消息遍历
  • 消息的阻塞和非阻塞读取
  • 消息的分组消费
  • 未完成消息的处理
  • 消息队列监控


关于Stream的一些基本入门篇章这里不做过多介绍,感兴趣的朋友可以去阅读下这篇文章:


https://xie.infoq.cn/article/cdb47caddc5ff49dc09ea58cd


下边的部分我们直接来进入关于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);
                        }
                    }
                }
            }
        }
    }
}



上一篇:SpringMVC 函数式编程进阶


下一篇:Android多进程之手动编写Binder类