2. 程式人生 > >延時訊息佇列

延時訊息佇列

阿新 • • 發佈：2018-11-24

在這裡插入圖片描述
下面程式碼按需要填寫
@Bean
public Queue delayQueuePerMessageTTL() {
Map<String, Object> argument = new HashMap<>();
argument.put(“x-message-ttl”, 1000 * 5);//How long a message published to a queue can live before it is discarded (milliseconds).(Sets the “x-message-ttl” argument.)
argument.put(“x-expires”, 100060);//How long a queue can be unused for before it is automatically deleted (milliseconds).(Sets the “x-expires” argument.)
argument.put(“x-max-length”, 10000);//How many (ready) messages a queue can contain before it starts to drop them from its head.(Sets the “x-max-length” argument.)
argument.put(“x-max-length-bytes”, 1024

1024*10);//Total body size for ready messages a queue can contain before it starts to drop them from its head.(Sets the “x-max-length-bytes” argument.)
argument.put(“x-dead-letter-exchange”, “exchangeName”);//Optional name of an exchange to which messages will be republished if they are rejected or expire.(Sets the “x-dead-letter-exchange” argument.)
argument.put(“x-dead-letter-routing-key”, “routing-key”);//Optional replacement routing key to use when a message is dead-lettered. If this is not set, the message’s original routing key will be used.(Sets the “x-dead-letter-routing-key” argument.)
argument.put(“x-max-priority”, 100);//Maximum number of priority levels for the queue to support; if not set, the queue will not support message priorities.(Sets the “x-max-priority” argument.)
argument.put(“x-queue-mode”, “lazy”);//Set the queue into lazy mode, keeping as many messages as possible on disk to reduce RAM usage; if not set, the queue will keep an in-memory cache to deliver messages as fast as possible.(Sets the “x-queue-mode” argument.)
argument.put(“x-queue-master-locator”, “”);//Set the queue into master location mode, determining the rule by which the queue master is located when declared on a cluster of nodes.(Sets the “x-queue-master-locator” argument.)
return QueueBuilder.durable(“deladyQueues”).withArguments(argument).autoDelete().exclusive().build();
}
在實際的業務中我們會遇見生產者產生的訊息，不立即消費，而是延時一段時間在消費。RabbitMQ本身沒有直接支援延遲佇列功能，但是我們可以根據其特性Per-Queue Message TTL和 Dead Letter Exchanges實現延時佇列。也可以通過改特性設定訊息的優先順序。

1.Per-Queue Message TTL
RabbitMQ可以針對訊息和佇列設定TTL（過期時間）。佇列中的訊息過期時間(Time To Live, TTL)有兩種方法可以設定。第一種方法是通過佇列屬性設定，佇列中所有訊息都有相同的過期時間。第二種方法是對訊息進行單獨設定，每條訊息TTL可以不同。如果上述兩種方法同時使用，則訊息的過期時間以兩者之間TTL較小的那個數值為準。訊息在佇列的生存時間一旦超過設定的TTL值，就成為dead message,消費者將無法再收到該訊息。
2.Dead Letter Exchanges
當訊息在一個佇列中變成死信後，它能被重新publish到另一個Exchange。訊息變成Dead Letter一向有以下幾種情況：
訊息被拒絕（basic.reject or basic.nack）並且requeue=false
訊息TTL過期
佇列達到最大長度
實際上就是設定某個佇列的屬性，當這個佇列中有Dead Letter時，RabbitMQ就會自動的將這個訊息重新發布到設定的Exchange中去，進而被路由到另一個佇列，publish可以監聽這個佇列中訊息做相應的處理，這個特性可以彌補RabbitMQ 3.0.0以前支援的immediate引數中的向publish確認的功能。

