2. 程式人生 > >java實現rabbitMQ延時佇列詳解以及spring-rabbit整合教程

java實現rabbitMQ延時佇列詳解以及spring-rabbit整合教程

阿新 • • 發佈：2019-01-25

java實現rabbitMQ延時佇列詳解

這是我在公司開發中使用的倆套方案，感興趣的話可以看一下：點選下載

在實際的業務中我們會遇見生產者產生的訊息，不立即消費，而是延時一段時間在消費。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"?>
<beans xmlns="http://www.springframework.org/schema/beans"
	xmlns:aop="http://www.springframework.org/schema/aop"
	xmlns:context="http://www.springframework.org/schema/context"
	xmlns:mvc="http://www.springframework.org/schema/mvc"
	xmlns:task="http://www.springframework.org/schema/task"
	xmlns:tx="http://www.springframework.org/schema/tx"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xmlns:rabbit="http://www.springframework.org/schema/rabbit"
	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
		http://www.springframework.org/schema/mvc http://www.springframework.org/schema/mvc/spring-mvc.xsd
		http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop.xsd
		http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx.xsd
		http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit.xsd
		http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.0.xsd">
	
	<!-- 定義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>
    
</beans>

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

相關推薦

java實現rabbitMQ延時佇列詳解以及spring-rabbit整合教程

java實現rabbitMQ延時佇列詳解 這是我在公司開發中使用的倆套方案，感興趣的話可以看一下：點選下載 在實際的業務中我們會遇見生產者產生的訊息，不立即消費，而是延時一段時間在消費。RabbitMQ本身沒有直接支援延遲佇列功能，但是我們可以根據其特性Per-Queu

【小家java】BlockingQueue阻塞佇列詳解以及5大實現（ArrayBlockingQueue、DelayQueue、LinkedBlockingQueue...）

相關閱讀 【小家java】java5新特性（簡述十大新特性） 重要一躍 【小家java】java6新特性（簡述十大新特性） 雞肋升級 【小家java】java7新特性（簡述八大新特性） 不溫不火 【小家java】java8新特性（簡述十大新特性） 飽受讚譽 【小家java】java9

Java多執行緒---阻塞佇列詳解（舉例說明）

一. 前言 　　在新增的Concurrent包中，BlockingQueue很好的解決了多執行緒中，如何高效安全“傳輸”資料的問題。通過這些高效並且執行緒安全的佇列類，為我們快速搭建高質量的多執行緒程式帶來極大的便利。本文詳細介紹了BlockingQueue家庭中的所有成員

使用Redis實現輕量級延時佇列

A:需求說明: 如果系統中需要用到定時執行計劃的，又不想用到中介軟體，如果輪詢資料庫的話，會導致大量資源消耗，這樣我們就可以使用Redis來實現類似功（需要使用rabbitMQ的請看這裡：https://blog.csdn.net/u010096717/article/de

SpringBoot使用RabbitMQ延時佇列

延時佇列延時佇列的使用場景：1.訂單業務：在電商中，使用者下單後30分鐘後未付款則取消訂單。2.簡訊通知：使用者下單並付款後，1分鐘後發簡訊給使用者。延時佇列實現思路AMQP協議和RabbitMQ佇列本身沒有直接支援延遲佇列功能，但是我們可以通過RabbitMQ的兩個特性來曲

Docker下RabbitMQ延時佇列實戰兩部曲之一：極速體驗

有的應用場景中，向RabbitMQ發出訊息後，我們希望消費方不要立即消費，可以通過延時佇列來實現，思路是將訊息傳送到A佇列，此佇列沒有消費者，等訊息過期後會進入A佇列的Dead Letter Exchange中，B佇列綁定了這個Dead Letter Excha

Docker下RabbitMQ延時佇列實戰兩部曲之二：細說開發

本章是《Docker下RabbitMQ延時佇列實戰兩部曲》的終篇，上一章《Docker下RabbitMQ延時佇列實戰兩部曲之一：極速體驗》我們快速體驗了延時佇列的生產和消費，今天來實戰整個開發過程； SpringBoot框架下進行RabbitMQ開發，相關

資料結構與算法系列2 線性表 使用java實現動態陣列+ArrayList原始碼詳解

## 資料結構與算法系列2 線性表 使用java實現動態陣列+ArrayList原始碼詳解 對陣列有不瞭解的可以先看看我的另一篇文章，那篇文章對陣列有很多詳細的解析，而本篇文章則著重講動態陣列，另一篇文章連結如下，可點選跳轉： 連結：[https://blog.csdn.net/pjh88/article/d

