好看的课外书,完美世界小说下载

新聞中心

這里有您想知道的互聯(lián)網營銷解決方案

實戰(zhàn)干貨：基于Redis6.0部署迷你版本消息隊列

技術研究背景

由于目前的研發(fā)團隊處于公司初創(chuàng)階段，尚未有能成熟的運維體系，對于市面上常見的成熟MQ搭建維護能力不足，但是又希望能有一款輕量級的消息系統(tǒng)供研發(fā)團隊的成員使用，因此開展了對該方面相關的技術調研工作。

目前創(chuàng)新互聯(lián)已為千余家的企業(yè)提供了網站建設、域名、虛擬空間、網站托管、企業(yè)網站設計、本溪網站維護等服務，公司將堅持客戶導向、應用為本的策略，正道將秉承"和諧、參與、激情"的文化，與客戶和合作伙伴齊心協(xié)力一起成長，共同發(fā)展。

通過相關的技術調研后，決定挑選基于Redis實現(xiàn)消息系統(tǒng)。

具體技術選型原因：

團隊內部已經有搭建相關的Redis服務，并且具備一定的運維能力，可以節(jié)省技術成本
業(yè)界有較多關于Redis搭建消息系統(tǒng)方面的技術文章
目前的系統(tǒng)的整體吞吐量并不高，接入消息系統(tǒng)的主要目的只是為了實現(xiàn)系統(tǒng)之間的解耦

為了方便讓讀者們從0到1地學習這塊內容，我將會從環(huán)節(jié)搭建開始介紹起。

基本環(huán)境的搭建

基于redis6.0.6版本搭建一套簡單的消息隊列系統(tǒng)。環(huán)境部署：

docker run -p 6379:6379 --name redis_6_0_6 -d redis:6.0.6

參數(shù)解釋： -d 后臺啟動 -p 端口映射 -name 容器名稱

如果本地沒有相關鏡像，可以嘗試通過搭建下方命令進行鏡像的拉?。?

docker pull redis:6.0.6

當redis的基礎環(huán)境配置好了之后，接下來便是基于redis內置的一些基本功能開發(fā)一款消息隊列組件了。

下邊我將分三種不同的技術方案來介紹如何實現(xiàn)一款輕量級的消息隊列。

基于常規(guī)的隊列結構來實現(xiàn)消息隊列

這塊的實現(xiàn)比較簡單，主要是基于Redis內部的List結構來落地的，發(fā)送方將消息從隊列的左邊寫入，然后消費方從隊列的右邊讀取。

package org.idea.mq.redis.framework.mq.list;
import com.alibaba.fastjson.JSON;
import org.idea.mq.redis.framework.bean.MsgWrapper;
import org.idea.mq.redis.framework.mq.IMQTemplate;
import org.idea.mq.redis.framework.redis.IRedisService;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
/**
 * @Author linhao
 * @Date created in 3:09 下午 2022/2/7
 */
@Component
public class RedisListMQTemplate implements IMQTemplate {
    @Resource
    private IRedisService iRedisService;
    @Override
    public boolean send(MsgWrapper msgWrapper) {
        try {
            String json = JSON.toJSONString(msgWrapper.getMsgInfo());
            iRedisService.lpush(msgWrapper.getTopic(),json);
            return true;
        }catch (Exception e){
            e.printStackTrace();
        }
        return false;
    }
}

問題思考

這里存在幾個問題點需要思考下：

多個服務之間如何訂閱同一個消息

這里我建議可以按照系統(tǒng)的項目名稱前綴+業(yè)務標識來組織。

例如：用戶系統(tǒng)中需要發(fā)布一條會員已升級的消息給到下游系統(tǒng)，此時可以將這條消息寫入到名為：user-service:member-upgrade-list 的List集合中。

如果訂單系統(tǒng)希望訪問用戶系統(tǒng)的消息，則需要在redis的key里指定user-service:member-upgrade-list關鍵字。

在這里插入圖片描述

消息的監(jiān)聽機制如何實現(xiàn)？

