2. 程式人生 > >Springboot+rabbitmq實現延時佇列的兩種方式

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

阿新 • • 發佈：2019-02-04

什麼是延時佇列，延時佇列應用於什麼場景

延時佇列顧名思義，即放置在該佇列裡面的訊息是不需要立即消費的，而是等待一段時間之後取出消費。
那麼，為什麼需要延遲消費呢？我們來看以下的場景

  1. 網上商城下訂單後30分鐘後沒有完成支付，取消訂單(如：淘寶、去哪兒網)
  2. 系統建立了預約之後，需要在預約時間到達前一小時提醒被預約的雙方參會
  3. 系統中的業務失敗之後，需要重試

這些場景都非常常見，我們可以思考，比如第二個需求，系統建立了預約之後，需要在預約時間到達前一小時提醒被預約的雙方參會。那麼一天之中肯定是會有很多個預約的，時間也是不一定的，假設現在有1點 2點 3點 三個預約，如何讓系統知道在當前時間等於0點 1點 2點給使用者傳送資訊呢，是不是需要一個輪詢，一直去檢視所有的預約，比對當前的系統時間和預約提前一小時的時間是否相等呢？這樣做非常浪費資源而且輪詢的時間間隔不好控制。如果我們使用延時訊息佇列呢，我們在建立時把需要通知的預約放入訊息中介軟體中，並且設定該訊息的過期時間，等過期時間到達時再取出消費即可。

Rabbitmq實現延時佇列一般而言有兩種形式：
第一種方式：利用兩個特性： Time To Live(TTL)、Dead Letter Exchanges（DLX）
第二種方式：利用rabbitmq中的外掛x-delay-message

利用TTL DLX實現延時佇列的方式

TTL DLX是什麼

  1. TTL
    RabbitMQ可以針對佇列設定x-expires(則佇列中所有的訊息都有相同的過期時間)或者針對Message設定x-message-ttl(對訊息進行單獨設定，每條訊息TTL可以不同)，來控制訊息的生存時間，如果超時(兩者同時設定以最先到期的時間為準)，則訊息變為dead letter(死信)

  2. Dead Letter Exchanges（DLX）
    RabbitMQ的Queue可以配置x-dead-letter-exchange和x-dead-letter-routing-key（可選）兩個引數，如果佇列內出現了dead letter，則按照這兩個引數重新路由轉發到指定的佇列。
    x-dead-letter-exchange：出現dead letter之後將dead letter重新發送到指定exchange
    x-dead-letter-routing-key：出現dead letter之後將dead letter重新按照指定的routing-key傳送

Springboot整合rabbitmq實現第一種方式

在pom.xml檔案中增加rabbitmq的依賴

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

初始化queue exchange和queue及exchange之間的binding關係
Config.java

package com.example.demo.deadLetter;
import java.util.HashMap;
import java.util.Map;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import com.example.demo.Constants.Constants;

@Configuration
public class Config {

    // 建立一個立即消費佇列
    @Bean
    public Queue immediateQueue() {
        // 第一個引數是建立的queue的名字，第二個引數是是否支援持久化
        return new Queue(Constants.IMMEDIATE_QUEUE, true);
    }

    // 建立一個延時佇列
    @Bean
    public Queue delayQueue() {
        Map<String, Object> params = new HashMap<>();
        // x-dead-letter-exchange 聲明瞭佇列裡的死信轉發到的DLX名稱，
        params.put("x-dead-letter-exchange", Constants.IMMEDIATE_EXCHANGE);
        // x-dead-letter-routing-key 聲明瞭這些死信在轉發時攜帶的 routing-key 名稱。
        params.put("x-dead-letter-routing-key", Constants.IMMEDIATE_ROUTING_KEY);
        return new Queue(Constants.DELAY_QUEUE, true, false, false, params);
    }

    @Bean
    public DirectExchange immediateExchange() {
        // 一共有三種構造方法，可以只傳exchange的名字， 第二種，可以傳exchange名字，是否支援持久化，是否可以自動刪除，
        //第三種在第二種引數上可以增加Map，Map中可以存放自定義exchange中的引數
        return new DirectExchange(Constants.IMMEDIATE_EXCHANGE, true, false);
    }

