目錄

• RabbitMQ 概念
• exchange交換機機制
  • 什么是交換機
  • binding？
  • Direct Exchange交換機
  • Topic Exchange交換機
  • Fanout Exchange交換機
  • Header Exchange交換機
• RabbitMQ 的 Hello - Demo（springboot實現）
• RabbitMQ 的 Hello Demo（spring xml實現）
• RabbitMQ 在生產環境下運用和出現的問題
  • Spring RabbitMQ 注解
  • 消息的 JSON 傳輸
  • 消息持久化，斷線重連，ACK。

RabbitMQ 概念

RabbitMQ 即一個消息隊列，主要是用來實現應用程序的異步和解耦，同時也能起到消息緩沖，消息分發的作用。RabbitMQ使用的是AMQP協議，它是一種二進制協議。默認啟動端口 5672。

在 RabbitMQ 中，如下圖結構：

rabbitmq

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

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

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

exchange交換機機制

什么是交換機

rabbitmq的message model實際上消息不直接發送到queue中，中間有一個exchange是做消息分發，producer甚至不知道消息發送到那個隊列中去。因此，當exchange收到message時，必須準確知道該如何分發。是append到一定規則的queue，還是append到多個queue中，還是被丟棄？這些規則都是通過exchagne的4種type去定義的。

The core idea in the messaging model in RabbitMQ is that the producer never sends any messages directly to a queue. Actually, quite often the producer doesn't even know if a message will be delivered to any queue at all.

Instead, the producer can only send messages to an exchange. An exchange is a very simple thing. On one side it receives messages from producers and the other side it pushes them to queues. The exchange must know exactly what to do with a message it receives. Should it be appended to a particular queue? Should it be appended to many queues? Or should it get discarded. The rules for that are defined by the exchange type.

exchange是一個消息的agent，每一個虛擬的host中都有定義。它的職責是把message路由到不同的queue中。

binding？

exchange和queue通過routing-key關聯，這兩者之間的關系是就是binding。如下圖所示,X表示交換機，紅色表示隊列，交換機通過一個routing-key去binding一個queue，routing-key有什么作用呢？看Direct exchange類型交換機。

Directed Exchange

路由鍵exchange，該交換機收到消息后會把消息發送到指定routing-key的queue中。那消息交換機是怎么知道的呢？其實，producer deliver消息的時候會把routing-key add到 message header中。routing-key只是一個messgae的attribute。

A direct exchange delivers messages to queues based on a message routing key. The routing key is a message attribute added into the message header by the producer. The routing key can be seen as an "address" that the exchange use to decide how to route the message. A message goes to the queue(s) whose binding key exactly matches the routing key of the message.

Default Exchange
這種是特殊的Direct Exchange，是rabbitmq內部默認的一個交換機。該交換機的name是空字符串，所有queue都默認binding 到該交換機上。所有binding到該交換機上的queue，routing-key都和queue的name一樣。

Topic Exchange

通配符交換機，exchange會把消息發送到一個或者多個滿足通配符規則的routing-key的queue。其中*表號匹配一個word，#匹配多個word和路徑，路徑之間通過.隔開。如滿足a.*.c的routing-key有a.hello.c；滿足#.hello的routing-key有a.b.c.helo。

Fanout Exchange

扇形交換機，該交換機會把消息發送到所有binding到該交換機上的queue。這種是publisher/subcribe模式。用來做廣播最好。
所有該exchagne上指定的routing-key都會被ignore掉。

The fanout copies and routes a received message to all queues that are bound to it regardless of routing keys or pattern matching as with direct and topic exchanges. Keys provided will simply be ignored.

Header Exchange

設置header attribute參數類型的交換機。

RabbitMQ 的 Hello Demo

安裝就不說了，建議按照官方文檔上做。先貼代碼，稍后解釋，代碼如下：

配置 交換機，隊列，交換機與隊列的綁定，消息監視容器：

