久久精品一区二区三区日韩,厚颜无耻有字幕在线观看 ,亚洲第一激情

基于Redis6.0 部署迷你版本消息隊(duì)列

技術(shù)研究背景

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

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

具體技術(shù)選型原因：

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

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

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

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

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

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

如果本地沒有相關(guān)鏡像，可以嘗試通過搭建下方命令進(jìn)行鏡像的拉取：

docker pull redis:6.0.6

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

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

基于常規(guī)的隊(duì)列結(jié)構(gòu)來實(shí)現(xiàn)消息隊(duì)列

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

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; } }

問題思考

這里存在幾個問題點(diǎn)需要思考下：

多個服務(wù)之間如何訂閱同一個消息

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

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

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

基于Redis6.0 部署迷你版本消息隊(duì)列

在這里插入圖片描述

消息的監(jiān)聽機(jī)制如何實(shí)現(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<String> values = iRedisService.brpop(msgWrapper.getTopic()); if (!CollectionUtils.isEmpty(values)) { values.forEach(value -> { System.out.println(value); }); } } }

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

在設(shè)計(jì)消息隊(duì)列的時候，我們非常看重的就是消息的可靠性保證。當(dāng)一條消息發(fā)送到消費(fèi)端之后，如果出現(xiàn)了異常，希望消息能夠?qū)崿F(xiàn)重新發(fā)送的效果。

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

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

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

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

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

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

list更多的是實(shí)現(xiàn)點(diǎn)對點(diǎn)方式的傳輸（P2P方式）
subscibe/publish則是可以實(shí)現(xiàn)廣播的方式和訂閱者進(jì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)聽的部分可以通過額外開啟一個線程來實(shí)現(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(); } }

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

問題思考

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

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

基于Stream實(shí)現(xiàn)消息隊(duì)列

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

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

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

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

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

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

首先是定義一個MQ相關(guān)的接口：

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

接著是基于這套接口做消息發(fā)送的實(shí)現(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<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的注解，該注解的內(nèi)部包含了一個@Component的注解，可以將使用了該注解的對象注入到Spring容器中。

為了能將這些個初始化類進(jìn)行自動裝配，還需要加入一個配置的對象，代碼如下：

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); //如果該集合非空，則觸發(fā)監(jiān)聽行為
                if (!CollectionUtils.isEmpty(streamConsumersInfos)) { for (StreamPendingEntry streamConsumersInfo : streamConsumersInfos) { StreamEntryID streamEntryID = streamConsumersInfo.getID(); //比當(dāng)前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 { //業(yè)務(wù)處理
                                    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) { //業(yè)務(wù)處理
                        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); } } } } } } }

其原理是在Spring容器啟動好了之后，監(jiān)聽Spring容器內(nèi)部發(fā)出的ApplicationReadyEvent事件，監(jiān)聽該事件，并且開啟兩個后臺線程用于處理redis內(nèi)部的stream數(shù)據(jù)。

封裝相關(guān)的消息發(fā)布功能

消息的發(fā)送部分比較簡單，直接通過redis往stream里面寫入數(shù)據(jù)即可

package org.idea.mq.redis.framework.producer; /**  * @Author linhao  * @Date created in 12:23 下午 2022/2/10  */ public interface IStreamProducer { /**  * 指定streamName發(fā)布消息  * @param streamName  * @param json  */ void sendMsg(String streamName, String json); }

消息的傳輸格式采用json字符串的方式寫入到redis內(nèi)部的stream當(dāng)中。

package org.idea.mq.redis.framework.producer; import org.idea.mq.redis.framework.redis.IRedisService; import javax.annotation.Resource; import java.util.HashMap; import java.util.Map; /**  * @Author linhao  * @Date created in 12:19 下午 2022/2/10  */ public class StreamProducer implements IStreamProducer{ @Resource
    private IRedisService iRedisService; @Override
    public void sendMsg(String streamName,String json){ Map<String,String> map = new HashMap<>(); map.put("json",json); iRedisService.xAdd(streamName,map); } }

注意，寫入底層的時候，我使用的是Redis內(nèi)部自動生成的ID序號，代碼如下：

@Override
public boolean xAdd(String streamName, Map<String, String> stringMap) { try (Jedis jedis = iRedisFactory.getConnection()) { jedis.xadd(streamName, StreamEntryID.NEW_ENTRY, stringMap); return true; } catch (Exception e) { throw new RuntimeException(e); } }

為了方便將其作為一個SpringBoot的starter組件供外界團(tuán)隊(duì)人員使用，我們可以將其封裝為一個starter組件：

基于Redis6.0 部署迷你版本消息隊(duì)列

在這里插入圖片描述

組件的測試

點(diǎn)對點(diǎn)發(fā)送測試

建立兩套微服務(wù)工程，user-service 和 order-service，其中user-service部署兩個服務(wù)節(jié)點(diǎn)，同屬user-service-group。order-service也要部署兩個服務(wù)節(jié)點(diǎn)，同屬order-service-group。

最后兩個微服務(wù)集群之間互相發(fā)布對方訂閱的消息，查看是否能夠正常接受，且同一個組內(nèi)一次只有一個節(jié)點(diǎn)接收消息。

基于Redis6.0 部署迷你版本消息隊(duì)列

在這里插入圖片描述

廣播發(fā)送測試

使用之前搭建好的user-service模塊，部署四個節(jié)點(diǎn)，訂閱同一個stream隊(duì)列，但是將其groupName設(shè)置為不同的屬性，最后發(fā)布消息，查看四個節(jié)點(diǎn)都能正常接收。

基于Redis6.0 部署迷你版本消息隊(duì)列

在這里插入圖片描述

具體細(xì)節(jié)在現(xiàn)有工程內(nèi)部已經(jīng)建立了測試模版，感興趣的朋友可以去閱讀下mq-redis-test模塊的部分。

問題思考

為何同一個StreamName需要采用雙線程消費(fèi)？

一個線程用于接受Stream內(nèi)部正常數(shù)據(jù)，如果業(yè)務(wù)正常處理則對其返回為ack信號，確認(rèn)該消息已經(jīng)消費(fèi)成功。如果處理過程中出現(xiàn)異常，則不反回ACK信號，此時Redis內(nèi)部會將該消息放入到Pending隊(duì)列中，而第二個線程專門用于處理Pending隊(duì)列內(nèi)部的數(shù)據(jù)。如果處于Pending狀態(tài)的消息第二次消費(fèi)依然失敗，則會進(jìn)行定時輪詢狀況。

是否支持延遲重試

目前的設(shè)計(jì)其實(shí)一直都存在不足點(diǎn)，例如當(dāng)消息消費(fèi)異常后會進(jìn)入輪詢，嚴(yán)重情況下可能會導(dǎo)致消息消費(fèi)出現(xiàn)死循環(huán)，并且一直堵塞。暫時還未實(shí)現(xiàn)類似于RocketMQ的那種間隔1，3，5...分鐘定時投遞消費(fèi)失敗消息都功能。感興趣的小伙伴可以基于現(xiàn)有代碼進(jìn)行簡單改造。

原文地址:https://mp.weixin.qq.com/s/sMQGHvy4enf2e34KtBls-A