RabbitMQ 即一個消息隊列，主要是用來實現(xiàn)應用程序的異步和解耦，同時也能起到消息緩沖，消息分發(fā)的作用。

成都創(chuàng)新互聯(lián)是一家專業(yè)提供三原企業(yè)網(wǎng)站建設,專注與成都網(wǎng)站建設、成都做網(wǎng)站、H5場景定制、小程序制作等業(yè)務。10年已為三原眾多企業(yè)、政府機構等服務。創(chuàng)新互聯(lián)專業(yè)網(wǎng)站制作公司優(yōu)惠進行中。

消息中間件在互聯(lián)網(wǎng)公司的使用中越來越多，剛才還看到新聞阿里將 RocketMQ 捐獻給了 Apache，當然了今天的主角還是講 RabbitMQ。消息中間件最主要的作用是解耦，中間件最標準的用法是生產(chǎn)者生產(chǎn)消息傳送到隊列，消費者從隊列中拿取消息并處理，生產(chǎn)者不用關心是誰來消費，消費者不用關心誰在生產(chǎn)消息，從而達到解耦的目的。在分布式的系統(tǒng)中，消息隊列也會被用在很多其它的方面，比如：分布式事務的支持，RPC 的調用等等。

以前一直使用的是 ActiveMQ，在實際的生產(chǎn)使用中也出現(xiàn)了一些小問題，在網(wǎng)絡查閱了很多的資料后，決定嘗試使用 RabbitMQ 來替換 ActiveMQ，RabbitMQ 的高可用性、高性能、靈活性等一些特點吸引了我們，查閱了一些資料整理出此文。

RabbitMQ 介紹
RabbitMQ 是實現(xiàn)AMQP（高級消息隊列協(xié)議）的消息中間件的一種，最初起源于金融系統(tǒng)，用于在分布式系統(tǒng)中存儲轉發(fā)消息，在易用性、擴展性、高可用性等方面表現(xiàn)不俗。 RabbitMQ 主要是為了實現(xiàn)系統(tǒng)之間的雙向解耦而實現(xiàn)的。當生產(chǎn)者大量產(chǎn)生數(shù)據(jù)時，消費者無法快速消費，那么需要一個中間層。保存這個數(shù)據(jù)。

AMQP，即 Advanced Message Queuing Protocol，高級消息隊列協(xié)議，是應用層協(xié)議的一個開放標準，為面向消息的中間件設計。消息中間件主要用于組件之間的解耦，消息的發(fā)送者無需知道消息使用者的存在，反之亦然。AMQP 的主要特征是面向消息、隊列、路由（包括點對點和發(fā)布/訂閱）、可靠性、安全。

RabbitMQ 是一個開源的 AMQP 實現(xiàn)，服務器端用Erlang語言編寫，支持多種客戶端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP 等，支持 AJAX。用于在分布式系統(tǒng)中存儲轉發(fā)消息，在易用性、擴展性、高可用性等方面表現(xiàn)不俗。

相關概念
通常我們談到隊列服務, 會有三個概念： 發(fā)消息者、隊列、收消息者，RabbitMQ 在這個基本概念之上, 多做了一層抽象, 在發(fā)消息者和隊列之間, 加入了交換器 (Exchange). 這樣發(fā)消息者和隊列就沒有直接聯(lián)系, 轉而變成發(fā)消息者把消息給交換器, 交換器根據(jù)調度策略再把消息再給隊列。

左側 P 代表 生產(chǎn)者，也就是往 RabbitMQ 發(fā)消息的程序。
中間即是 RabbitMQ，其中包括了交換機和隊列。
右側 C 代表 消費者，也就是往 RabbitMQ 拿消息的程序。

那么，其中比較重要的概念有 4 個，分別為：虛擬主機，交換機，隊列，和綁定。

虛擬主機：一個虛擬主機持有一組交換機、隊列和綁定。為什么需要多個虛擬主機呢？很簡單， RabbitMQ 當中，用戶只能在虛擬主機的粒度進行權限控制。 因此，如果需要禁止A組訪問B組的交換機/隊列/綁定，必須為A和B分別創(chuàng)建一個虛擬主機。每一個 RabbitMQ 服務器都有一個默認的虛擬主機“/”。