@Configuration @Data public class RabbitMQConfig {final static String queueName = "spring-boot";@BeanQueue queue() {return new Queue(queueName, false);}@BeanTopicExchange exchange() {return new TopicExchange("spring-boot-exchange");}@BeanBinding binding(Queue queue, TopicExchange exchange) {return BindingBuilder.bind(queue).to(exchange).with(queueName);}@BeanSimpleMessageListenerContainer container(ConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter) {SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();container.setConnectionFactory(connectionFactory);container.setQueueNames(queueName);container.setMessageListener(listenerAdapter);return container;}@BeanReceiver receiver() {return new Receiver();}@BeanMessageListenerAdapter listenerAdapter(Receiver receiver) {return new MessageListenerAdapter(receiver, "receiveMessage");} }

配置接收信息者（即消費者）：

public class Receiver {private CountDownLatch latch = new CountDownLatch(1);public void receiveMessage(String message) {System.out.println("Received <" + message + ">");latch.countDown();}public CountDownLatch getLatch() {return latch;} }

配置發送信息者（即生產者）：

@RestController public class Test {@AutowiredRabbitTemplate rabbitTemplate;@RequestMapping(value = "/test/{abc}",method = RequestMethod.GET)public String test(@PathVariable(value = "abc") String abc){rabbitTemplate.convertAndSend("spring-boot", abc + " from RabbitMQ!");return "abc";} }

以上便可實現一個簡單的 RabbitMQ Demo，具體代碼在：點這里

那么，這里，分為三個部分分析：發消息，交換機隊列，收消息。

• 對于發送消息：我們一般可以使用 RabbitTemplate，這個是 Spring 封裝給了我們，便于我們發送信息，我們調用?rabbitTemplate.convertAndSend("spring-boot", xxx);?即可發送信息。
• 對于交換機隊列：如上代碼，我們需要配置交換機?TopicExchange，配置隊列?Queue，并且配置他們之間的綁定?Binding
• 對于接受消息：首先需要創建一個消息監聽容器，然后把我們的接受者注冊到該容器中，這樣，隊列中有信息，那么就會調用接收者的對應的方法。如上代碼?container.setMessageListener(listenerAdapter);?其中，MessageListenerAdapter 可以看做是 我們接收者的一個包裝類，new MessageListenerAdapter(receiver, "receiveMessage");?指明了如果有消息來，那么調用接收者哪個方法進行處理。

RabbitMQ 的 Hello Demo（spring xml實現）

spring xml方式實現RabbitMQ簡單，可讀性較好，配置簡單，配置和實現如下所示。

上文已經講述了rabbitmq的配置，xml方式通過properites文件存放用戶配置信息：

mq.host=127.0.0.1 mq.username=guest mq.password=guest mq.port=5672

配置application-mq.xml配置文件，聲明連接、交換機、queue以及consumer監聽。

<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xmlns:rabbit="http://www.springframework.org/schema/rabbit"xmlns:context="http://www.springframework.org/schema/context"xsi:schemaLocation="http://www.springframework.org/schema/beanshttp://www.springframework.org/schema/beans/spring-beans-3.0.xsdhttp://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsdhttp://www.springframework.org/schema/rabbithttp://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd" ><description>rabbitmq 連接服務配置</description><!-- 連接配置 --><context:property-placeholder location="classpath:mq.properties" /><rabbit:connection-factory id="connectionFactory" host="${mq.host}" username="${mq.username}" password="${mq.password}" port="${mq.port}"/><rabbit:admin connection-factory="connectionFactory"/><!-- spring template聲明--><rabbit:template exchange="amqpExchange" id="amqpTemplate" connection-factory="connectionFactory" /><!--申明queue--><rabbit:queue id="test_queue_key" name="test_queue_key" durable="true" auto-delete="false" exclusive="false" /><!--申明exchange交換機并綁定queue--><rabbit:direct-exchange name="amqpExchange" durable="true" auto-delete="false" id="amqpExchange"><rabbit:bindings><rabbit:binding queue="test_queue_key" key="test_queue_key"/></rabbit:bindings></rabbit:direct-exchange><!--consumer配置監聽--><bean id="reveiver" class="com.demo.mq.receive.Reveiver" /><rabbit:listener-container connection-factory="connectionFactory" acknowledge="auto"><rabbit:listener queues="test_queue_key" ref="reveiver" method="receiveMessage"/></rabbit:listener-container> </beans>

上述代碼中，引入properties文件就不多說了。

<rabbit:connection-factory>標簽聲明創建connection的factory工廠。

