RabbitMQ 即一個(gè)消息隊(duì)列亚再，主要是用來實(shí)現(xiàn)應(yīng)用程序的異步和解耦卵蛉，同時(shí)也能起到消息緩沖回铛，消息分發(fā)的作用。

消息中間件在互聯(lián)網(wǎng)公司的使用中越來越多跋炕，剛才還看到新聞阿里將RocketMQ捐獻(xiàn)給了apache赖晶，當(dāng)然了今天的主角還是講RabbitMQ。消息中間件最主要的作用是解耦辐烂，中間件最標(biāo)準(zhǔn)的用法是生產(chǎn)者生產(chǎn)消息傳送到隊(duì)列遏插，消費(fèi)者從隊(duì)列中拿取消息并處理，生產(chǎn)者不用關(guān)心是誰來消費(fèi)纠修，消費(fèi)者不用關(guān)心誰在生產(chǎn)消息胳嘲，從而達(dá)到解耦的目的。在分布式的系統(tǒng)中扣草，消息隊(duì)列也會(huì)被用在很多其它的方面了牛，比如：分布式事務(wù)的支持，RPC的調(diào)用等等辰妙。

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

RabbitMQ介紹

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

AMQP韵吨，即Advanced Message Queuing Protocol，高級消息隊(duì)列協(xié)議移宅，是應(yīng)用層協(xié)議的一個(gè)開放標(biāo)準(zhǔn)归粉，為面向消息的中間件設(shè)計(jì)。消息中間件主要用于組件之間的解耦漏峰，消息的發(fā)送者無需知道消息使用者的存在糠悼，反之亦然。AMQP的主要特征是面向消息浅乔、隊(duì)列倔喂、路由（包括點(diǎn)對點(diǎn)和發(fā)布/訂閱）、可靠性靖苇、安全席噩。

RabbitMQ是一個(gè)開源的AMQP實(shí)現(xiàn)，服務(wù)器端用Erlang語言編寫贤壁，支持多種客戶端悼枢，如：Python、Ruby芯砸、.NET萧芙、Java给梅、JMS假丧、C、PHP动羽、ActionScript包帚、XMPP、STOMP等运吓，支持AJAX渴邦。用于在分布式系統(tǒng)中存儲(chǔ)轉(zhuǎn)發(fā)消息疯趟，在易用性、擴(kuò)展性谋梭、高可用性等方面表現(xiàn)不俗信峻。

相關(guān)概念

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

• 左側(cè) P 代表 生產(chǎn)者隘庄，也就是往 RabbitMQ 發(fā)消息的程序踢步。
• 中間即是 RabbitMQ，其中包括了 交換機(jī) 和 隊(duì)列丑掺。
• 右側(cè) C 代表 消費(fèi)者获印，也就是往 RabbitMQ 拿消息的程序。

那么街州，其中比較重要的概念有 4 個(gè)兼丰，分別為：虛擬主機(jī)，交換機(jī)唆缴，隊(duì)列地粪，和綁定。

• 虛擬主機(jī)：一個(gè)虛擬主機(jī)持有一組交換機(jī)琐谤、隊(duì)列和綁定蟆技。為什么需要多個(gè)虛擬主機(jī)呢？很簡單斗忌，RabbitMQ當(dāng)中质礼，用戶只能在虛擬主機(jī)的粒度進(jìn)行權(quán)限控制。 因此织阳，如果需要禁止A組訪問B組的交換機(jī)/隊(duì)列/綁定眶蕉，必須為A和B分別創(chuàng)建一個(gè)虛擬主機(jī)。每一個(gè)RabbitMQ服務(wù)器都有一個(gè)默認(rèn)的虛擬主機(jī)“/”唧躲。
• 交換機(jī)：Exchange 用于轉(zhuǎn)發(fā)消息造挽，但是它不會(huì)做存儲(chǔ) ，如果沒有 Queue bind 到 Exchange 的話弄痹，它會(huì)直接丟棄掉 Producer 發(fā)送過來的消息饭入。 這里有一個(gè)比較重要的概念：路由鍵 。消息到交換機(jī)的時(shí)候肛真，交互機(jī)會(huì)轉(zhuǎn)發(fā)到對應(yīng)的隊(duì)列中谐丢，那么究竟轉(zhuǎn)發(fā)到哪個(gè)隊(duì)列，就要根據(jù)該路由鍵。
• 綁定：也就是交換機(jī)需要和隊(duì)列相綁定乾忱，這其中如上圖所示讥珍，是多對多的關(guān)系。