交換機：Exchange 用于轉發(fā)消息，但是它不會做存儲 ，如果沒有 Queue bind 到 Exchange 的話，它會直接丟棄掉 Producer 發(fā)送過來的消息。 這里有一個比較重要的概念：路由鍵 。消息到交換機的時候，交互機會轉發(fā)到對應的隊列中，那么究竟轉發(fā)到哪個隊列，就要根據(jù)該路由鍵。

綁定：也就是交換機需要和隊列相綁定，這其中如上圖所示，是多對多的關系。

交換機(Exchange)
交換機的功能主要是接收消息并且轉發(fā)到綁定的隊列，交換機不存儲消息，在啟用ack模式后，交換機找不到隊列會返回錯誤。交換機有四種類型：Direct, topic, Headers and Fanout

Direct：direct 類型的行為是”先匹配, 再投送”. 即在綁定時設定一個 routing_key, 消息的routing_key 匹配時, 才會被交換器投送到綁定的隊列中去.

Topic：按規(guī)則轉發(fā)消息（最靈活）

Headers：設置 header attribute 參數(shù)類型的交換機

Fanout：轉發(fā)消息到所有綁定隊列

Direct Exchange
Direct Exchange 是 RabbitMQ 默認的交換機模式，也是最簡單的模式，根據(jù)key全文匹配去尋找隊列。

第一個 X - Q1 就有一個 binding key，名字為 orange； X - Q2 就有 2 個 binding key，名字為 black 和 green。當消息中的 路由鍵 和 這個 binding key 對應上的時候，那么就知道了該消息去到哪一個隊列中。
Ps：為什么 X 到 Q2 要有 black，green，2個 binding key呢，一個不就行了嗎？ - 這個主要是因為可能又有 Q3，而Q3只接受 black 的信息，而Q2不僅接受black 的信息，還接受 green 的信息。
Topic Exchange
Topic Exchange 轉發(fā)消息主要是根據(jù)通配符。 在這種交換機下，隊列和交換機的綁定會定義一種路由模式，那么，通配符就要在這種路由模式和路由鍵之間匹配后交換機才能轉發(fā)消息。
在這種交換機模式下：

路由鍵必須是一串字符，用句號（.） 隔開，比如說 agreements.us，或者 agreements.eu.stockholm 等。
路由模式必須包含一個 星號（），主要用于匹配路由鍵指定位置的一個單詞，比如說，一個路由模式是這樣子：agreements..b.，那么就只能匹配路由鍵是這樣子的：第一個單詞是 agreements，第四個單詞是 b。 井號（#）就表示相當于一個或者多個單詞，例如一個匹配模式是 agreements.eu.berlin.#，那么，以agreements.eu.berlin 開頭的路由鍵都是可以的。

具體代碼發(fā)送的時候還是一樣，第一個參數(shù)表示交換機，第二個參數(shù)表示 routing key，第三個參數(shù)即消息。如下：

rabbitTemplate.convertAndSend("testTopicExchange","key1.a.c.key2", " this is RabbitMQ!");

topic 和 direct 類似, 只是匹配上支持了”模式”, 在”點分”的 routing_key 形式中, 可以使用兩個通配符:

*表示一個詞.

#表示零個或多個詞.

Headers Exchange
headers 也是根據(jù)規(guī)則匹配, 相較于 direct 和 topic 固定地使用 routing_key , headers 則是一個自定義匹配規(guī)則的類型. 在隊列與交換器綁定時, 會設定一組鍵值對規(guī)則, 消息中也包括一組鍵值對( headers 屬性), 當這些鍵值對有一對, 或全部匹配時, 消息被投送到對應隊列.
Fanout Exchange
Fanout Exchange 消息廣播的模式，不管路由鍵或者是路由模式，會把消息發(fā)給綁定給它的全部隊列，如果配置了 routing_key 會被忽略。
Spring Boot 集成 RabbitMQ
Spring Boot 集成 RabbitMQ 非常簡單，如果只是簡單的使用配置非常少，Spring Boot 提供了spring-boot-starter-amqp 項目對消息各種支持。
簡單使用
1、配置 Pom 包，主要是添加 spring-boot-starter-amqp 的支持
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2、配置文件

配置 RabbitMQ 的安裝地址、端口以及賬戶信息

