2. 程式人生 > >Spring Boot RabbitMQ 延遲訊息實現完整版

Spring Boot RabbitMQ 延遲訊息實現完整版

阿新 • • 發佈：2018-12-12

概述

曾經去網易面試的時候，面試官問了我一個問題，說

下完訂單後，如果使用者未支付，需要取消訂單，可以怎麼做

我當時的回答是，用定時任務掃描DB表即可。面試官不是很滿意，提出：

用定時任務無法做到準實時通知，有沒有其他辦法？

我當時的回答是：

可以用佇列，訂單下完後，傳送一個訊息到佇列裡，並指定過期時間，時間一到，執行回撥介面。

面試官聽完後，就不再問了。其實我當時的思路是對的，只不過講的不是很專業而已。專業說法是利用延遲訊息。

其實用定時任務，確實有點問題，原本業務系統希望10分鐘後，如果訂單未支付，就馬上取消訂單，並釋放商品庫存。但是一旦資料量大的話，就會加長獲取未支付訂單資料的時間，部分訂單就做不到10分鐘後取消了，可能是15分鐘，20分鐘之類的。這樣的話，庫存就無法及時得到釋放，也就會影響成單數。而利用延遲訊息，則理論上是可以做到按照設定的時間，進行訂單取消操作的。

目前網上關於使用RabbitMQ實現延遲訊息的文章，大多都是講如何利用RabbitMQ的死信佇列來實現，實現方案看起來都很繁瑣複雜,並且還是使用原始的RabbitMQ Client API來實現的，更加顯得囉嗦。

Spring Boot 已經對RabbitMQ Client API進行了包裝，使用起來簡潔很多，下面詳細介紹一下如何利用rabbitmq_delayed_message_exchange 外掛和Spring Boot來實現延遲訊息。

軟體準備

erlang

本文使用的版本是:

Erlang 20.3

RabbitMQ

本文使用的是window版本的RabbitMQ，版本號是:

3.7.4

rabbitmq_delayed_message_exchange外掛

外掛下載地址：

開啟網址後，ctrl + f，搜尋rabbitmq_delayed_message_exchange。這裡寫圖片描述

千萬記住，一定選好版本號，由於我使用的是RabbitMQ 3.7.4,因此對應的rabbitmq_delayed_message_exchange外掛也必須選擇3.7.x的。

如果沒有選對版本，在使用延遲訊息的時候，會遇到各種各樣的奇葩問題，而且網上還找不到解決方案。我因為這個問題，折騰了整整一個晚上。請牢記，要選對外掛版本。

下載完外掛後，將其放置到RabbitMQ安裝目錄下的plugins

目錄下，並使用如下命令啟動這個外掛：

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

如果啟動成功會出現如下資訊：

The following plugins have been enabled: rabbitmq_delayed_message_exchange

啟動外掛成功後，記得重啟一下RabbitMQ，讓其生效。

整合RabbitMQ

這個就非常簡單了,直接在maven工程的pom.xml檔案中加入

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
  • 1
  • 2
  • 3
  • 4

Spring Boot的版本我使用的是2.0.1.RELEASE.

接下來在application.properties檔案中加入redis配置：

spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
  • 1
  • 2
  • 3
  • 4

定義ConnectionFactory和RabbitTemplate

也很簡單，程式碼如下：

package com.mq.rabbitmq;

import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
@ConfigurationProperties(prefix = "spring.rabbitmq")
public class RabbitMqConfig {
    private String host;
    private int port;
    private String userName;
    private String password;

    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory(host,port);
        cachingConnectionFactory.setUsername(userName);
        cachingConnectionFactory.setPassword(password);
        cachingConnectionFactory.setVirtualHost("/");
        cachingConnectionFactory.setPublisherConfirms(true);
        return cachingConnectionFactory;
    }

    @Bean
    public RabbitTemplate rabbitTemplate() {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory());
        return rabbitTemplate;
    }

    public String getHost() {
        return host;
    }

    public void setHost(String host) {
        this.host = host;
    }

    public int getPort() {
        return port;
    }

    public void setPort(int port) {
        this.port = port;
    }

    public String getUserName() {
        return userName;
    }

    public void setUserName(String userName) {
        this.userName = userName;
    }

    public String getPassword() {
        return password;
    }

    public void setPassword(String password) {
        this.password = password;
    }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66

Exchange和Queue配置

package com.mq.rabbitmq;

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

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