交換機(jī)(Exchange)

交換機(jī)的功能主要是接收消息并且轉(zhuǎn)發(fā)到綁定的隊(duì)列窄瘟，交換機(jī)不存儲(chǔ)消息衷佃，在啟用ack模式后，交換機(jī)找不到隊(duì)列會(huì)返回錯(cuò)誤蹄葱。交換機(jī)有四種類型：Direct, topic, Headers and Fanout

• Direct：direct 類型的行為是”先匹配, 再投送”. 即在綁定時(shí)設(shè)定一個(gè)routing_key, 消息的routing_key 匹配時(shí), 才會(huì)被交換器投送到綁定的隊(duì)列中去.
• Topic：按規(guī)則轉(zhuǎn)發(fā)消息（最靈活）
• Headers：設(shè)置header attribute參數(shù)類型的交換機(jī)
• Fanout：轉(zhuǎn)發(fā)消息到所有綁定隊(duì)列

Direct Exchange

Direct Exchange是RabbitMQ默認(rèn)的交換機(jī)模式纲酗，也是最簡單的模式，根據(jù)key全文匹配去尋找隊(duì)列新蟆。

第一個(gè) X - Q1 就有一個(gè) binding key觅赊，名字為 orange； X - Q2 就有 2 個(gè) binding key琼稻，名字為 black 和 green吮螺。當(dāng)消息中的 路由鍵 和 這個(gè) binding key 對應(yīng)上的時(shí)候，那么就知道了該消息去到哪一個(gè)隊(duì)列中帕翻。

Ps：為什么 X 到 Q2 要有 black鸠补，green，2個(gè) binding key呢嘀掸，一個(gè)不就行了嗎紫岩？ - 這個(gè)主要是因?yàn)榭赡苡钟?Q3，而Q3只接受 black 的信息睬塌，而Q2不僅接受black 的信息泉蝌，還接受 green 的信息。

Topic Exchange

Topic Exchange 轉(zhuǎn)發(fā)消息主要是根據(jù)通配符揩晴。 在這種交換機(jī)下勋陪，隊(duì)列和交換機(jī)的綁定會(huì)定義一種路由模式，那么硫兰，通配符就要在這種路由模式和路由鍵之間匹配后交換機(jī)才能轉(zhuǎn)發(fā)消息诅愚。

在這種交換機(jī)模式下：

• 路由鍵必須是一串字符，用句號（.） 隔開劫映，比如說 agreements.us违孝，或者 agreements.eu.stockholm 等。
• 路由模式必須包含一個(gè) 星號（）泳赋，主要用于匹配路由鍵指定位置的一個(gè)單詞雌桑，比如說，一個(gè)路由模式是這樣子：agreements..b.摹蘑，那么就只能匹配路由鍵是這樣子的：第一個(gè)單詞是 agreements筹燕，第四個(gè)單詞是 b。 井號（#）就表示相當(dāng)于一個(gè)或者多個(gè)單詞衅鹿，例如一個(gè)匹配模式是agreements.eu.berlin.#撒踪，那么，以agreements.eu.berlin開頭的路由鍵都是可以的大渤。

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

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

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

• *表示一個(gè)詞.
• #表示零個(gè)或多個(gè)詞.

Headers Exchange

headers 也是根據(jù)規(guī)則匹配, 相較于 direct 和 topic 固定地使用 routing_key , headers 則是一個(gè)自定義匹配規(guī)則的類型. 在隊(duì)列與交換器綁定時(shí), 會(huì)設(shè)定一組鍵值對規(guī)則, 消息中也包括一組鍵值對( headers 屬性), 當(dāng)這些鍵值對有一對, 或全部匹配時(shí), 消息被投送到對應(yīng)隊(duì)列.

Fanout Exchange

Fanout Exchange 消息廣播的模式，不管路由鍵或者是路由模式烫幕，會(huì)把消息發(fā)給綁定給它的全部隊(duì)列俺抽，如果配置了routing_key會(huì)被忽略。

springboot集成RabbitMQ

springboot集成RabbitMQ非常簡單较曼，如果只是簡單的使用配置非常少磷斧，springboot提供了spring-boot-starter-amqp項(xiàng)目對消息各種支持。

簡單使用

1捷犹、配置pom包弛饭，主要是添加spring-boot-starter-amqp的支持

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

2、配置文件

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

spring.application.name=spirng-boot-rabbitmq

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

