電子發燒友網>電子資料下載>電子資料>delaymq高性能的任意延時消息隊列

delaymq高性能的任意延時消息隊列

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

普通下載普通下載

資料介紹

授權協議 GPL

開發語言 Java

操作系統跨平臺

軟件類型開源軟件

所屬分類服務器軟件、 JMS/消息中間件

軟件簡介

RocketMQ 開源版本任意時間延時隊列實現

定時消息：Producer將消息發送到消息隊列RocketMQ版服務端，但并不期望立馬投遞這條消息，而是推遲到在當前時間點之后的某一個時間投遞到Consumer進行消費，該消息即定時消息。

延時消息：Producer將消息發送到消息隊列RocketMQ版服務端，但并不期望立馬投遞這條消息，而是延遲一定時間后才投遞到Consumer進行消費，該消息即延時消息。

定時消息與延時消息在代碼配置上存在一些差異，但是最終達到的效果相同：消息在發送到消息隊列RocketMQ版服務端后并不會立馬投遞，而是根據消息中的屬性延遲固定時間后才投遞給消費者。

適用場景

定時消息和延時消息適用于以下一些場景：

消息生產和消費有時間窗口要求，例如在電商交易中超時未支付關閉訂單的場景，在訂單創建時會發送一條延時消息。

這條消息將會在30分鐘以后投遞給消費者，消費者收到此消息后需要判斷對應的訂單是否已完成支付。

如支付未完成，則關閉訂單。如已完成支付則忽略。

通過消息觸發一些定時任務，例如在某一固定時間點向用戶發送提醒消息。

使用方式定時消息和延時消息的使用在代碼編寫上存在略微的區別：

發送定時消息需要明確指定消息發送時間點之后的某一時間點作為消息投遞的時間點。

發送延時消息時需要設定一個延時時間長度，消息將從當前發送時間點開始延遲固定時間之后才開始投遞。

注意事項

定時消息的精度會有1s~2s的延遲誤差。

定時和延時消息的msg.setStartDeliverTime參數需要設置成當前時間戳之后的某個時刻（單位毫秒）。

如果被設置成當前時間戳之前的某個時刻，消息將立刻投遞給消費者。

定時和延時消息的msg.setStartDeliverTime參數可設置40天內的任何時刻（單位毫秒），超過40天消息發送將失敗。

StartDeliverTime是服務端開始向消費端投遞的時間。如果消費者當前有消息堆積，那么定時和延時消息會排在堆積消息后面，將不能嚴格按照配置的時間進行投遞。

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

如何使用

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

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

消息發送

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阿里云身份驗證，在阿里云RAM控制臺創建。
        properties.put(PropertyKeyConst.AccessKey, "XXX");
        // AccessKey Secret阿里云身份驗證，在阿里云RAM控制臺創建。
        properties.put(PropertyKeyConst.SecretKey, "XXX");
        // 設置TCP接入域名，進入消息隊列RocketMQ版控制臺實例詳情頁面的接入點區域查看。
        properties.put(PropertyKeyConst.NAMESRV_ADDR, "localhost:9876");
        

        Producer producer = ONSFactory.createProducer(properties);
        // 在發送消息前，必須調用start方法來啟動Producer，只需調用一次即可。
        producer.start();

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