    @Bean
    public DirectExchange deadLetterExchange() {
        // 一共有三種構造方法，可以只傳exchange的名字， 第二種，可以傳exchange名字，是否支援持久化，是否可以自動刪除，
        //第三種在第二種引數上可以增加Map，Map中可以存放自定義exchange中的引數
        return new DirectExchange(Constants.DEAD_LETTER_EXCHANGE, true, false);
    }

    @Bean
    //把立即消費的佇列和立即消費的exchange繫結在一起
    public Binding immediateBinding() {
        return BindingBuilder.bind(immediateQueue()).to(immediateExchange()).with(Constants.IMMEDIATE_ROUTING_KEY);
    }

    @Bean
    //把立即消費的佇列和立即消費的exchange繫結在一起
    public Binding delayBinding() {
        return BindingBuilder.bind(delayQueue()).to(deadLetterExchange()).with(Constants.DELAY_ROUTING_KEY);
    }
}

生產者生產訊息
ImmediateSender.java

package com.example.demo.deadLetter;

import java.text.SimpleDateFormat;
import java.util.Date;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.example.demo.Constants.Constants;
import com.example.demo.model.Booking;

@Component
public class ImmediateSender {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void send(Booking booking, int delayTime) {
        System.out.println("delayTime" + delayTime);
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        this.rabbitTemplate.convertAndSend(Constants.DEAD_LETTER_EXCHANGE, Constants.DELAY_ROUTING_KEY, booking, message -> {
            message.getMessageProperties().setExpiration(delayTime + "");
            System.out.println(sdf.format(new Date()) + " Delay sent.");
            return message;
        });
    }
}

消費者消費訊息
ImmediateReceiver.java

package com.example.demo.deadLetter;

import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;

import com.example.demo.Constants.Constants;
import com.example.demo.model.Booking;

@Component
@EnableRabbit
@Configuration
public class ImmediateReceiver {

    @RabbitListener(queues = Constants.IMMEDIATE_QUEUE)
    @RabbitHandler
    public void get(Booking booking) {
        System.out.println("收到延時訊息了" + booking);
    }
}

model類book
Book.java

package com.example.demo.model;

import java.io.Serializable;
import java.util.Date;

public class Booking implements Serializable {

    private static final long serialVersionUID = 1L;
    private String bookingName;
    private Date bookingTime;
    private String bookingContent;
    private String operatorName;

    public Booking() {
    }

    public String getBookingName() {
        return bookingName;
    }

    public void setBookingName(String bookingName) {
        this.bookingName = bookingName;
    }

    public Date getBookingTime() {
        return bookingTime;
    }

    public void setBookingTime(Date bookingTime) {
        this.bookingTime = bookingTime;
    }

    public String getBookingContent() {
        return bookingContent;
    }

    public void setBookingContent(String bookingContent) {
        this.bookingContent = bookingContent;
    }

    public String getOperatorName() {
        return operatorName;
    }

    public void setOperatorName(String operatorName) {
        this.operatorName = operatorName;
    }

    @Override
    public String toString() {
        return super.toString();
    }
}

測試類
Test.java

package com.example.demo;

import java.util.Date;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import com.example.demo.Immediate.Sender;
import com.example.demo.deadLetter.ImmediateSender;
import com.example.demo.model.Booking;

@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitMqTestApplicationTests {

	@Autowired
	ImmediateSender immediateSender;
	
	@Test
	public void test() {
	    Booking booking = new Booking();
        booking.setBookingContent("hhaha");
        booking.setBookingName("預定房子");
        booking.setBookingTime(new Date());
        booking.setOperatorName("hellen");
	    immediateSender.send(booking, 1000);
	}
}

總結第一種方式：經過測試，我們可以發現，當我們先增加一條過期時間大(10000)的A訊息進入，之後再增加一個過期時間小的(1000)訊息B,並沒有出現想象中的B訊息先被消費，A訊息後被消費，而是出現了當10000過去的時候，AB訊息同時被消費，也就是B訊息的消費被阻塞了。

為什麼會出現這樣的現象呢？
我們知道利用TTL DLX特性實現的方式，實際上在第一個延時佇列C裡面設定了dlx，生產者生產了一條帶ttl的訊息放入了延時佇列C中，等到延時時間到了，延時佇列C中的訊息變成了死信，根據延時佇列C中設定的dlx的exchange的轉發規則，轉發到了實際消費佇列D中，當該佇列中的監聽器監聽到訊息時就會正式開始消費。那麼實際上延時佇列中的訊息也是放入佇列中的，佇列滿足先進先出，而延時大的訊息A還沒出隊，所以B訊息也不能順利出隊。

