初探 RabbitMQ 訊息佇列
初探 RabbitMQ 訊息佇列
rabbitmq基礎概念常見應用場景匯入依賴屬性配置具體編碼定義佇列實體類控制器訊息消費者主函式測試總結說點什麼
SpringBoot
是為了簡化Spring
應用的建立、執行、除錯、部署等一系列問題而誕生的產物,自動裝配的特性讓我們可以更好的關注業務本身而不是外部的XML配置,我們只需遵循規範,引入相關的依賴就可以輕易的搭建出一個 WEB 工程
MQ全稱(Message Queue)又名訊息佇列,是一種非同步通訊的中介軟體。可以將它理解成郵局,傳送者將訊息傳遞到郵局,然後由郵局幫我們傳送給具體的訊息接收者(消費者),具體傳送過程與時間我們無需關心,它也不會干擾我進行其它事情。常見的MQ有kafka、activemq、zeromq、rabbitmq 等等,各大MQ的對比和優劣勢可以自行Google
rabbitmq
RabbitMQ是一個遵循AMQP協議,由面向高併發的erlanng
語言開發而成,用在實時的對可靠性要求比較高的訊息傳遞上,支援多種語言客戶端。支援延遲佇列(這是一個非常有用的功能)
….
基礎概念
Broker:簡單來說就是訊息佇列伺服器實體
Exchange:訊息交換機,它指定訊息按什麼規則,路由到哪個佇列
Queue:訊息佇列載體,每個訊息都會被投入到一個或多個佇列
Binding:繫結,它的作用就是把exchange
和queue
按照路由規則繫結起來
Routing Key:路由關鍵字,exchange
根據這個關鍵字進行訊息投遞
vhost:虛擬主機,一個broker
vhost
,用作不同使用者的許可權分離producer:訊息生產者,就是投遞訊息的程式
consumer:訊息消費者,就是接受訊息的程式
channel:訊息通道,在客戶端的每個連線裡,可建立多個
channel
,每個channel
代表一個會話任務
基於Centos7.x安裝請參考: http://blog.battcn.com/2017/08/20/linux/linux-centos7-ribbitmq/
常見應用場景
-
郵箱傳送:使用者註冊後投遞訊息到
rabbitmq
中,由訊息的消費方非同步的傳送郵件,提升系統響應速度 -
流量削峰:一般在秒殺活動中應用廣泛,秒殺會因為流量過大,導致應用掛掉,為了解決這個問題,一般在應用前端加入訊息佇列。用於控制活動人數,將超過此一定閥值的訂單直接丟棄。緩解短時間的高流量壓垮應用。
-
訂單超時:利用
rabbitmq
的延遲佇列,可以很簡單的實現訂單超時的功能,比如使用者在下單後30分鐘未支付取消訂單 -
還有更多應用場景就不一一列舉了…..
匯入依賴
在 pom.xml
中新增 spring-boot-starter-amqp
的依賴
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.46</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
屬性配置
在 application.properties
檔案中配置rabbitmq
相關內容,值得注意的是這裡配置了手動ACK的開關
spring.rabbitmq.username=battcn
spring.rabbitmq.password=battcn
spring.rabbitmq.host=192.168.0.133
spring.rabbitmq.port=5672
spring.rabbitmq.virtual-host=/
# 手動ACK 不開啟自動ACK模式,目的是防止報錯後未正確處理訊息丟失 預設 為 none
spring.rabbitmq.listener.simple.acknowledge-mode=manual
具體編碼
定義佇列
如果手動建立過或者RabbitMQ
中已經存在該佇列那麼也可以省略下述程式碼…
package com.battcn.config;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* RabbitMQ配置
*
* @author Levin
* @since 2018/4/11 0011
*/
@Configuration
public class RabbitConfig {
public static final String DEFAULT_BOOK_QUEUE = "dev.book.register.default.queue";
public static final String MANUAL_BOOK_QUEUE = "dev.book.register.manual.queue";
@Bean
public Queue defaultBookQueue() {
// 第一個是 QUEUE 的名字,第二個是訊息是否需要持久化處理
return new Queue(DEFAULT_BOOK_QUEUE, true);
}
@Bean
public Queue manualBookQueue() {
// 第一個是 QUEUE 的名字,第二個是訊息是否需要持久化處理
return new Queue(MANUAL_BOOK_QUEUE, true);
}
}
實體類
建立一個Book
類
public class Book implements java.io.Serializable {
private static final long serialVersionUID = -2164058270260403154L;
private String id;
private String name;
// 省略get set ...
}
控制器
編寫一個Controller
類,用於訊息傳送工作
package com.battcn.controller;
import com.battcn.config.RabbitConfig;
import com.battcn.entity.Book;
import com.battcn.handler.BookHandler;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
/**
* @author Levin
* @since 2018/4/2 0002
*/
@RestController
@RequestMapping(value = "/books")
public class BookController {
private final RabbitTemplate rabbitTemplate;
@Autowired
public BookController(RabbitTemplate rabbitTemplate) {
this.rabbitTemplate = rabbitTemplate;
}
/**
* this.rabbitTemplate.convertAndSend(RabbitConfig.DEFAULT_BOOK_QUEUE, book); 對應 {@link BookHandler#listenerAutoAck}
* this.rabbitTemplate.convertAndSend(RabbitConfig.MANUAL_BOOK_QUEUE, book); 對應 {@link BookHandler#listenerManualAck}
*/
@GetMapping
public void defaultMessage() {
Book book = new Book();
book.setId("1");
book.setName("一起來學Spring Boot");
this.rabbitTemplate.convertAndSend(RabbitConfig.DEFAULT_BOOK_QUEUE, book);
this.rabbitTemplate.convertAndSend(RabbitConfig.MANUAL_BOOK_QUEUE, book);
}
}
訊息消費者
預設情況下 spring-boot-data-amqp
是自動ACK
機制,就意味著 MQ 會在訊息消費完畢後自動幫我們去ACK,這樣依賴就存在這樣一個問題:如果報錯了,訊息不會丟失,會無限迴圈消費,很容易就吧磁碟空間耗完,雖然可以配置消費的次數但這種做法也有失優雅。目前比較推薦的就是我們手動ACK然後將消費錯誤的訊息轉移到其它的訊息佇列中,做補償處理
package com.battcn.handler;
import com.battcn.config.RabbitConfig;
import com.battcn.entity.Book;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.io.IOException;
/**
* BOOK_QUEUE 消費者
*
* @author Levin
* @since 2018/4/11 0011
*/
@Component
public class BookHandler {
private static final Logger log = LoggerFactory.getLogger(BookHandler.class);
/**
* <p>TODO 該方案是 spring-boot-data-amqp 預設的方式,不太推薦。具體推薦使用 listenerManualAck()</p>
* 預設情況下,如果沒有配置手動ACK, 那麼Spring Data AMQP 會在訊息消費完畢後自動幫我們去ACK
* 存在問題:如果報錯了,訊息不會丟失,但是會無限迴圈消費,一直報錯,如果開啟了錯誤日誌很容易就吧磁碟空間耗完
* 解決方案:手動ACK,或者try-catch 然後在 catch 裡面講錯誤的訊息轉移到其它的系列中去
* spring.rabbitmq.listener.simple.acknowledge-mode=manual
* <p>
*
* @param book 監聽的內容
*/
@RabbitListener(queues = {RabbitConfig.DEFAULT_BOOK_QUEUE})
public void listenerAutoAck(Book book, Message message, Channel channel) {
// TODO 如果手動ACK,訊息會被監聽消費,但是訊息在佇列中依舊存在,如果 未配置 acknowledge-mode 預設是會在消費完畢後自動ACK掉
final long deliveryTag = message.getMessageProperties().getDeliveryTag();
try {
log.info("[listenerAutoAck 監聽的訊息] - [{}]", book.toString());
// TODO 通知 MQ 訊息已被成功消費,可以ACK了
channel.basicAck(deliveryTag, false);
} catch (IOException e) {
try {
// TODO 處理失敗,重新壓入MQ
channel.basicRecover();
} catch (IOException e1) {
e1.printStackTrace();
}
}
}
@RabbitListener(queues = {RabbitConfig.MANUAL_BOOK_QUEUE})
public void listenerManualAck(Book book, Message message, Channel channel) {
log.info("[listenerManualAck 監聽的訊息] - [{}]", book.toString());
try {
// TODO 通知 MQ 訊息已被成功消費,可以ACK了
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
} catch (IOException e) {
// TODO 如果報錯了,那麼我們可以進行容錯處理,比如轉移當前訊息進入其它佇列
}
}
}
主函式
package com.battcn;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
/**
* @author Levin
*/
@SpringBootApplication
public class Chapter11Application {
public static void main(String[] args) {
SpringApplication.run(Chapter11Application.class, args);
}
}
測試
完成準備事項後,啟動Chapter11Application
訪問 http://localhost:8080/books 將會看到如下內容,就代表一切正常….
2018-05-22 19:04:26.708 INFO 23752 --- [cTaskExecutor-1] com.battcn.handler.BookHandler : [listenerAutoAck 監聽的訊息] - [[email protected]]
2018-05-22 19:04:26.709 INFO 23752 --- [cTaskExecutor-1] com.battcn.handler.BookHandler : [listenerManualAck 監聽的訊息] - [[email protected]]
總結
目前很多大佬都寫過關於 SpringBoot 的教程了,如有雷同,請多多包涵,本教程基於最新的 spring-boot-starter-parent:2.0.2.RELEASE
編寫,包括新版本的特性都會一起介紹…