@Configuration
public class QueueConfig {

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

    @Bean
    public Queue queue() {
        Queue queue = new Queue("test_queue_1", true);
        return queue;
    }

    @Bean
    public Binding binding() {
        return BindingBuilder.bind(queue()).to(delayExchange()).with("test_queue_1").noargs();
    }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31

這裡要特別注意的是，使用的是CustomExchange,不是DirectExchange，另外CustomExchange的型別必須是x-delayed-message。

實現訊息傳送

package com.mq.rabbitmq;

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 java.text.SimpleDateFormat;
import java.util.Date;

@Service
public class MessageServiceImpl {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendMsg(String queueName,String msg) {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        System.out.println("訊息傳送時間:"+sdf.format(new Date()));
        rabbitTemplate.convertAndSend("test_exchange", queueName, msg, new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setHeader("x-delay",3000);
                return message;
            }
        });
    }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31

注意在傳送的時候，必須加上一個header

x-delay

在這裡我設定的延遲時間是3秒。

訊息消費者

package com.mq.rabbitmq;

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

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

@Component
public class MessageReceiver {

    @RabbitListener(queues = "test_queue_1")
    public void receive(String msg) {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        System.out.println("訊息接收時間:"+sdf.format(new Date()));
        System.out.println("接收到的訊息:"+msg);
    }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19

執行Spring Boot程式和傳送訊息

直接在main方法裡執行Spring Boot程式,Spring Boot會自動解析MessageReceiver類的。

接下來只需要用Junit執行一下發送訊息的介面即可。

package com.mq.rabbitmq;

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;

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

    @Autowired
    private MessageServiceImpl messageService;

    @Test
    public void send() {
        messageService.sendMsg("test_queue_1","hello i am delay msg");
    }

}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22

執行完後，可以看到如下資訊：

訊息傳送時間:2018-05-03 12:44:53

3秒鐘後，Spring Boot控制檯會輸出:

訊息接收時間:2018-05-03 12:44:56 接收到的訊息:hello i am delay msg

--------------------- 本文來自 Sam哥哥 的CSDN 部落格 ，全文地址請點選：https://blog.csdn.net/linsongbin1/article/details/80178122?utm_source=copy

相關推薦

Spring Boot RabbitMQ 延遲訊息實現完整版

概述 曾經去網易面試的時候，面試官問了我一個問題，說 下完訂單後，如果使用者未支付，需要取消訂單，可以怎麼做 我當時的回答是，用定時任務掃描DB表即可。面試官不是很滿意，提出： 用定時任務無法做到準實時通知，有沒有其他辦法？ 我當時的回答是： 可以用

Spring Boot RabbitMQ 延遲消息實現完整版

throws 曾經 nfa header 面試 you ble 程序 記得 概述 曾經去網易面試的時候，面試官問了我一個問題，說 下完訂單後，如果用戶未支付，需要取消訂單，可以怎麽做 我當時的回答是，用定時任務掃描DB表即可。面試官不是很滿意，提出： 用定時任務無法

Spring Boot + RabbitMQ死信機制實現延遲佇列

本文轉載自部落格：http://blog.csdn.net/nexttake/article/details/78607486 ------------------------------------------------------------------------------------

輕輕鬆鬆學習SpringBoot2：第二十五篇： Spring Boot和Mongodb整合（完整版）

今天主要講的是Spring Boot和Mongodb整合我們先來回顧一下前面章節的相關內容前面我們講了SpringBoot和mysql整合，並且講了操作資料庫的幾種方式自動生成表資料庫操作操作篇回到正題，mongodb的安裝在這就不累述了，win版本的去官網下載，然後一直下一

spring boot 應用釋出到 docker (完整版)

spring boot 應用釋出到 docker 完整版 一、概述     spring boot 和 docker 本身就不多介紹了，本文主要介紹使用 docker-maven-plugin 外掛，直接將 spring boot 應用一鍵釋出到 docker 容

spring boot Rabbitmq集成,延時消息隊列實現

-c 轉發 包裝 let err 接下來 rec ger str 本篇主要記錄Spring boot 集成Rabbitmq，分為兩部分， 第一部分為創建普通消息隊列， 第二部分為延時消息隊列實現： spring boot提供對mq消息隊列支持amqp相關包，引入即可：

Spring boot:WebSocket+SockJS+Stomp實現廣播和點對點訊息傳送

筆記 廣播式 STS工具新建spring boot專案 使用Thymeleaf和Websocket依賴 pom.xml: <?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://mave

