springboot中使用RabbitMq

 轉載：http://www.ityouknow.com/springboot/2016/11/30/spring-boot-rabbitMQ.html

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

消息中間件最主要的作用是解耦，中間件最標准的用法是生產者生產消息傳送到隊列，消費者從隊列中拿取消息並處理，生產者不用關心是誰來消費，消費者不用關心誰在生產消息，從而達到解耦的目的。在分布式的系統中，消息隊列也會被用在很多其它的方面，比如：分布式事務的支持，RPC 的調用等等。

RabbitMQ 介紹

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

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

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

相關概念

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

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

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

• 虛擬主機：一個虛擬主機持有一組交換機、隊列和綁定。為什么需要多個虛擬主機呢？很簡單， RabbitMQ 當中，用戶只能在虛擬主機的粒度進行權限控制。 因此，如果需要禁止A組訪問B組的交換機/隊列/綁定，必須為A和B分別創建一個虛擬主機。每一個 RabbitMQ 服務器都有一個默認的虛擬主機“/”。
• 交換機：Exchange 用於轉發消息，但是它不會做存儲 ，如果沒有 Queue bind 到 Exchange 的話，它會直接丟棄掉 Producer 發送過來的消息。 這里有一個比較重要的概念：路由鍵 。消息到交換機的時候，交互機會轉發到對應的隊列中，那么究竟轉發到哪個隊列，就要根據該路由鍵。
• 綁定：也就是交換機需要和隊列相綁定，這其中如上圖所示，是多對多的關系。

交換機(Exchange)

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

• Direct：direct 類型的行為是”先匹配, 再投送”. 即在綁定時設定一個 routing_key, 消息的routing_key 匹配時, 才會被交換器投送到綁定的隊列中去.
• Topic：按規則轉發消息（最靈活）
• Headers：設置 header attribute 參數類型的交換機
• Fanout：轉發消息到所有綁定隊列

Direct Exchange

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 轉發消息主要是根據通配符。 在這種交換機下，隊列和交換機的綁定會定義一種路由模式，那么，通配符就要在這種路由模式和路由鍵之間匹配后交換機才能轉發消息。

在這種交換機模式下：

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

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

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

Headers Exchange

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

Fanout Exchange

Fanout Exchange 消息廣播的模式，不管路由鍵或者是路由模式，會把消息發給綁定給它的全部隊列，如果配置了 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.rabbitmq.host=127.0.0.1 spring.rabbitmq.port=5672 spring.rabbitmq.username=admin spring.rabbitmq.password=admin spring.rabbitmq.virtual-host=testhost

#消費者數量
spring.rabbitmq.listener.simple.concurrency= 10
spring.rabbitmq.listener.simple.max-concurrency= 10
#消費者每次從隊列獲取的消息數量
spring.rabbitmq.listener.simple.prefetch= 1
#消費者自動啟動
spring.rabbitmq.listener.simple.auto-startup=true
#消費失敗，自動重新入隊
spring.rabbitmq.listener.simple.default-requeue-rejected= true
#啟用發送重試
spring.rabbitmq.template.retry.enabled=true
spring.rabbitmq.template.retry.initial-interval=1000
spring.rabbitmq.template.retry.max-attempts=3
spring.rabbitmq.template.retry.max-interval=10000
spring.rabbitmq.template.retry.multiplier=1.0

3、隊列配置

@Configuration public class RabbitConfig { @Bean public Queue queue(){ return new Queue("hello"); } }

3、發送者

 @Test public void testRabbitmq(){ amqpTemplate.convertAndSend("hello","fdsa"); }

4、接收者

@Component @RabbitListener(queues = "hello") public class HelloReceiver { @RabbitHandler public void process(String message) { System.out.println("Receiver  : " + message); } }

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

多對多使用

一對多發送

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

@Test public void testRabbitmq(){ for (int i = 1; i <= 10; i++) { amqpTemplate.convertAndSend("hello","這是第"+i+"條信息"); } }

