電子發(fā)燒友網(wǎng)>電子資料下載>電子資料>delaymq高性能的任意延時(shí)消息隊(duì)列

delaymq高性能的任意延時(shí)消息隊(duì)列

2388610 2022-06-14 | zip | 26.84 MB | 次下載 | 2積分

普通下載普通下載

資料介紹

授權(quán)協(xié)議 GPL

開(kāi)發(fā)語(yǔ)言 Java

操作系統(tǒng) 跨平臺(tái)

軟件類(lèi)型開(kāi)源軟件

所屬分類(lèi) 服務(wù)器軟件、 JMS/消息中間件

軟件簡(jiǎn)介

RocketMQ 開(kāi)源版本任意時(shí)間延時(shí)隊(duì)列實(shí)現(xiàn)

定時(shí)消息：Producer將消息發(fā)送到消息隊(duì)列RocketMQ版服務(wù)端，但并不期望立馬投遞這條消息，而是推遲到在當(dāng)前時(shí)間點(diǎn)之后的某一個(gè)時(shí)間投遞到Consumer進(jìn)行消費(fèi)，該消息即定時(shí)消息。

延時(shí)消息：Producer將消息發(fā)送到消息隊(duì)列RocketMQ版服務(wù)端，但并不期望立馬投遞這條消息，而是延遲一定時(shí)間后才投遞到Consumer進(jìn)行消費(fèi)，該消息即延時(shí)消息。

定時(shí)消息與延時(shí)消息在代碼配置上存在一些差異，但是最終達(dá)到的效果相同：消息在發(fā)送到消息隊(duì)列RocketMQ版服務(wù)端后并不會(huì)立馬投遞，而是根據(jù)消息中的屬性延遲固定時(shí)間后才投遞給消費(fèi)者。

適用場(chǎng)景

定時(shí)消息和延時(shí)消息適用于以下一些場(chǎng)景：

消息生產(chǎn)和消費(fèi)有時(shí)間窗口要求，例如在電商交易中超時(shí)未支付關(guān)閉訂單的場(chǎng)景，在訂單創(chuàng)建時(shí)會(huì)發(fā)送一條延時(shí)消息。

這條消息將會(huì)在30分鐘以后投遞給消費(fèi)者，消費(fèi)者收到此消息后需要判斷對(duì)應(yīng)的訂單是否已完成支付。

如支付未完成，則關(guān)閉訂單。如已完成支付則忽略。

通過(guò)消息觸發(fā)一些定時(shí)任務(wù)，例如在某一固定時(shí)間點(diǎn)向用戶發(fā)送提醒消息。

使用方式定時(shí)消息和延時(shí)消息的使用在代碼編寫(xiě)上存在略微的區(qū)別：

發(fā)送定時(shí)消息需要明確指定消息發(fā)送時(shí)間點(diǎn)之后的某一時(shí)間點(diǎn)作為消息投遞的時(shí)間點(diǎn)。

發(fā)送延時(shí)消息時(shí)需要設(shè)定一個(gè)延時(shí)時(shí)間長(zhǎng)度，消息將從當(dāng)前發(fā)送時(shí)間點(diǎn)開(kāi)始延遲固定時(shí)間之后才開(kāi)始投遞。

注意事項(xiàng)

定時(shí)消息的精度會(huì)有1s~2s的延遲誤差。

定時(shí)和延時(shí)消息的msg.setStartDeliverTime參數(shù)需要設(shè)置成當(dāng)前時(shí)間戳之后的某個(gè)時(shí)刻（單位毫秒）。

如果被設(shè)置成當(dāng)前時(shí)間戳之前的某個(gè)時(shí)刻，消息將立刻投遞給消費(fèi)者。

定時(shí)和延時(shí)消息的msg.setStartDeliverTime參數(shù)可設(shè)置40天內(nèi)的任何時(shí)刻（單位毫秒），超過(guò)40天消息發(fā)送將失敗。

StartDeliverTime是服務(wù)端開(kāi)始向消費(fèi)端投遞的時(shí)間。如果消費(fèi)者當(dāng)前有消息堆積，那么定時(shí)和延時(shí)消息會(huì)排在堆積消息后面，將不能?chē)?yán)格按照配置的時(shí)間進(jìn)行投遞。

由于客戶端和服務(wù)端可能存在時(shí)間差，消息的實(shí)際投遞時(shí)間與客戶端設(shè)置的投遞時(shí)間之間可能存在偏差。

如何使用

推薦使用阿里云提供的rocketmq版本的pom

       <dependency>
            <groupId>com.aliyun.openservicesgroupId>
            <artifactId>ons-clientartifactId>
            <version>1.8.4.Finalversion>
        dependency>

消息發(fā)送

import com.aliyun.openservices.ons.api.*;
import com.aliyun.openservices.shade.org.apache.commons.lang3.time.DateFormatUtils;

import java.util.Date;
import java.util.Properties;

public class ProducerDelayTest {
    public static void main(String[] args) {
        Properties properties = new Properties();
        // AccessKey ID阿里云身份驗(yàn)證，在阿里云RAM控制臺(tái)創(chuàng)建。
        properties.put(PropertyKeyConst.AccessKey, "XXX");
        // AccessKey Secret阿里云身份驗(yàn)證，在阿里云RAM控制臺(tái)創(chuàng)建。
        properties.put(PropertyKeyConst.SecretKey, "XXX");
        // 設(shè)置TCP接入域名，進(jìn)入消息隊(duì)列RocketMQ版控制臺(tái)實(shí)例詳情頁(yè)面的接入點(diǎn)區(qū)域查看。
        properties.put(PropertyKeyConst.NAMESRV_ADDR, "localhost:9876");
        

        Producer producer = ONSFactory.createProducer(properties);
        // 在發(fā)送消息前，必須調(diào)用start方法來(lái)啟動(dòng)Producer，只需調(diào)用一次即可。
        producer.start();

         {
            Message msg = new Message(
                    // 您在消息隊(duì)列RocketMQ版控制臺(tái)創(chuàng)建的Topic。
                    "TopicTest",
                    // Message Tag,可理解為Gmail中的標(biāo)簽，對(duì)消息進(jìn)行再歸類(lèi)，方便Consumer指定過(guò)濾條件在消息隊(duì)列RocketMQ版服務(wù)器過(guò)濾。
                    "TagA",
                    // Message Body可以是任何二進(jìn)制形式的數(shù)據(jù)，消息隊(duì)列RocketMQ版不做任何干預(yù)，需要Producer與Consumer協(xié)商好一致的序列化和反序列化方式。
                    "演示15秒鐘>>> ".getBytes());
            // 設(shè)置代表消息的業(yè)務(wù)關(guān)鍵屬性，請(qǐng)盡可能全局唯一。
            // 以方便您在無(wú)法正常收到消息情況下，可通過(guò)控制臺(tái)查詢(xún)消息并補(bǔ)發(fā)。
            // 注意：不設(shè)置也不會(huì)影響消息正常收發(fā)。
            msg.setKey("ORDERID_100e");
            try {
                // 延時(shí)消息，單位毫秒（ms），在指定延遲時(shí)間（當(dāng)前時(shí)間之后）進(jìn)行投遞，例如消息在15秒后投遞。
                long delayTime = System.currentTimeMillis() + 15000;
                System.out.println("發(fā)送時(shí)間>>" + DateFormatUtils.format(new Date(), "yyyy-MM-dd HH