利用Rabbitmq的外掛x-delay-message實現

為了解決上面的問題，Rabbitmq實現了一個外掛x-delay-message來實現延時佇列。

x-delay-message安裝

uzip rabbitmq_delayed_message_exchange-20171215-3.6.x.zip

將外掛移到rabbitmq安裝的路徑

sudo cp -r rabbitmq_delayed_message_exchange-20171215-3.6.x.ez /usr/lib/rabbitmq/lib/rabbitmq_server-3.6.15/plugins

Enable外掛

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

windows同理

Springboot整合rabbitmq實現第二種方式

XdelayConfig.java

package com.example.demo.Xdelay;

import java.util.HashMap;
import java.util.Map;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.CustomExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import com.example.demo.Constants.Constants;

@Configuration
public class XdelayConfig {

    // 建立一個立即消費佇列
    @Bean
    public Queue immediateQueue() {
        // 第一個引數是建立的queue的名字，第二個引數是是否支援持久化
        return new Queue(Constants.IMMEDIATE_QUEUE_XDELAY, true);
    }

    @Bean
    public CustomExchange delayExchange() {
        Map<String, Object> args = new HashMap<String, Object>();
        args.put("x-delayed-type", "direct");
        return new CustomExchange(Constants.DELAYED_EXCHANGE_XDELAY, "x-delayed-message", true, false, args);
    }

    @Bean
    public Binding bindingNotify() {
        return BindingBuilder.bind(immediateQueue()).to(delayExchange()).with(Constants.DELAY_ROUTING_KEY_XDELAY).noargs();
    }
}

XdelaySender.java

package com.example.demo.Xdelay;

import java.text.SimpleDateFormat;
import java.util.Date;

import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.example.demo.Constants.Constants;
import com.example.demo.model.Booking;

@Service
public class XdelaySender {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void send(Booking booking, int delayTime) {
        System.out.println("delayTime" + delayTime);
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        this.rabbitTemplate.convertAndSend(Constants.DELAYED_EXCHANGE_XDELAY, Constants.DELAY_ROUTING_KEY_XDELAY, booking, new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setDelay(delayTime);
                System.out.println(sdf.format(new Date()) + " Delay sent.");
                return message;
            }
        });
    }
}

XdelayReceiver.java

package com.example.demo.Xdelay;

import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;

import com.example.demo.Constants.Constants;
import com.example.demo.model.Booking;

@Component
@EnableRabbit
@Configuration
public class XdelayReceiver {

    @RabbitListener(queues = Constants.IMMEDIATE_QUEUE_XDELAY)
    public void get(Booking booking) {
        System.out.println("Receive" + booking);
    }
}

Test.java

package com.example.demo;

import java.util.Date;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import com.example.demo.Xdelay.XdelaySender;
import com.example.demo.model.Booking;

@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitMqTestApplicationTests {

	
	@Autowired
	XdelaySender xdelaySender;
	@Test
	public void test11() {
	    Booking booking = new Booking();
        booking.setBookingContent("hhaha");
        booking.setBookingName("預定房子");
        booking.setBookingTime(new Date());
        booking.setOperatorName("hellen");
        xdelaySender.send(booking, 2000);
	}
}

相關推薦

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

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

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

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

springboot-rabbitmq:實現延時隊列

中間 edate clas 設置 重新 letter direct chang stack 延時隊列應用於什麽場景 延時隊列顧名思義，即放置在該隊列裏面的消息是不需要立即消費的，而是等待一段時間之後取出消費。那麽，為什麽需要延遲消費呢？我們來看以下的場景 網上商城下訂

基於Maven的SpringBoot專案實現熱部署的兩種方式

前言 JRebel是JavaEE中比較流行的熱部署外掛，可快速實現熱部署，節省了大量重啟時間，提高了個人開發效率。看一張圖就知道了： 我這裡提到JRebel的目的主要是說明熱部署工具帶來的優勢：節省時間，提高開發效率 下面我將介紹使用maven構建的S

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

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

