簡(jiǎn)介：在大部分場(chǎng)景下業(yè)務(wù)系統(tǒng)如果只需要實(shí)現(xiàn)異步解耦、削峰填谷等能力，常規(guī)的普通消息就可以滿足此類(lèi)需求。除此之外，在某些特殊的業(yè)務(wù)場(chǎng)景中，普通消息類(lèi)型存在無(wú)法滿足需求的情況。這就需要消息隊(duì)列服務(wù)本身支持一些特殊的消息類(lèi)型，或者開(kāi)發(fā)者通過(guò)開(kāi)發(fā)一些定制化的代碼實(shí)現(xiàn)目的。

前言

消息隊(duì)列服務(wù)相信大家一定都不陌生了，在很多應(yīng)用系統(tǒng)中，都有一些場(chǎng)景會(huì)使用到消息隊(duì)列服務(wù)，簡(jiǎn)單來(lái)說(shuō)，我們可以把消息隊(duì)列比作是一個(gè)存放消息的容器，上游發(fā)送端將消息發(fā)送到消息隊(duì)列，下游消費(fèi)端從消息隊(duì)列里消費(fèi)消息。消息隊(duì)列是分布式系統(tǒng)中重要的組件，核心作用可以幫助我們實(shí)現(xiàn)異步、解耦以及削峰，從而提高系統(tǒng)性能和穩(wěn)定性。

在大部分場(chǎng)景下業(yè)務(wù)系統(tǒng)如果只需要實(shí)現(xiàn)異步解耦、削峰填谷等能力，常規(guī)的普通消息就可以滿足此類(lèi)需求。除此之外，在某些特殊的業(yè)務(wù)場(chǎng)景中，普通消息類(lèi)型存在無(wú)法滿足需求的情況。這就需要消息隊(duì)列服務(wù)本身支持一些特殊的消息類(lèi)型，或者開(kāi)發(fā)者通過(guò)開(kāi)發(fā)一些定制化的代碼實(shí)現(xiàn)目的。這里我們列舉在使用消息隊(duì)列過(guò)程中幾種特殊場(chǎng)景的例子：

順序消費(fèi)場(chǎng)景

生產(chǎn)者按照一定的先后順序發(fā)布消息，消費(fèi)者按照既定的先后順序消費(fèi)消息，即先發(fā)布的消息一定會(huì)先被客戶端消費(fèi)。

分布式事務(wù)場(chǎng)景

分布式架構(gòu)下，隨著系統(tǒng)的演進(jìn)，數(shù)據(jù)庫(kù)也進(jìn)行了垂直拆分，如果選擇使用消息隊(duì)列進(jìn)行上下游解耦的話，生產(chǎn)者和消費(fèi)者需要保證數(shù)據(jù)一致性。

延時(shí)消費(fèi)場(chǎng)景

生產(chǎn)者將消息發(fā)送到消息隊(duì)列后，并不期望立馬投遞這條消息，而是推遲到某個(gè)時(shí)間點(diǎn)之后將消息投遞給消費(fèi)者進(jìn)行消費(fèi)。

對(duì)于順序消息和事務(wù)消息，這里就不進(jìn)行詳細(xì)介紹了，大家有興趣可以自行研究，本文后續(xù)內(nèi)容會(huì)和大家一起詳細(xì)討論下延時(shí)消息更多的細(xì)節(jié)及應(yīng)用場(chǎng)景。

延時(shí)消息介紹

延時(shí)(定時(shí))消息的特點(diǎn)就是發(fā)送者成功發(fā)送一條消息后，這條消息并不會(huì)馬上被消費(fèi)者消費(fèi)，而是在某個(gè)特定的時(shí)間或者延遲一段時(shí)間后，消息才被消費(fèi)者可見(jiàn)并進(jìn)行后續(xù)的消費(fèi)，延時(shí)消息整個(gè)生命周期可以用如下示意圖來(lái)表示：