3侣颂、隊(duì)列配置

@Configuration
public class RabbitConfig {

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

}

4、發(fā)送者

rabbitTemplate是springboot 提供的默認(rèn)實(shí)現(xiàn)

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);
    }

}

5枪孩、接收者

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

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

}

6憔晒、測試

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

    @Autowired
    private HelloSender helloSender;

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

}

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

多對多使用

一個(gè)發(fā)送者蔑舞，N個(gè)接收者或者N個(gè)發(fā)送者和N個(gè)接收者會(huì)出現(xiàn)什么情況呢丛晌？

一對多發(fā)送

對上面的代碼進(jìn)行了小改造,接收端注冊了兩個(gè)Receiver,Receiver1和Receiver2，發(fā)送端加入?yún)?shù)計(jì)數(shù)斗幼，接收端打印接收到的參數(shù)澎蛛，下面是測試代碼，發(fā)送一百條消息蜕窿，來觀察兩個(gè)接收端的執(zhí)行效果

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

結(jié)果如下：

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

根據(jù)返回結(jié)果得到以下結(jié)論 > 一個(gè)發(fā)送者谋逻，N個(gè)接受者,經(jīng)過測試會(huì)均勻的將消息發(fā)送到N個(gè)接收者中

多對多發(fā)送

復(fù)制了一份發(fā)送者终惑，加入標(biāo)記决乎，在一百個(gè)循環(huán)中相互交替發(fā)送

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

結(jié)果如下：

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

結(jié)論：和一對多一樣，接收端仍然會(huì)均勻接收到消息

高級使用

對象的支持

springboot以及完美的支持對象的發(fā)送和接收茶行，不需要格外的配置阴挣。

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

...

//接受者
@RabbitHandler
public void process(User user) {
    System.out.println("Receiver object : " + user);
}

結(jié)果如下：

Sender object: User{name='neo', pass='123456'}
Receiver object : User{name='neo', pass='123456'}

Topic Exchange

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

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

@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同時(shí)匹配兩個(gè)隊(duì)列，queueMessage只匹配”topic.message”隊(duì)列

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會(huì)匹配到topic.#和topic.message 兩個(gè)Receiver都可以收到消息茎芭，發(fā)送send2只有topic.#可以匹配所有只有Receiver2監(jiān)聽到消息

Fanout Exchange

Fanout 就是我們熟悉的廣播模式或者訂閱模式揖膜，給Fanout交換機(jī)發(fā)送消息，綁定了這個(gè)交換機(jī)的所有隊(duì)列都收到這個(gè)消息梅桩。

Fanout 相關(guān)配置

@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三個(gè)隊(duì)列綁定到Fanout交換機(jī)上面宿百，發(fā)送端的routing_key寫任何字符都會(huì)被忽略：

public void send() {
        String context = "hi, fanout msg ";
        System.out.println("Sender : " + context);
        this.rabbitTemplate.convertAndSend("fanoutExchange","", context);
}

結(jié)果如下：

Sender : hi, fanout msg 
...
fanout Receiver B: hi, fanout msg 
fanout Receiver A  : hi, fanout msg 
fanout Receiver C: hi, fanout msg 

結(jié)果說明趁仙，綁定到fanout交換機(jī)上面的隊(duì)列都收到了消息

上面所有的例子，代碼在這里

參考

• RabbitMQ 使用參考
• RabbitMQ：Spring 集成 RabbitMQ 與其概念垦页，消息持久化雀费，ACK機(jī)制

往期回顧

• Spring Boot系列(七)：springboot+mybatis多數(shù)據(jù)源最簡解決方案
• Spring Boot系列(六)：如何優(yōu)雅的使用mybatis
• Spring Boot系列(五)：spring data jpa的使用
• Spring Boot系列(四)：thymeleaf使用詳解
• Spring Boot系列(三)：Spring Boot中Redis的使用
• Spring Boot系列(二)：web綜合開發(fā)
• Spring Boot系列(一)：Spring Boot 入門篇

作者：純潔的微笑
出處：www.ityouknow.com
版權(quán)所有，歡迎保留原文鏈接進(jìn)行轉(zhuǎn)載：)

更多干貨內(nèi)容請關(guān)注：極樂科技痊焊。
在學(xué)習(xí)過程如果有任何疑問盏袄，請來極樂網(wǎng)提問，或者掃描下方二維碼宋光，關(guān)注極樂官方微信貌矿，在平臺下方留言。