Springboot整合Rabbitmq實現延時消費，並實現可靠的訊息處理

一、Rabbitmq簡介1.1 rabbitmq 架構1.2 rabbitmq相關元件介紹exchange: 交換機,主要用來將生產者傳送的訊息路由給伺服器中的佇列。routing-key: 訊息路由的key，生產者在將訊息發到到exchange的時候，需要指定routing

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

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

c++中實現延時的幾種方法

方法七：對於精確度要求更高的定時操作，則應該使用QueryPerformanceFrequency()和 QueryPerformanceCounter()函式。這兩個函式是VC提供的僅供Windows 95及其後續版本使用的精確時間函式，並要求計算機從硬體上支援精確定時器。如示例工程中的Timer7、Ti

RabbitMQ 的延時佇列和映象佇列原理與實戰

摘要：在阿里雲棲開發者沙龍PHP技術專場上，掌閱資深後端工程師、掘金小測《Redis深度歷險》作者錢文品為大家介紹了RabbitM

使用aspectJ實現Spring AOP的兩種方式

classpath .org 導入 ntc www. 之前 oid 方式 public 方式一:基於aspectJ的XML配置 方式二:基於aspectJ的註解方式 基於aspectJ的XML配置 1) 引入相關jar包 2) 創建Spr

基於aspectj實現AOP操作的兩種方式——xml配置

aop ont 方式 alt 分享圖片 文件 分享 spa 技術分享 1. 要導入的 jar 包： 常用的aspectj表達式： 權限修飾符可以省略，以下表示：返回值類型為任意，com.chy.service包以及其子包下的、任意類的、參數任意的、任意方法 ex

18.2 實現線程的兩種方式

pri java虛擬機 run 構造 語法 oid except 方式 reads 一、概述 　　在Java中主要提供兩種方式實現線程，分別為繼承java.lang.Thread類與實現java.lang.Runnable接口。 二、繼承Thread類 　　a、Thread

【C語言】實現一個計算器（兩種方式）

1.使用switch…case…語句實現 #define _CRT_SECURE_NO_DEPRECATE 1 #include <stdio.h> #include <stdlib.h> int add(int x, int y) {   &nb

java:多執行緒(匿名內部類實現執行緒的兩種方式)

public class Demo4_Thread { public static void main(String[] args) { new Thread() {//1.繼承Thread類 public void run() {//2.重寫run方法 for(int i

Go基礎系列：Go實現工作池的兩種方式

worker pool簡介 worker pool其實就是執行緒池thread pool。對於go來說，直接使用的是goroutine而非執行緒，不過這裡仍然以執行緒來解釋執行緒池。 線上程池模型中，有2個佇列一個池子：任務佇列、已完成任務佇列和執行緒池。其中已完成任務佇列可能存在也可能不存在，依據實際需

vue-router 原始碼實現前端路由的兩種方式

在學習 vue-router 的程式碼之前，先來簡單瞭解一下前端路由。 前端路由主要有兩種實現方法： Hash 路由 History 路由 先來看看這兩種方法的實現原理。 接著我們將用它們來簡單實現一個自己的前端路由。 前端路由 Hash 路由 u

Springboot之接收json字串的兩種方式

第一種方式、通過關鍵欄位@RequestBody，標明這個物件接收json字串。還有第二種方式，直接通過request來獲取流。在spring中，推薦使用。 1、通過@RequestBody 接收json 直接通過@RequestBody 的方式，直接將json的資料注入到了JSONObject

java實現執行緒的兩種方式

第一種： new Thread(){}.start();這表示呼叫Thread 子類物件的 run 方法， new Thread(){}表示一個Thread 的匿名子類的例項物件，子類加上 run 方法後的程式碼如下：new Thread(){public void run(){}}.s

jquery實現省市聯動的兩種方式（xml，，，json）

省市聯動，xml傳值方式 匯入相應的jar 建立表單 <%@ page language="java" contentType="text/html; charset=UTF-8" pageEncoding="UTF-8"%> &

實現序列化的兩種方式Serializable和Parcelable

Serializable       Serializable是Java所提供的一個序列化介面,它是一個空介面,為物件提供標準的序列化和反序列化操作        libcore/ojluni/src/main/java/java/io/Serializable.java