雖然 consumer 從來看不到過期的 message ，但是在過期 message 到達 queue 的頭部時確實會被真正的丟棄（或者 dead-lettered ）。當對每一個 queue 設定了 TTL 值時不會產生任何問題，因為過期的 message 總是會出現在 queue 的頭部。當對每一條 message 設定了 TTL 時，過期的 message 可能會排隊於未過期 message 的後面，直到這些訊息被 consume 到或者過期了。在這種情況下，這些過期的 message 使用的資源將不會被釋放，且會在 queue 統計資訊中被計算進去（例如，queue 中存在的 message 的數量）。對於第一種設定佇列TTL屬性的方法，一旦訊息過期，就會從佇列中抹去，而第二種方法裡，即使訊息過期，也不會馬上從佇列中抹去，因為每條訊息是否過期時在即將投遞到消費者之前判定的，為什麼兩者得處理方法不一致？因為第一種方法裡，佇列中已過期的訊息肯定在佇列頭部，RabbitMQ只要定期從隊頭開始掃描是否有過期訊息即可，而第二種方法裡，每條訊息的過期時間不同，如果要刪除所有過期訊息，勢必要掃描整個佇列，所以不如等到此訊息即將被消費時再判定是否過期，如果過期，再進行刪除。

一、在佇列上設定TTL

1.建立delay.exchange

這裡Internal設定為NO，否則將無法接受dead letter，YES表示這個exchange不可以被client用來推送訊息，僅用來進行exchange和exchange之間的繫結。

2.建立延時佇列(delay queue)

如上配置延時5min佇列(x-message-ttl=300000)

x-max-length:最大積壓的訊息個數，可以根據自己的實際情況設定，超過限制訊息不會丟失，會立即轉向delay.exchange進行投遞

x-dead-letter-exchange:設定為剛剛配置好的delay.exchange,訊息過期後會通過delay.exchange進行投遞

這裡不需要配置"dead letter routing key"否則會覆蓋掉訊息傳送時攜帶的routingkey，導致後面無法路由為剛才配置的delay.exchange

3.配置延時路由規則

需要延時的訊息到exchange後先路由到指定的延時佇列

1）建立delaysync.exchange通過Routing key將訊息路由到延時佇列

2.配置delay.exchange 將訊息投遞到正常的消費佇列

配置完成。

下面使用程式碼測試一下：

生產者：

package cn.slimsmart.study.rabbitmq.delayqueue.queue;

import java.io.IOException;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class Producer {

private static String queue_name = "test.queue";  

public static void main(String[] args) throws IOException {  
    ConnectionFactory factory = new ConnectionFactory();  
    factory.setHost("10.1.199.169");  
    factory.setUsername("admin");  
    factory.setPassword("123456");  
    Connection connection = factory.newConnection();  
    Channel channel = connection.createChannel();  
    // 宣告佇列  
    channel.queueDeclare(queue_name, true, false, false, null);  
    String message = "hello world!" + System.currentTimeMillis();  
    channel.basicPublish("delaysync.exchange", "deal.message", null, message.getBytes());  
    System.out.println("sent message: " + message + ",date:" + System.currentTimeMillis());  
    // 關閉頻道和連線  
    channel.close();  
    connection.close();  
}

}

消費者：

package cn.slimsmart.study.rabbitmq.delayqueue.queue;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;

public class Consumer {

private static String queue_name = "test.queue";  

public static void main(String[] args) throws Exception {  
    ConnectionFactory factory = new ConnectionFactory();  
    factory.setHost("10.1.199.169");  
    factory.setUsername("admin");  
    factory.setPassword("123456");  
    Connection connection = factory.newConnection();  
    Channel channel = connection.createChannel();  
    // 宣告佇列  
    channel.queueDeclare(queue_name, true, false, false, null);  
    QueueingConsumer consumer = new QueueingConsumer(channel);  
    // 指定消費佇列  
    channel.basicConsume(queue_name, true, consumer);  
    while (true) {  
        // nextDelivery是一個阻塞方法（內部實現其實是阻塞佇列的take方法）  
        QueueingConsumer.Delivery delivery = consumer.nextDelivery();  
        String message = new String(delivery.getBody());  
        System.out.println("received message:" + message + ",date:" + System.currentTimeMillis());  
    }  
}

}
二、在訊息上設定TTL

實現程式碼：

生產者：

package cn.slimsmart.study.rabbitmq.delayqueue.message;