對于List的消息可以采用輪詢的方式獲取，例如下邊這段案例代碼：

/**
 * 輪詢的方式獲取數(shù)據(jù)
 *
 * @param msgWrapper
 */
private void pollingGet(MsgWrapper msgWrapper) {
    while (true) {
        String value = iRedisService.rpop(msgWrapper.getTopic());
        if (!StringUtils.isEmpty(value)) {
            System.out.println(value);
        }
        //減少訪問壓力，定期睡眠一段時間
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

但是輪詢的方式比較消耗性能，所以可以嘗試使用Redis的阻塞式彈出指令，例如下邊這種方式來監(jiān)聽消息的觸發(fā)行為：

/**
 * 阻塞的方式獲取數(shù)據(jù)
 */
private void blockGet(MsgWrapper msgWrapper) {
    while (true) {
        List values = iRedisService.brpop(msgWrapper.getTopic());
        if (!CollectionUtils.isEmpty(values)) {
            values.forEach(value -> {
                System.out.println(value);
            });
        }
    }
}

消息的可靠性傳輸如何確保？

在設計消息隊列的時候，我們非?？粗氐木褪窍⒌目煽啃员ＷC。當一條消息發(fā)送到消費端之后，如果出現(xiàn)了異常，希望消息能夠實現(xiàn)重新發(fā)送的效果。

對于這種場景的設計我們可以嘗試使用 BRPOPLPUSH 這條指令，這條指令可以幫助我們在Redis內部將數(shù)據(jù)彈出時寫入到另一個備份隊列中，這樣即使彈出的消息消費失敗了，備份隊列中還有一份備用消息可以使用，而且彈出和寫入備份隊列操作在Redis內部做了封裝，外界調用可以視作為一個原子操作。

是否可以支持廣播的模式？

從List集合的實現(xiàn)原理來看，Redis彈出的元素只能返回給一個客戶端鏈接，因此無法支持廣播這種效果的實現(xiàn)。

基于發(fā)布訂閱功能實現(xiàn)消息隊列

Redis的內部提供了一個叫做發(fā)布訂閱的功能，通過subscibe命令和publish指令可以幫助我們實現(xiàn)關于消息發(fā)布和通知的功能。

使用subscibe/publish命令實現(xiàn)的效果和List結構最大的不同在于它的傳輸方式：

list更多的是實現(xiàn)點對點方式的傳輸（P2P方式）
subscibe/publish則是可以實現(xiàn)廣播的方式和訂閱者進行通信

publish部分的案例代碼：

@Override
public boolean publish(String channel, String content) {
    try (Jedis jedis = iRedisFactory.getConnection()) {
        jedis.publish(channel, content);
        return true;
    } catch (Exception e) {
        throw new RuntimeException(e);
    }
}

subscibe部分的代碼：

@Override
public boolean subscribe(JedisPubSub jedisPubSub, String... channel) {
    try (Jedis jedis = iRedisFactory.getConnection()) {
        jedis.subscribe(jedisPubSub, channel);
        return true;
    } catch (Exception e) {
        throw new RuntimeException(e);
    }
}

監(jiān)聽的部分可以通過額外開啟一個線程來實現(xiàn)這部分效果：

@Component
public class RedisSubscribeMQListener implements IMQListener {
    @Resource
    private IRedisService iRedisService;
    class TestChannel extends JedisPubSub {
        @Override
        public void onMessage(String channel, String message) {
            super.onMessage(channel, message);
            System.out.println("channel " + channel + " 接收到消息：" + message);
        }
        @Override
        public void onSubscribe(String channel, int subscribedChannels) {
            System.out.println(String.format("subscribe redis channel success, channel %s, subscribedChannels %d",
                    channel, subscribedChannels));
        }
        @Override
        public void onUnsubscribe(String channel, int subscribedChannels) {
            System.out.println(String.format("unsubscribe redis channel, channel %s, subscribedChannels %d",
                    channel, subscribedChannels));
        }
    }
    //所有頻道的消息都監(jiān)聽
    @Override
    public void onMessageReach(MsgWrapper msgWrapper) {
        Thread thread = new Thread(new Runnable() {
            @Override
            public void run() {
                iRedisService.subscribe(new TestChannel(), msgWrapper.getTopic());
            }
        });
        thread.start();
    }
}

要注意，回調通知的時候需要注入一個JedisPubSub的對象，這個對象的內部定義了接收消息之后的處理操作。

問題思考

如何保證消息的可靠性傳輸？

通過subscibe/publish處理的消息沒有持久化的特性，一旦出現(xiàn)網絡中斷，Redis宕機這類異常的時候就會導致消息丟失，而且也沒有較好的機制取支持消息重復消費的問題。因此可靠性方面較差。

基于Stream實現(xiàn)消息隊列

Redis5.0中發(fā)布的Stream類型，也用來實現(xiàn)典型的消息隊列。提供了消息的持久化和主備復制功能，可以讓任何客戶端訪問任何時刻的數(shù)據(jù)，并且能記住每一個客戶端的訪問位置，還能保證消息不丟失。該Stream類型的出現(xiàn)，幾乎滿足了消息隊列具備的全部內容，包括但不限于：

消息ID的序列化生成
消息遍歷
消息的阻塞和非阻塞讀取
消息的分組消費
未完成消息的處理
消息隊列監(jiān)控

關于Stream的一些基本入門篇章這里不做過多介紹，感興趣的朋友可以去閱讀下這篇文章：

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

下邊的部分我們直接來進入關于Redis XStream相關的實戰(zhàn)環(huán)節(jié)。

封裝消息監(jiān)聽功能

首先是定義一個MQ相關的接口：

public interface RedisStreamListener {
    /**
     * 處理正常消息
     */
    HandlerResult handleMsg(StreamEntry streamEntry);
}

接著是基于這套接口做消息發(fā)送的實現(xiàn)：

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 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 {
    @Resource
    private ApplicationContext applicationContext;
    @Resource
    private IRedisService iRedisService;
    private static Logger logger = LoggerFactory.getLogger(StreamListenerConfiguration.class);
    @Override
    public void onApplicationEvent(ApplicationReadyEvent applicationReadyEvent) {
        Map 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 streamConsumersInfos = iRedisService.xpending(listenerInitWrapper.getStreamName(), listenerInitWrapper.getGroupName(), new StreamEntryID(startId), 1);
                //如果該集合非空，則觸發(fā)監(jiān)聽行為
                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>> result = iRedisService.xreadGroup(listenerInitWrapper.getStreamName(), listenerInitWrapper.getGroupName(), new StreamEntryID(beforeId), 1, listenerInitWrapper.getConsumerName());
                        for (Map.Entry> streamInfo : result) {
                            List streamEntries = streamInfo.getValue();
                            for (StreamEntry streamEntry : streamEntries) {
                                try {
                                    //業(yè)務處理
                                    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>> streamConsumersInfos = iRedisService.xreadGroup(listenerInitWrapper.getStreamName(), listenerInitWrapper.getGroupName(), StreamEntryID.UNRECEIVED_ENTRY, 1, listenerInitWrapper.getConsumerName());
                for (Map.Entry> streamInfo : streamConsumersInfos) {
                    List streamEntries = streamInfo.getValue();
                    for (StreamEntry streamEntry : streamEntries) {
                        //業(yè)務處理
                        try {
                            HandlerResult result = redisStreamMQListener.handleMsg(streamEntry);
                            if (SUCCESS.equals(result)) {
                                iRedisService.xack(listenerInitWrapper.getStreamName(), listenerInitWrapper.getGroupName(), streamEntry.getID());
                            }
                        } catch (Exception e)                                                

                                                文章題目：實戰(zhàn)干貨：基于Redis6.0部署迷你版本消息隊列                                                

                                                瀏覽地址：http://fisionsoft.com.cn/article/djsjcgo.html

新聞中心

技術研究背景

基本環(huán)境的搭建

問題思考

多個服務之間如何訂閱同一個消息

其他資訊