<rabbit-template>聲明spring template，和上文spring中使用template一樣。template可聲明exchange。

<rabbit:queue>聲明一個queue并設置queue的配置項，直接看標簽屬性就可以明白queue的配置項。

<rabbit:direct-exchange>聲明交換機并綁定queue。

<rabbit:listener-container>申明監聽container并配置consumer和監聽routing-key。

剩下就簡單了，application-context.xml中把rabbitmq配置import進去。

<?xml version="1.0" encoding="UTF-8"?> <beansxmlns="http://www.springframework.org/schema/beans"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xmlns:task="http://www.springframework.org/schema/task"xmlns:context="http://www.springframework.org/schema/context"xmlns:aop="http://www.springframework.org/schema/aop"xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsdhttp://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsdhttp://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.0.xsdhttp://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-3.0.xsd"><context:component-scan base-package="com.demo.**" /><import resource="application-mq.xml" /> </beans>

Producer實現，發送消息還是使用template的convertAndSend() deliver消息。

@Service public class Producer {@Autowiredprivate AmqpTemplate amqpTemplate;private final static Logger logger = LoggerFactory.getLogger(Producer.class);public void sendDataToQueue(String queueKey, Object object) {try {amqpTemplate.convertAndSend(queueKey, object);} catch (Exception e) {e.printStackTrace();logger.error("exeception={}",e);}} }

配置consumer

package com.demo.mq.receive;import org.springframework.stereotype.Service; import java.util.concurrent.CountDownLatch;@Service public class Reveiver {private CountDownLatch latch = new CountDownLatch(1);public void receiveMessage(String message) {System.out.println("reveice msg=" + message.toString());latch.countDown();} }

測試deliver消息

Controller @RequestMapping("/demo/") public class TestController {private final static Logger logger = LoggerFactory.getLogger(TestController.class);@Resourceprivate Producer producer;@RequestMapping("/test/{msg}")public String send(@PathVariable("msg") String msg){logger.info("#TestController.send#abc={msg}", msg);System.out.println("msg="+msg);producer.sendDataToQueue("test_queue_key",msg);return "index";} }

RabbitMQ 在生產環境下運用和出現的問題

在生產環境中，由于 Spring 對 RabbitMQ 提供了一些方便的注解，所以首先可以使用這些注解。例如：

• @EnableRabbit：@EnableRabbit 和 @Configuration 注解在一個類中結合使用，如果該類能夠返回一個 RabbitListenerContainerFactory 類型的 bean，那么就相當于能夠把該終端（消費端）和 RabbitMQ 進行連接。Ps：（生成端不是通過 RabbitListenerContainerFactory 來和 RabbitMQ 連接，而是通過 RabbitTemplate ）
• @RabbitListener：當對應的隊列中有消息的時候，該注解修飾下的方法會被執行。
• @RabbitHandler：接收者可以監聽多個隊列，不同的隊列消息的類型可能不同，該注解可以使得不同的消息讓不同方法來響應。

具體這些注解的使用，可以參考這里的代碼：點這里

首先，生產環境下的 RabbitMQ 可能不會在生產者或者消費者本機上，所以需要重新定義 ConnectionFactory，即：

@Bean ConnectionFactory connectionFactory() {CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host, port);connectionFactory.setUsername(userName);connectionFactory.setPassword(password);connectionFactory.setVirtualHost(vhost);return connectionFactory; }

這里，可以重新設置需要連接的 RabbitMQ 的 ip，端口，虛擬主機，用戶名，密碼。

然后，可以先從生產端考慮，生產端需要連接 RabbitMQ，那么可以通過 RabbitTemplate 進行連接。 Ps：（RabbitTemplate 用于生產端發送消息到交換機中），如下代碼：

@Bean(name="myTemplate") RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {RabbitTemplate template = new RabbitTemplate(connectionFactory);template.setMessageConverter(integrationEventMessageConverter());template.setExchange(exchangeName);return template; }

在該代碼中，new RabbitTemplate(connectionFactory);?設置了生產端連接到RabbitMQ，template.setMessageConverter(integrationEventMessageConverter());?設置了 生產端發送給交換機的消息是以什么格式的，在?integrationEventMessageConverter()?代碼中：