結果:

Receiver2接收到了：這是第2條信息
Receiver2接收到了：這是第4條信息
Receiver2接收到了：這是第6條信息
Receiver2接收到了：這是第8條信息
Receiver2接收到了：這是第10條信息
Receiver1接收到了：這是第1條信息
Receiver1接收到了：這是第3條信息
Receiver1接收到了：這是第5條信息
Receiver1接收到了：這是第7條信息
Receiver1接收到了：這是第9條信息

根據返回結果得到以下結論：一個發送者，N個接受者,經過測試會均勻的將消息發送到N個接收者中。

多對多發送

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

@Test public void testRabbitmq(){ for (int i = 1; i <= 10; i++) { amqpTemplate.convertAndSend("hello","這是send1的第"+i+"條信息"); amqpTemplate.convertAndSend("hello","這是send2的第"+i+"條信息"); } }

結果：

Receiver2接收到了：這是send2的第1條信息
Receiver1接收到了：這是send1的第1條信息
Receiver2接收到了：這是send2的第2條信息
Receiver2接收到了：這是send2的第3條信息
Receiver2接收到了：這是send2的第4條信息
Receiver2接收到了：這是send2的第5條信息
Receiver2接收到了：這是send2的第6條信息
Receiver2接收到了：這是send2的第7條信息
Receiver2接收到了：這是send2的第8條信息
Receiver2接收到了：這是send2的第9條信息
Receiver1接收到了：這是send1的第2條信息
Receiver2接收到了：這是send2的第10條信息
Receiver1接收到了：這是send1的第3條信息
Receiver1接收到了：這是send1的第4條信息
Receiver1接收到了：這是send1的第5條信息
Receiver1接收到了：這是send1的第6條信息
Receiver1接收到了：這是send1的第7條信息
Receiver1接收到了：這是send1的第8條信息
Receiver1接收到了：這是send1的第9條信息
Receiver1接收到了：這是send1的第10條信息

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

高級使用

對象的支持

Spring Boot 已經完美的支持對象的發送和接收，不需要格外的配置。

@Test public void testRabbitmq(){ Map<String,Object>map = new HashMap<>(1); map.put("k1","v1"); amqpTemplate.convertAndSend("hello",map); }

消費：

@Component @RabbitListener(queues="hello") public class Receiver1 { @RabbitHandler public void receiver(Map<String,Object> map){ System.err.println("Receiver1接收到了："+map); } }

結果:

Receiver1接收到了：{k1=v1}

Topic Exchange

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

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

隊列queueMessage只使用路由topic.message進行綁定關聯，隊列queueMessages使用路由topic.#鏡像綁定，也就是路由鍵topic.開頭的所有路由都可以將數據推送到該隊列。

@Test public void testRabbitmq(){ Map<String,Object>map = new HashMap<>(1); map.put("k1","v1"); amqpTemplate.convertAndSend("exchange","topic.me",map); map.put("k2","v2"); amqpTemplate.convertAndSend("exchange","topic.message",map); }

@Component @RabbitListener(queues="topic.message") public class Receiver1 { @RabbitHandler public void receiver(Map<String,Object> map){ System.err.println("Receiver1接收到了："+map); } }

@Component @RabbitListener(queues="topic.messges") public class Receiver2 { @RabbitHandler public void receiver(Map<String,Object> message){ System.err.println("Receiver2接收到了："+message); } }

結果：

Receiver2接收到了：{k1=v1} Receiver1接收到了：{k1=v1, k2=v2} Receiver2接收到了：{k1=v1, k2=v2}

第一個發送只有Receiver2能接收到，第二個發送Receiver1、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("fanout"); } @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 寫任何字符都會被忽略：

@Test public void testRabbitmq(){ Map<String,Object>map = new HashMap<>(1); map.put("k1","v1"); amqpTemplate.convertAndSend("fanout","",map); }

結果：

Receiver2接收到了：{k1=v1} Receiver1接收到了：{k1=v1}

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