消息發(fā)布者將一條延時(shí)消息發(fā)送到消息隊(duì)列服務(wù)端；

在預(yù)計(jì)投遞時(shí)間未到之前，消息對(duì)消費(fèi)者不可見(jiàn)，消費(fèi)者此時(shí)無(wú)法立刻消費(fèi)；

投遞時(shí)間到達(dá)后，消息才對(duì)消費(fèi)者可見(jiàn)，消費(fèi)者此時(shí)可以消費(fèi)；

消費(fèi)者獲取此條消息并進(jìn)行消費(fèi)；

消費(fèi)者成功消費(fèi)后，進(jìn)行確認(rèn)，此條消息將不再被消費(fèi)。

延時(shí)消息應(yīng)用場(chǎng)景

交易場(chǎng)景

在生產(chǎn)者和消費(fèi)者有時(shí)間窗口的要求下，我們可以考慮使用延時(shí)消息。如在電商交易場(chǎng)景下，交易中超時(shí)未支付的訂單需要被關(guān)閉的場(chǎng)景，在訂單創(chuàng)建時(shí)會(huì)發(fā)送一條延時(shí)消息。這條消息將會(huì)在30分鐘以后投遞給消費(fèi)者，消費(fèi)者收到此消息后，需要判斷對(duì)應(yīng)的訂單是否已完成支付；如支付未完成，則關(guān)閉訂單。

游戲場(chǎng)景

再比如在游戲社區(qū)里，游戲運(yùn)營(yíng)方經(jīng)常會(huì)發(fā)起一些活動(dòng)，玩家在活動(dòng)期間內(nèi)按照規(guī)則完成一系列任務(wù)，活動(dòng)時(shí)間截止后，游戲后臺(tái)根據(jù)玩家完成任務(wù)的情況進(jìn)行判定，發(fā)送系統(tǒng)通知或者進(jìn)行rank排名并派發(fā)獎(jiǎng)勵(lì)等。

此種場(chǎng)景也可以采用延時(shí)消息來(lái)實(shí)現(xiàn)，上游系統(tǒng)發(fā)布活動(dòng)公告后，同時(shí)發(fā)送一條延時(shí)消息，延時(shí)時(shí)間設(shè)置為活動(dòng)周期的時(shí)間。當(dāng)活動(dòng)截止后，下游系統(tǒng)可以隨即消費(fèi)消息并進(jìn)行相應(yīng)的邏輯處理。

其他場(chǎng)景

同時(shí)延時(shí)消息也可以廣泛應(yīng)用于信息提醒等比較通用的場(chǎng)景。

如何實(shí)現(xiàn)延時(shí)消息

介紹完延時(shí)消息的一些概念及應(yīng)用場(chǎng)景后，我們接下來(lái)分析一下目前比較主流的幾款開(kāi)源消息中間件對(duì)延時(shí)消息的支持情況以及實(shí)現(xiàn)方式。

Kafka

原生Kafka默認(rèn)是不支持延時(shí)消息的，需要開(kāi)發(fā)者自己實(shí)現(xiàn)一層代理服務(wù)，比如發(fā)送端將消息發(fā)送到延時(shí)Topic，代理服務(wù)消費(fèi)延時(shí)Topic的消息然后轉(zhuǎn)存起來(lái)，代理服務(wù)通過(guò)一定的算法，計(jì)算延時(shí)消息所附帶的延時(shí)時(shí)間是否到達(dá)，然后將延時(shí)消息取出來(lái)并發(fā)送到實(shí)際的Topic里面，消費(fèi)端從實(shí)際的Topic里面進(jìn)行消費(fèi)。

RabbitMQ

RabbitMQ實(shí)現(xiàn)延時(shí)消息有兩種方案，第一種是采用rabbitmq-delayed-message-exchange 插件實(shí)現(xiàn)，第二種則是利用DLX（Dead Letter Exchanges）+ TTL（消息存活時(shí)間）來(lái)間接實(shí)現(xiàn)。大致的實(shí)現(xiàn)思路如下：