public MessageConverter integrationEventMessageConverter() {Jackson2JsonMessageConverter messageConverter = new Jackson2JsonMessageConverter();return messageConverter; }

如上?Jackson2JsonMessageConverter?指明了 JSON。上述代碼的最后?template.setExchange(exchangeName);?指明了 要把生產者要把消息發送到哪個交換機上。

有了上述，那么，我們即可使用?rabbitTemplate.convertAndSend("spring-boot", xxx);?發送消息，xxx 表示任意類型，因為上述的設置會幫我們把這些類型轉化成 JSON 傳輸。

接著，生產端發送我們說過了，那么現在可以看看消費端：

對于消費端，我們可以只創建?SimpleRabbitListenerContainerFactory，它能夠幫我們生成 RabbitListenerContainer，然后我們再使用 @RabbitListener 指定接收者收到信息時處理的方法。

@Bean(name="myListenContainer") public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() {SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();factory.setMessageConverter(integrationEventMessageConverter());factory.setConnectionFactory(connectionFactory());return factory; }

這其中?factory.setMessageConverter(integrationEventMessageConverter());?指定了我們接受消息的時候，以 JSON 傳輸的消息可以轉換成對應的類型傳入到方法中。例如：

@Slf4j @Component @RabbitListener(containerFactory = "helloRabbitListenerContainer",queues = "spring-boot") public class Receiver {@RabbitHandlerpublic void receiveTeacher(Teacher teacher) {log.info("##### = {}",teacher);} }

可能出現的問題：

消息持久化

在生產環境中，我們需要考慮萬一生產者掛了，消費者掛了，或者 rabbitmq 掛了怎么樣。一般來說，如果生產者掛了或者消費者掛了，其實是沒有影響，因為消息就在隊列里面。那么萬一 rabbitmq 掛了，之前在隊列里面的消息怎么辦，其實可以做消息持久化，RabbitMQ 會把信息保存在磁盤上。

做法是可以先從 Connection 對象中拿到一個 Channel 信道對象，然后再可以通過該對象設置 消息持久化。

生產者或者消費者斷線重連

這里 Spring 有自動重連機制。

ACK 確認機制

每個Consumer可能需要一段時間才能處理完收到的數據。如果在這個過程中，Consumer出錯了，異常退出了，而數據還沒有處理完成，那么 非常不幸，這段數據就丟失了。因為我們采用no-ack的方式進行確認，也就是說，每次Consumer接到數據后，而不管是否處理完 成，RabbitMQ Server會立即把這個Message標記為完成，然后從queue中刪除了。

如果一個Consumer異常退出了，它處理的數據能夠被另外的Consumer處理，這樣數據在這種情況下就不會丟失了（注意是這種情況下）。
為了保證數據不被丟失，RabbitMQ支持消息確認機制，即acknowledgments。為了保證數據能被正確處理而不僅僅是被Consumer收到，那么我們不能采用no-ack。而應該是在處理完數據后發送ack。

在處理數據后發送的ack，就是告訴RabbitMQ數據已經被接收，處理完成，RabbitMQ可以去安全的刪除它了。
如果Consumer退出了但是沒有發送ack，那么RabbitMQ就會把這個Message發送到下一個Consumer。這樣就保證了在Consumer異常退出的情況下數據也不會丟失。

個人對 RabbitMQ ACK 的一些疑問，求助：點這里

總結

RabbitMQ 作用：異步，解耦，緩沖，消息分發。

RabbitMQ 主要分為3個部分，生產者，交換機和隊列，消費者。

需要注意消息持久化，目的為了防止 RabbitMQ 宕機；考慮 ACK 機制，目的為了如果消費者對消息的處理失敗了，那么后續要如何處理。

寫在最后

寫出來，說出來才知道對不對，知道不對才能改正，改正了才能成長。

在技術方面，希望大家眼里都容不得沙子。如果有不對的地方或者需要改進的地方希望可以指出，萬分感謝。

?

文／Mooner_guo（簡書作者）
原文鏈接：http://www.jianshu.com/p/a5f7fce67803
著作權歸作者所有，轉載請聯系作者獲得授權，并標注“簡書作者”

總結

以上是生活随笔為你收集整理的RabbitMQ exchange交换机机制的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。