Spring Boot 整合 Elasticsearch，實現 function score query 權重分查詢

search 小寫 業務 jpg 啟動會 last cti cal agen 摘要: 原創出處 www.bysocket.com 「泥瓦匠BYSocket 」歡迎轉載，保留摘要，謝謝！ 『 預見未來最好的方式就是親手創造未來 – 《史蒂夫·喬布斯

spring boot+tkmybatis(mapper)工具實現通用數據操作接口時遇到的問題

之前 開始 ng- key lang ces 解決 是不是 就是 最近在做一個sb+tkmybatis整合的可以提供通用數據操作接口的demo，然而出現了很詭異的事情.，使用mapper提供的selectByPrimaryKey接口返回結果時報異常： java.lang.

spring boot高性能實現二維碼掃碼登錄(上)——單服務器版

cnblogs org var del mov 登錄 leaf 發現 relative 前言 　　目前網頁的主流登錄方式是通過手機掃碼二維碼登錄。我看了網上很多關於掃碼登錄博客後，發現基本思路大致是：打開網頁，生成uuid，然後長連接請求後端並等待登錄認證相應結

Spring Boot集成ElasticSearch實現簡單的增刪查改接口

java.net tin ID request exce 有一個 port client water SpringBoot集成ElasticSearch pom.xml文件中，依賴的各jar包版本如下： <parent> <groupId>o

spring boot整合reids 然後實現緩存分頁(方法之一) 以及RedisTemplate存到reids 裏面get 就消失的坑

temp ima 首頁 page ng- ext 使用 after opera 業務需求 首頁 實現緩存分頁 spring boot 整合redis （我的是2.0.3版本的） 在pom 文件寫上依賴包即可 <dependency><!--依賴包

Spring Boot集成MyBatis實現通用Mapper

TBase cti 自動生成代碼 port sig max 添加 記錄 connect 前言 MyBatis關於MyBatis，大部分人都很熟悉。MyBatis 是一款優秀的持久層框架，它支持定制化 SQL、存儲過程以及高級映射。MyBatis 避免了幾乎所有的 JDBC

在微服務領域Spring Boot自動伸縮如何實現

shutdown app 有一個 line down 滿足 51cto 完成 pivotal 自動伸縮是每個人都想要的，尤其是在微服務領域。讓我們看看如何在基於Spring Boot的應用程序中實現。 我們決定使用 Kubernetes 、 Pivotal Cloud F

Spring Boot如何配置SSL實現https協議

參考:https://jingyan.baidu.com/article/5225f26bb16d09e6fa0908e4.html 一、生成SSL證書   1 本文的SSL證書為JDK生成的自簽名證

RabbitMQ 延遲佇列實現訂單支付結果非同步階梯性通知

在第三方支付中，例如支付寶、或者微信，對於訂單請求，第三方支付系統採用的是訊息同步返回、非同步通知+主動補償查詢的補償機制。 由於網際網路通訊的不可靠性，例如雙方網路、伺服器、應用等因素的影響，不管是同步返回、非同步通知、主動查詢報文都可能出現超時無響應、報文丟失等情況，所以像

[轉]spring boot下WebSocket訊息推送

原文連結：spring boot下WebSocket訊息推送， 修正部分錯別字 WebSocket協議 WebSocket是一種在單個TCP連線上進行全雙工通訊的協議。WebSocket通訊協議於2011年被IETF定為標準RFC 6455，並由RFC7936補充規範。WebSocket A

Spring Boot定時任務註解實現

    在部落格專案開發過程中，需要一個任務每間隔一定的時間執行一次。譬如推送訊息，檢查文章訪問情況，清除一定時間前訪問者的id（用於控制文章閱讀數量隨重新整理而一直增長）。用Spring Boot定時執行任務。     首先在啟動類上加入&n

Spring Boot -- RabbitMQ

Spring Boot -- RabbitMQ 1. pom.xml 2. application.properties 3. config 4. 生產者 5. 消費者 6. 測試 發現物件訊息 一個生產者多個消費者 多個

Spring Boot集成Hazelcast實現集群與分布式內存緩存

exe 端口 ava 客戶端訪問 oot cacheable nts app 區域 Hazelcast是Hazelcast公司開源的一款分布式內存數據庫產品，提供彈性可擴展、高性能的分布式內存計算。並通過提供諸如Map，Queue，ExecutorService，Lock和