RabbitMQ 詳解

1、RabbitMQ簡介

RabbitMQ即一個消息隊列，主要是用來實現應用程序的異步和解耦，同時也能起到消息緩衝，消息分發的做用。

消息中間件在互聯網公司的使用中愈來愈多，消息中間件最主要的做用是解耦，中間件最標準的用法是生產者生產消息傳送到隊列，消費者從隊列中拿取消息並處理，生產者不用關心是誰來消費，消費者不用關心誰在生產消息，從而達到解耦的目的。在分佈式的系統中，消息隊列也會被用在不少其它的方面，好比：分佈式事務的支持，RPC的調用等等。

RabbitMQ是實現AMQP（高級消息隊列）的消息中間件的一種，最初起源於金融系統，用於在分佈式系統中存儲轉發消息，在易用性、擴展性、高可用性等方面表現不俗。RabbitMQ主要是爲了實現系統之間的雙向解耦而實現的。當生產者大量產生數據時，消費者沒法快速消費，那麼須要一箇中間層保存這個數據。

AMQP，即advanced  message queuing protocol，高級消息隊列協議，是應用層協議的一個開放標準，爲面向消息的中間件設計。消息中間件主要用於組件之間的解耦，消息的發送者無需知道消息使用者的存在，反之亦然。AMQP的主要特徵是面向消息、隊列、路由（包括點對點和發佈/訂閱）、可靠安全。

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

2、相關概念

一般咱們談到隊列服務，會有三個概念：發消息者、隊列、收消息者，RabbitMQ在這個基本概念之上，多作了一層抽象，在發消息者和隊列之間加入了交換器exchange，這樣發消息者和隊列就沒有直接聯繫，轉而編程發消息者把消息給交換器，交換器根據調度策略再把消息給隊列。

• 左側P表明生產者，也就是RabbitMQ 發消息的程序。
• 中間即時RabbitMQ，其中包括了交換機和隊列。
• 右側C表明消費者，也就是從RabbitMQ中拿消息的程序。

那麼，其中比較重要的概念就有四個，分別是：虛擬主機、交換機、隊列和綁定。

• 虛擬主機：一個虛擬主機持有一組交換機、隊列和綁定。爲何須要多個虛擬主機呢？很簡單， RabbitMQ 當中，用戶只能在虛擬主機的粒度進行權限控制。所以，若是須要禁止A組訪問B組的交換機、隊列、綁定，必須爲A和B分別建立一個虛擬主機、每個 RabbitMQ 服務器都有一個默認的虛擬主機「/」。
• 交換機：Exchange用於轉發消息，可是它不會作存儲，若是沒有queue bind到Exchange的話，它會直接丟掉Producer發送過來的消息。這裏有一個比較重要的概念：路由鍵。消息到交換機的時候，交換機轉發到對應的隊列中，那麼究竟轉發到哪一個隊列，就要根據該路由鍵。
• 綁定：也就是交換機須要和隊列相綁定，這其中如上圖所示，是多對多的關係。

3、交換機（Exchange）

交換機的功能主要是接收消息並轉發到綁定的隊列，交換機不存儲消息，在啓用ack模式後，交換機找不到隊列會返回錯誤。交換機有四種類型：Direct 、Topic、Headers、Fanout

• Direct ：direct類型的行爲是「先匹配再投送」，即在綁定時設定一個routing_key，消息的routing_key匹配時，纔會被交換機投送到綁定的隊列中去。
• Topic：按規則轉發消息（最靈活）
• Headers：設置header attribute參數類型的交換機（基本不用）
• Fanout：轉發消息到全部綁定隊列

一、Direct Exchange是RabbitMQ默認的交換機模型，也是最簡單的模型，根據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轉發消息主要是根據通配符。在這種交換機下，隊列和交換機的綁定會定義一種路由模式，那麼，通配符就要在這種路由模式和路由鍵之間匹配後交換機才能轉發消息。

在這種交換機模式下：

（1）路由鍵必須是一串字符，用句號隔開，好比說agreements.us，或者agreements.eu.stockholm等。

（2）路由模式必須包含一個星號（*），主要用於匹配路由鍵指定位置的一個單詞，好比說，一個路由模式是這樣的： agreements..b.*， 那麼就只能匹配路由鍵是這樣子的：第一個單詞是agreements，第四個單詞是b。井號就表示至關於一個或多個單詞，例如一個匹配模式是 agreements.eu.berlin.#，那麼，以agreements.eu.berlin 開頭的路由鍵都是能夠的。

具體代碼發送的時候仍是同樣，第一個參數表示交換機，第二個參數表示routing key，第三個參數即消息。以下：

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

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

*表示一個詞
#表示零個或多個詞

三、Header Exchange 

headers 也是根據規則匹配, 相較於 direct 和 topic 固定地使用 routing_key , headers 則是一個自定義匹配規則的類型. 在隊列與交換器綁定時, 會設定一組鍵值對規則, 消息中也包括一組鍵值對( headers 屬性), 當這些鍵值對有一對, 或所有匹配時, 消息被投送到對應隊列。

四、 Fanout Exchange 

Fanout Exchange  消息廣播的模式，無論路由鍵或者是路由模式，會把消息發給綁定給它的所有隊列，若是配置了routing_key會被忽略。

4、springboot集成RabbitMQ

Spring Boot 集成 RabbitMQ 很是簡單，若是隻是簡單的使用配置很是少，Spring Boot 提供了spring-boot-starter-amqp項目對消息各類支持。 

一、pom.xml

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

二、配置 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

三、隊列配置

@Configuration
public class RabbitConfig {

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

四、發送者

rabbitTemplate 是 Spring Boot 提供的默認實現

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

五、接收者

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

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

六、測試

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

	@Autowired
	private HelloSender helloSender;

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

發送者和接收者的queue name必須一致，不然不能接收

5、多對多使用

一個發送者，N個接收者或者N個發送者和N個接收者會出現什麼狀況呢？

一、一對多發送

對上面的代碼進行了小改造，接收端註冊了兩個 Receiver,Receiver1 和 Receiver2，發送端加入參數計數，接收端打印接收到的參數，下面是測試代碼，發送一百條消息，來觀察兩個接收端的執行效果

@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

根據返回結果獲得如下結論

一個發送者，N個接收者，通過測試會均勻的將消息發送到N個接收者中

二、多對多發送

複製了一份發送者，加入標記，在一百個循環中相互交替發送

@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

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

6、高級使用

一、對象的支持

springboot完美的支持對象的發送和接收，不須要額外的配置。

//發送者
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);
}

結果以下：

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

二、Topic  Exchange

topic 是 RabbitMQ 中最靈活的一種方式，能夠根據 routing_key 自由的綁定不一樣的隊列

首先對 topic 規則配置，這裏使用兩個隊列來測試

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

發送send1會匹配到topic.#和topic.message 兩個Receiver均可以收到消息，發送send2只有topic.#能夠匹配全部只有Receiver2監聽到消

三、 Fanout  Exchange

Fanout 就是咱們熟悉的廣播模式或者訂閱模式，給 Fanout 交換機發送消息，綁定了這個交換機的全部隊列都收到這個消息。

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 交換機上面，發送端的 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 交換機上面的隊列都收到了消息

 

一杯咖啡，一行代碼，是一種境界，也是一種追求！