import java.io.IOException;
import java.util.HashMap;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class Producer {

private static String queue_name = "message_ttl_queue";  

public static void main(String[] args) throws IOException {  
    ConnectionFactory factory = new ConnectionFactory();  
    factory.setHost("10.1.199.169");  
    factory.setUsername("admin");  
    factory.setPassword("123456");  
    Connection connection = factory.newConnection();  
    Channel channel = connection.createChannel();  
    HashMap<String, Object> arguments = new HashMap<String, Object>();  
    arguments.put("x-dead-letter-exchange", "amq.direct");  
    arguments.put("x-dead-letter-routing-key", "message_ttl_routingKey");  
    channel.queueDeclare("delay_queue", true, false, false, arguments);  

    // 宣告佇列  
    channel.queueDeclare(queue_name, true, false, false, null);  
    // 繫結路由  
    channel.queueBind(queue_name, "amq.direct", "message_ttl_routingKey");  

    String message = "hello world!" + System.currentTimeMillis();  
    // 設定延時屬性  
    AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();  
    // 永續性 non-persistent (1) or persistent (2)  
    AMQP.BasicProperties properties = builder.expiration("300000").deliveryMode(2).build();  
    // routingKey =delay_queue 進行轉發  
    channel.basicPublish("", "delay_queue", properties, message.getBytes());  
    System.out.println("sent message: " + message + ",date:" + System.currentTimeMillis());  
    // 關閉頻道和連線  
    channel.close();  
    connection.close();  
}

}

消費者：

package cn.slimsmart.study.rabbitmq.delayqueue.message;

import java.util.HashMap;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;

public class Consumer {

private static String queue_name = "message_ttl_queue";  

public static void main(String[] args) throws Exception {  
    ConnectionFactory factory = new ConnectionFactory();  
    factory.setHost("10.1.199.169");  
    factory.setUsername("admin");  
    factory.setPassword("123456");  
    Connection connection = factory.newConnection();  
    Channel channel = connection.createChannel();  
    HashMap<String, Object> arguments = new HashMap<String, Object>();  
    arguments.put("x-dead-letter-exchange", "amq.direct");  
    arguments.put("x-dead-letter-routing-key", "message_ttl_routingKey");  
    channel.queueDeclare("delay_queue", true, false, false, arguments);  

    // 宣告佇列  
    channel.queueDeclare(queue_name, true, false, false, null);  
    // 繫結路由  
    channel.queueBind(queue_name, "amq.direct", "message_ttl_routingKey");  

    QueueingConsumer consumer = new QueueingConsumer(channel);  
    // 指定消費佇列  
    channel.basicConsume(queue_name, true, consumer);  
    while (true) {  
        // nextDelivery是一個阻塞方法（內部實現其實是阻塞佇列的take方法）  
        QueueingConsumer.Delivery delivery = consumer.nextDelivery();  
        String message = new String(delivery.getBody());  
        System.out.println("received message:" + message + ",date:" + System.currentTimeMillis());  
    }  
}

}

spring-rabbit整合教程

maven依賴：

    <dependency>
        <groupId>org.springframework.amqp</groupId>
        <artifactId>spring-rabbit</artifactId>
        <version>1.4.6.RELEASE</version>
    </dependency>

spring配置檔案（在檔案頭部引入rabbit的名稱空間和約束檔案）：

<?xml version="1.0" encoding="UTF-8"?> 

<!-- 定義Rabbit,指定連線工廠 -->
<rabbit:connection-factory id="connectionFactory"  host="你的rabbitMQ服務的ip"  virtual-host="/vhost名稱"  username="使用者名稱"  password="密碼"  port="5672"  />      

<!-- MQ的管理，包括佇列、交換器等 -->
<rabbit:admin connection-factory="connectionFactory"/>

<!-- 定義Rabbit模板，指定連線工廠以及定義exchange -->
<rabbit:template id="amqpTemplate" exchange="my_exchange" connection-factory="connectionFactory"  />	

<!-- queue 佇列宣告 --> 
<!--    durable 是否持久化 ,exclusive 僅建立者可以使用的私有佇列，斷開後自動刪除 ,auto-delete 當所有消費端連線斷開後，是否自動刪除佇列   -->
<rabbit:queue name="my_queue" durable="true" auto-delete="false" exclusive="false"/>