JAVA中常見的阻塞佇列詳解

![](https://img2020.cnblogs.com/other/2024393/202011/2024393-20201116085210744-2073386236.jpg) * 在之前的執行緒池的介紹中我們看到了很多阻塞佇列，這篇文章我們主要來說說阻塞佇列的事。 * 阻塞佇列也就是 `Blo

Redis的批量操作是什麼？怎麼實現的延時佇列？以及訂閱模式、LRU。

## 前言 這次的內容是我自己為了總結Redis知識而擴充的，上一篇其實已經總結了幾點知識了，但是Redis的強大，以及適用範圍之廣可不是單單一篇博文就能總結清的。所以這次準備繼續總結，因為第一個問題，Redis的批量操作，是我在面試過程中被真實問到的，當時沒答上來，也是因為確實沒了解過Redis的批量操作。

RabbitMQ詳解以及和SpringBoot整合

各訊息元件的適用場景 ActiveMQ ActiveMQ 是一款比較早期的訊息元件，由Apache開源出來的，它能滿足吞吐量一般的業務場景，但是對於高併發場景，效能較差。 Kafka Kafka追求高吞吐量的特性，它一開始使用於日誌的收集。缺點是訊息可靠性支援

java環境變數 的配置與詳解（全網最詳細教程）

筆者這學期開始學習java課程，學習java開發首先需要配置java執行環境變數。雖然上課老師也講了如何配置java環境變數，可是筆者的同學還是有好多都不會配置，所以筆者最近配置了特別多次java環境變數。如下筆者詳細解釋從JDK安裝到環境變數的裝配。 目錄

Java中NullPointerException異常的原因詳解以及解決方法

NullPointerException是當您嘗試使用指向記憶體中空位置的引用（null）時發生的異常，就好像它引用了一個物件一樣。 當我們宣告引用變數（即物件）時，實際上是在建立指向物件的指標。考慮以下程式碼，您可以在其中宣告基本型別的整型變數x： int x; x = 10; 在

Java 使用RabbitMQ外掛實現延時佇列

Springboot專案，windows環境 環境配置 在rabbitmq 3.5.7及以上的版本提供了一個外掛（rabbitmq-delayed-message-exchange）來實現延遲佇列功能。同時外掛依賴Erlang/OPT 18.0及以上。 外掛下載地址： http

rabbitmq實現延時佇列（死信佇列）

基於佇列和基於訊息的TTL TTL是time to live 的簡稱，顧名思義指的是訊息的存活時間。rabbitMq可以從兩種維度設定訊息過期時間，分別是佇列和訊息本身。 佇列訊息過期時間-Per-Queue Message TTL： 通過設定佇列的x-message-ttl引數來設定指定佇列上訊息的存活時

Springboot+rabbitmq實現延時佇列的兩種方式

什麼是延時佇列，延時佇列應用於什麼場景 延時佇列顧名思義，即放置在該佇列裡面的訊息是不需要立即消費的，而是等待一段時間之後取出消費。 那麼，為什麼需要延遲消費呢？我們來看以下的場景 網上商城下訂單後30分鐘後沒有完成支付，取消訂單(如：淘寶、去哪兒網) 系統

BFS和DFS詳解以及java實現(轉載)

作者： Leo-Yang 原文都先發布在作者個人部落格： http://www.leoyang.net/ 本文版權歸作者和部落格園共有，歡迎轉載，但未經作者同意必須保留此段宣告，且在文章頁面明顯位置給出原文連線，否則保留追究法律責任的權利. 前言

Java佇列詳解之 LinkedList 類

Java佇列詳解之 LinkedList 類 1. 類簡介 類釋義 A collection designed for holding elements prior to processing. Besides basic Collection oper

設計模式（建立型）：Java常用23種設計模式之單例模式詳解以及Java程式碼實現

可以說單例模式是所有設計模式中最簡單的一種。 單例模式就是說系統中對於某類的只能有一個物件，不可能出來第二個。 單例模式也是23中設計模式中在面試時少數幾個會要求寫程式碼的模式之一。主要考察的是多執行緒下面單例模式的執行緒安全性問題。 1.多執行緒安全單例模式例項一(不使用同步鎖)

Java 進階——多執行緒優化之執行緒池 ThreadPoolExecutor的核心容器阻塞佇列詳解（一）

#引言 多執行緒我想無論是後端開發，還是對於App開發者來說都不會陌生，何況Android強制要求不能在主執行緒中做網路請求，於是乎，在很多初學者或者App的原始碼中會出現會多的new Thread…的方式，這樣的程式碼是不優雅而且存在很多的隱患，假如說在使用者