spring.application.name=Spring-boot-rabbitmq

spring.rabbitmq.host=192.168.0.86
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=123456

3、隊列配置

@Configuration
public class RabbitConfig {

@Bean
public Queue Queue() {
    return new Queue("hello");
}

}

3、發(fā)送者

rabbitTemplate 是 Spring Boot 提供的默認實現(xiàn)

@component
public class HelloSender {

@Autowired
private AmqpTemplate rabbitTemplate;

public void send() {
    String context = "hello " + new Date();
    System.out.println("Sender : " + context);
    this.rabbitTemplate.convertAndSend("hello", context);
}

}

4、接收者

@Component
@RabbitListener(queues = "hello")
public class HelloReceiver {

@RabbitHandler
public void process(String hello) {
    System.out.println("Receiver  : " + hello);
}

}

5、測試

@RunWith(SpringRunner.class)@SpringBootTest
br/>@SpringBootTest

@Autowired
private HelloSender helloSender;

@Test
public void hello() throws Exception {
    helloSender.send();
}

}

注意，發(fā)送者和接收者的 queue name 必須一致，不然不能接收

多對多使用
一個發(fā)送者，N 個接收者或者 N 個發(fā)送者和 N 個接收者會出現(xiàn)什么情況呢？
一對多發(fā)送
對上面的代碼進行了小改造，接收端注冊了兩個 Receiver,Receiver1 和 Receiver2，發(fā)送端加入?yún)?shù)計數(shù)，接收端打印接收到的參數(shù)，下面是測試代碼，發(fā)送一百條消息，來觀察兩個接收端的執(zhí)行效果

@Test
public void oneToMany() throws Exception {
for (int i=0;i<100;i++){
neoSender.send(i);
}
}

結果如下：

Receiver 1: Spring boot neo queue**11
Receiver 2: Spring boot neo queue**12
Receiver 2: Spring boot neo queue**14
Receiver 1: Spring boot neo queue**13
Receiver 2: Spring boot neo queue**15
Receiver 1: Spring boot neo queue**16
Receiver 1: Spring boot neo queue**18
Receiver 2: Spring boot neo queue**17
Receiver 2: Spring boot neo queue**19
Receiver 1: Spring boot neo queue**20

根據(jù)返回結果得到以下結論

一個發(fā)送者，N個接受者,經(jīng)過測試會均勻的將消息發(fā)送到N個接收者中

多對多發(fā)送

復制了一份發(fā)送者，加入標記，在一百個循環(huán)中相互交替發(fā)送

@Test
public void manyToMany() throws Exception {
for (int i=0;i<100;i++){
neoSender.send(i);
neoSender2.send(i);
}
}

結果如下：

Receiver 1: Spring boot neo queue**20
Receiver 2: Spring boot neo queue**20
Receiver 1: Spring boot neo queue**21
Receiver 2: Spring boot neo queue**21
Receiver 1: Spring boot neo queue**22
Receiver 2: Spring boot neo queue**22
Receiver 1: Spring boot neo queue**23
Receiver 2: Spring boot neo queue**23
Receiver 1: Spring boot neo queue**24
Receiver 2: Spring boot neo queue**24
Receiver 1: Spring boot neo queue**25
Receiver 2: Spring boot neo queue**25

結論：和一對多一樣，接收端仍然會均勻接收到消息

高級使用

對象的支持

Spring Boot 以及完美的支持對象的發(fā)送和接收，不需要格外的配置。

//發(fā)送者
public void send(User user) {
System.out.println("Sender object: " + user.toString());
this.rabbitTemplate.convertAndSend("object", user);
}

...

//接收者@RabbitHandler
br/>@RabbitHandler
System.out.println("Receiver object : " + user);
}

結果如下：
Sender object: User{name='neo', pass='123456'}
Receiver object : User{name='neo', pass='123456'}

Topic Exchange

topic 是 RabbitMQ 中最靈活的一種方式，可以根據(jù) routing_key 自由的綁定不同的隊列

首先對 topic 規(guī)則配置，這里使用兩個隊列來測試