創(chuàng)建一個(gè)普通隊(duì)列delay_queue，為此隊(duì)列設(shè)置死信交換機(jī) (通過(guò)x-dead-letter-exchange參數(shù)) 和 RoutingKey (通過(guò)x-dead-letter-routing-key參數(shù))，生產(chǎn)者將向delay_queue發(fā)送延時(shí)消息。

創(chuàng)建步驟1中設(shè)置的死信交換機(jī)，同時(shí)創(chuàng)建一個(gè)目標(biāo)隊(duì)列 target_queue，并使用步驟1中設(shè)置的RoutingKey將兩者綁定起來(lái)。消費(fèi)者將從target_queue里面消費(fèi)延時(shí)消息。

設(shè)置消息的存活時(shí)間TTL，可以在步驟1中設(shè)置到隊(duì)列級(jí)別delay_queue的消息存活時(shí)間，或者在發(fā)送消息時(shí)動(dòng)態(tài)設(shè)置消息級(jí)別的存活時(shí)間。

RocketMQ

開(kāi)源RocketMQ支持延遲消息，但是不支持秒級(jí)精度。默認(rèn)支持18個(gè)level的延遲消息，這是通過(guò)broker端的messageDelayLevel配置項(xiàng)確定的messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h

消息隊(duì)列服務(wù)在啟動(dòng)時(shí)，會(huì)創(chuàng)建一個(gè)內(nèi)部topic：SCHEDULE_TOPIC_XXXX，根據(jù)延遲level的個(gè)數(shù)，創(chuàng)建對(duì)應(yīng)數(shù)量的隊(duì)列。生產(chǎn)者發(fā)送消息時(shí)可以設(shè)置延時(shí)等級(jí)，示例代碼:

Message msg=new Message(); msg.setTopic("TopicA"); msg.setBody("this is a delay message".getBytes()); //設(shè)置延遲level為5，對(duì)應(yīng)延遲1分鐘 msg.setDelayTimeLevel(5); producer.send(msg);

發(fā)送的消息會(huì)暫存在Broker對(duì)應(yīng)的內(nèi)部topic中，再通過(guò)定時(shí)任務(wù)從內(nèi)部topic中拉取數(shù)據(jù)，如果延遲時(shí)間到了，就會(huì)把消息轉(zhuǎn)發(fā)到目標(biāo)topic下，消費(fèi)者從目標(biāo)topic消費(fèi)消息。

阿里云消息隊(duì)列RocketMQ版

通過(guò)上一章節(jié)的討論，我們可以看出目前幾款主流的開(kāi)源消息隊(duì)列服務(wù)，在支持延時(shí)消息的場(chǎng)景下或多或少有些不完美的地方。主要體現(xiàn)在以下幾點(diǎn):

Kafka不支持延時(shí)消息，需要完全開(kāi)發(fā)代理服務(wù)來(lái)實(shí)現(xiàn)，工作量大。

RabbitMQ需要額外的插件，或者利用DLX+TTL的方式進(jìn)行中轉(zhuǎn)，實(shí)現(xiàn)不是非常直觀。

RocketMQ支持延時(shí)消息，但是無(wú)法支持秒級(jí)延時(shí)。

那么有沒(méi)有一款消息隊(duì)列服務(wù)，能夠完美的支持延時(shí)（定時(shí)）消息。本節(jié)我們將介紹阿里云消息隊(duì)列RocketMQ版。

阿里云消息隊(duì)列RocketMQ版基于Apache RocketMQ構(gòu)建的低延遲、高并發(fā)、高可用、高可靠的分布式消息中間件。消息隊(duì)列RocketMQ版既可為分布式應(yīng)用系統(tǒng)提供異步解耦和削峰填谷的能力，同時(shí)也具備互聯(lián)網(wǎng)應(yīng)用所需的海量消息堆積、高吞吐、可靠重試等特性。同時(shí)支持豐富的消息類(lèi)型包括普通消息、順序消息、事務(wù)消息以及我們本文討論的延時(shí)消息。接下來(lái)我們看下阿里云RocketMQ為延時(shí)消息提供的能力及優(yōu)勢(shì)：