<!-- 交換機定義 -->
<!-- direct-exchange 模式:訊息與一個特定的路由器完全匹配，才會轉發; topic-exchange 模式:按規則轉發訊息，最靈活 -->
<rabbit:topic-exchange name="my_exchange" durable="true" auto-delete="false">
	    <rabbit:bindings>
	        <!-- 設定訊息Queue匹配的pattern (direct模式為key) -->
	        <rabbit:binding  queue="my_queue"  pattern="my_patt"/>
	    </rabbit:bindings>
</rabbit:topic-exchange>


<!-- 引入消費者 -->
<bean id="rabbitmqService" class="com.group.service.RabbitmqService"  />
    
<!-- 配置監聽 消費者   acknowledeg = manual,auto,none   -->
<rabbit:listener-container connection-factory="connectionFactory" acknowledge="auto" >
    <!-- queues 監聽佇列，多個用逗號分隔; ref 監聽器 -->
    <rabbit:listener queue-names="my_queue"  ref="rabbitmqService"  method="test"/>
</rabbit:listener-container>

那麼在專案中裝配amqpTemplate中就可以傳送訊息了

作者：MC-閏土
來源：CSDN
原文：https://blog.csdn.net/qq_22075041/article/details/78885113
版權宣告：本文為博主原創文章，轉載請附上博文連結！

相關推薦

延時訊息佇列