@Configuration
public class TopicRabbitConfig {

final static String message = "topic.message";
final static String messages = "topic.messages";

@Bean
public Queue queueMessage() {
    return new Queue(TopicRabbitConfig.message);
}

@Bean
public Queue queueMessages() {
    return new Queue(TopicRabbitConfig.messages);
}

@Bean
TopicExchange exchange() {
    return new TopicExchange("exchange");
}

@Bean
Binding bindingExchangeMessage(Queue queueMessage, TopicExchange exchange) {
    return BindingBuilder.bind(queueMessage).to(exchange).with("topic.message");
}

@Bean
Binding bindingExchangeMessages(Queue queueMessages, TopicExchange exchange) {
    return BindingBuilder.bind(queueMessages).to(exchange).with("topic.#");
}

}

使用 queueMessages 同時匹配兩個隊列，queueMessage 只匹配 “topic.message” 隊列

public void send1() {
String context = "hi, i am message 1";
System.out.println("Sender : " + context);
this.rabbitTemplate.convertAndSend("exchange", "topic.message", context);
}

public void send2() {
String context = "hi, i am messages 2";
System.out.println("Sender : " + context);
this.rabbitTemplate.convertAndSend("exchange", "topic.messages", context);
}

發(fā)送send1會匹配到topic.#和topic.message 兩個Receiver都可以收到消息，發(fā)送send2只有topic.#可以匹配所有只有Receiver2監(jiān)聽到消息
Fanout Exchange
Fanout 就是我們熟悉的廣播模式或者訂閱模式，給 Fanout 交換機發(fā)送消息，綁定了這個交換機的所有隊列都收到這個消息。
Fanout 相關配置

@Configuration
public class FanoutRabbitConfig {

@Bean
public Queue AMessage() {
    return new Queue("fanout.A");
}

@Bean
public Queue BMessage() {
    return new Queue("fanout.B");
}

@Bean
public Queue CMessage() {
    return new Queue("fanout.C");
}

@Bean
FanoutExchange fanoutExchange() {
    return new FanoutExchange("fanoutExchange");
}

@Bean
Binding bindingExchangeA(Queue AMessage,FanoutExchange fanoutExchange) {
    return BindingBuilder.bind(AMessage).to(fanoutExchange);
}

@Bean
Binding bindingExchangeB(Queue BMessage, FanoutExchange fanoutExchange) {
    return BindingBuilder.bind(BMessage).to(fanoutExchange);
}

@Bean
Binding bindingExchangeC(Queue CMessage, FanoutExchange fanoutExchange) {
    return BindingBuilder.bind(CMessage).to(fanoutExchange);
}

}

這里使用了 A、B、C 三個隊列綁定到 Fanout 交換機上面，發(fā)送端的 routing_key 寫任何字符都會被忽略：
public void send() {
String context = "hi, fanout msg ";
System.out.println("Sender : " + context);
this.rabbitTemplate.convertAndSend("fanoutExchange","", context);
}

結果如下：
Sender : hi, fanout msg
...
fanout Receiver B: hi, fanout msg
fanout Receiver A : hi, fanout msg
fanout Receiver C: hi, fanout msg

結果說明，綁定到 fanout 交換機上面的隊列都收到了消息

創(chuàng)新互聯(lián)www.cdcxhl.cn，專業(yè)提供香港、美國云服務器，動態(tài)BGP最優(yōu)骨干路由自動選擇，持續(xù)穩(wěn)定高效的網(wǎng)絡助力業(yè)務部署。公司持有工信部辦法的idc、isp許可證， 機房獨有T級流量清洗系統(tǒng)配攻擊溯源，準確進行流量調度，確保服務器高可用性。佳節(jié)活動現(xiàn)已開啟，新人活動云服務器買多久送多久。

新聞標題：SpringBoot之RabbitMQ-創(chuàng)新互聯(lián)
當前網(wǎng)址：http://jinyejixie.com/article10/ccegdo.html

成都網(wǎng)站建設公司_創(chuàng)新互聯(lián)，為您提供服務器托管、軟件開發(fā)、網(wǎng)站改版、網(wǎng)站維護、品牌網(wǎng)站設計、移動網(wǎng)站建設

廣告

聲明：本網(wǎng)站發(fā)布的內容（圖片、視頻和文字）以用戶投稿、用戶轉載內容為主，如果涉及侵權請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內容未經(jīng)允許不得轉載，或轉載時需注明來源： 創(chuàng)新互聯(lián)