支持秒級(jí)的延時(shí)(定時(shí))消息，同時(shí)延時(shí)時(shí)間可以最大設(shè)置為40天，基本滿足所有場(chǎng)景。

延時(shí)(定時(shí))消息的投遞精度可控制在1~2秒之內(nèi)。

延時(shí)(定時(shí))消息在某段時(shí)間內(nèi)是對(duì)消費(fèi)者不可見(jiàn)的，從另一個(gè)維度看也屬于積壓的消息，阿里云消息隊(duì)列RocketMQ版的不同實(shí)例規(guī)格可以支持億級(jí)的消息積壓。

提供了多語(yǔ)言支持，包括Java、.NET、CC++、GO、Python、PHP、Node.js等

使用阿里云消息隊(duì)列RocketMQ版收發(fā)延時(shí)(定時(shí))消息，只需要在控制臺(tái)創(chuàng)建Topic的時(shí)候選擇定時(shí)/延時(shí)消息類(lèi)型，既可以使用TCP或者h(yuǎn)ttp協(xié)議進(jìn)行消息收發(fā)。

控制臺(tái)創(chuàng)建定時(shí)/延時(shí)Topic

Java語(yǔ)言示例代碼（TCP協(xié)議）

• 發(fā)送定時(shí)消息

// 定時(shí)消息，單位毫秒（ms），在指定時(shí)間戳（當(dāng)前時(shí)間之后）進(jìn)行投遞，例如2020-03-07 16:21:00投遞。如果被設(shè)置成當(dāng)前時(shí)間戳之前的某個(gè)時(shí)刻，消息將立刻投遞給消費(fèi)者。 long timeStamp = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse("2020-03-07 16:21:00").getTime(); msg.setStartDeliverTime(timeStamp); // 發(fā)送消息，只要不拋異常就是成功。 SendResult sendResult = producer.send(msg);

• 發(fā)送延時(shí)消息

// 延時(shí)消息，單位毫秒（ms），在指定延遲時(shí)間（當(dāng)前時(shí)間之后）進(jìn)行投遞，例如消息在3秒后投遞。 long delayTime = System.currentTimeMillis() + 3000; // 設(shè)置消息需要被投遞的時(shí)間。 msg.setStartDeliverTime(delayTime); SendResult sendResult = producer.send(msg);

同時(shí)訂閱延時(shí)消息的邏輯無(wú)需任何改造，完全可以按照訂閱普通消息的方式，沒(méi)有任何的代碼侵入性。

結(jié)束語(yǔ)

到此我們討論了延時(shí)消息的特性、應(yīng)用場(chǎng)景，對(duì)比了各類(lèi)消息隊(duì)列對(duì)延時(shí)消息的支持情況，同時(shí)也向大家介紹了阿里云消息隊(duì)列RocketMQ版。我們?cè)趯?duì)消息中間件進(jìn)行選型時(shí)，也會(huì)考慮到多方面的因素。除了消息中間件本身所能提供的能力外，也包括服務(wù)性能、穩(wěn)定性、可擴(kuò)展能力，以及需要結(jié)合開(kāi)發(fā)團(tuán)隊(duì)自身的技術(shù)棧等情況。最后如果大家想了解更多阿里云消息隊(duì)列RocketMQ版。

作者：阿里云解決方案架構(gòu)師 鹿玄

原文鏈接

本文為阿里云原創(chuàng)內(nèi)容，未經(jīng)允許不得轉(zhuǎn)載

《新程序員》：云原生和全面數(shù)字化實(shí)踐50位技術(shù)專家共同創(chuàng)作，文字、視頻、音頻交互閱讀

總結(jié)

以上是生活随笔為你收集整理的计算发送延时与传播延迟_消息队列——延时消息应用解析及实践的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。