下面程式碼按需要填寫 @Bean public Queue delayQueuePerMessageTTL() { Map<String, Object> argument = new HashMap<>(); argument.put(“x-message-ttl

Redis延時訊息佇列、非同步訊息佇列的實現

package list; import java.lang.reflect.Type; import java.util.Set; import java.util.UUID; import com.alibaba.fastjson.JSON; import com.a

玩轉redis-延時訊息佇列

上一篇基於redis的list實現了一個簡單的訊息佇列：玩轉redis-簡單訊息佇列 原始碼地址 使用demo 產品經理經常說的一句話，我們不光要有X功能，還要Y功能，這樣客戶才能更滿意。同樣的，只有簡單訊息佇列是不夠的，還要有延時訊息佇列才能算是一個完整的訊息佇列。 看看redis的命令，放眼望去，的有序

Spring boot實戰專案整合阿里雲RocketMQ （非開源版）訊息佇列實現傳送普通訊息，延時訊息 --附程式碼

一.為什麼選擇RocketMQ訊息佇列？ 首先RocketMQ是阿里巴巴自研出來的，也已開源。其效能和穩定性從雙11就能看出來，借用阿里的一句官方介紹：歷年雙 11 購物狂歡節零點千萬級 TPS、萬億級資料洪峰，創造了全球最大的業務訊息併發以及流轉紀錄（日誌類訊息除外）；  在始終保證高效能前提下

RabbitMQ 延時訊息設計

 問題背景 所謂"延時訊息"是指當訊息被髮送以後，並不想讓消費者立即拿到訊息，而是等待指定時間後，消費者才拿到這個訊息進行消費。 場景一：客戶A在十二點下了一個訂單，我想半個小時後來檢查一下這個訂單的付款狀態，根據付款狀態來作下一步的處理。 a. 針對場景一，建議採用方案資料庫儲存+sc

Springboot2（28）整合rabbitmq實現延時訊息

原始碼地址 springboot2教程系列 rabbitmq實現訊息的確認機制和延時訊息的傳送 訊息生產者程式碼實現的主要配置 @Configuration @Slf4j public class PrividerRabbitm

RocketMQ原理學習--延時訊息實現原理

         RocketMQ提供了延時訊息型別，簡單來說就是生產者在傳送訊息的時候指定一個延時時間，當到達延時時間之後訊息才能夠被投送到消費者。       首先我們可以考慮一下RocketMQ的延時訊息是

rabbitmq發延時訊息以及通過一個exchange發到不同的queue

public static void main(String[] args) throws Exception { producer(1); producer(2); producer(3); } private static void

RocketMQ延時訊息的使用和延時級別的配置

RocketMQ 支援定時訊息，但是不支援任意時間精度，僅支援特定的 level，例如定時 5s， 10s， 1m 等。 其中，level=0 級表示不延時，level=1 表示 1 級延時，level=2 表示 2 級延時，以此類推。 如果要支援任意的時間精度，在Broker層面，必須要做訊息排序，

RocketMQ延時訊息

一個延時訊息被髮出到消費成功經歷以下幾個過程：設定訊息的延時級別delayLevelproducer傳送訊息broker收到訊息在準備將訊息寫入儲存的時候，判斷是延時訊息則更改Message的topic為延時訊息佇列的topic，也就是將訊息投遞到延時訊息佇列有定時任務從延時

阿里MQ普通+順序+延時訊息 整合Spring

前言 由於公司專案需要，研究了下AliWareMQ。阿里mq的普通訊息和延時訊息還是挺簡單的。不過在順序訊息的時候出現了一些瓶頸。後來查閱原始碼和依據demo整理了一版融合Spring的版本。 例項 mq配置檔案(Spring) 主要是順序訊息的

RocketMQ學習(七)——RocketMQ延時訊息

使用者傳送的訊息也可以指定延時時間（更準確的說是延時等級），然後在指定延時時間之後投遞訊息，然後被consumer消費。阿里雲的ons還支援定時訊息，而且延時訊息是直接指定延時時間，其實阿里雲的延時訊息也是定時訊息的另一種表述方式，都是通過設定訊息被投遞的時間

Java關於DelayQueue做延時訊息推送

最近比較閒，看某專案原始碼時看到有用DelayQueue類來做延時的訊息推送。 DelayQueue是Delayed元素的一個無界阻塞佇列，只有在延遲期滿時才能從中提取元素。該佇列的頭部 是延遲期滿後儲存時間最長的Delayed元素。如果延遲都還沒有期滿，則佇列沒有頭部，並

手把手實現一條延時訊息

前言 近期在維護公司的排程平臺，其中有個關鍵功能那就是定時任務；定時任務大家平時肯定接觸的不少，比如 JDK 中的 Timer、ScheduledExecutorService、排程框架 Quartz 等。 通常用於實現 XX 時間後的延時任務，或週期性任務； 比如一個常見的業務場景：使用者下單 N 分鐘

Redis 非同步訊息佇列與延時佇列

        訊息中介軟體，大家都會想到  Rabbitmq 和 Kafka 作為訊息佇列中介軟體，來給應用程式之間增加非同步訊息傳遞功能。這兩個中介軟體都是專業的訊息佇列中介軟體，特性之多超出了大多數人的理解能力。但是這種屬於重量級的應

效能提升五十倍：訊息佇列延時聚合通知的重要性

前言 這個話題對我而言，是影響很久的事情。從第一次使用訊息佇列開始，業務背景是報名系統通知到我們的系統。正常流量下資料都能正常通知過來，但遇到匯入報名人時，採用了Task非同步通知，資料量一大，佇列就死了。當時是儘量採用同步方式，減少併發量。  後來業務上有了專門的營銷系統

Redis實現可靠低延時的訊息佇列

不過因為使用的是decrBy會導致一種情況出現，當前庫存還剩1個，2個執行緒同時請求，一個請求減去1個庫存，另一個請求減去2個庫存，，如果減去2個庫存的先執行，他會返回一個-1，然後我會加回去，但是在加回去的之前，減1庫存的執行緒執行了，會返回-2，依然沒有辦法減庫存成功，所以在這中情況下，我採用當減庫存返回

如何解決訊息佇列的延時以及過期失效問題？訊息佇列滿了以後該怎麼處理？有幾百萬訊息持續積壓幾小時，說說怎麼解決？

開發十年，就只剩下這套架構體系了！ >>>   

RabbitMQ高階之訊息限流與延時佇列

>人生終將是場單人旅途，孤獨之前是迷茫，孤獨過後是成長。 ## 楔子 本篇是訊息佇列`RabbitMQ`的第五彈。 上篇本來打算講述`RabbitMQ`的一些高階用法： * 如何保證訊息的可靠性？ * 訊息佇列如何進行限流？ * 如何設定延時佇列進行延時消費？ 最終因為篇幅緣故，上篇只講了`

php訂單延時處理-延時佇列

延遲佇列，顧名思義它是一種帶有延遲功能的訊息佇列。 那麼，是在什麼場景下我才需要這樣的佇列呢？ 一、背景 先看看一下業務場景： 1.會員過期前3天傳送召回通知 2.訂單支付成功後，5分鐘後檢測下游環節是否都正常，比如使用者購買會員後，各種會員狀態是否都設定成功 3.如何定期檢查處於退款狀態