RabbitMQ 即一個消息隊列久妆，主要是用來實現應用程序的異步和解耦缩麸，同時也能起到消息緩沖窒升，消息分發(fā)的作用。
消息中間件在互聯網公司的使用中越來越多慕匠，剛才還看到新聞阿里將RocketMQ捐獻給了apache饱须，當然了今天的主角還是講RabbitMQ。消息中間件最主要的作用是解耦台谊，中間件最標準的用法是生產者生產消息傳送到隊列蓉媳，消費者從隊列中拿取消息并處理，生產者不用關心是誰來消費锅铅，消費者不用關心誰在生產消息酪呻，從而達到解耦的目的。在分布式的系統(tǒng)中盐须，消息隊列也會被用在很多其它的方面玩荠，比如：分布式事務的支持，RPC的調用等等贼邓。
以前一直使用的是ActiveMQ阶冈，在實際的生產使用中也出現了一些小問題，在網絡查閱了很多的資料后塑径，決定嘗試使用RabbitMQ來替換ActiveMQ女坑，RabbitMQ的高可用性、高性能统舀、靈活性等一些特點吸引了我們匆骗，查閱了一些資料整理出此文劳景。
RabbitMQ是實現AMQP（高級消息隊列協(xié)議）的消息中間件的一種，最初起源于金融系統(tǒng)碉就，用于在分布式系統(tǒng)中存儲轉發(fā)消息盟广，在易用性、擴展性铝噩、高可用性等方面表現不俗衡蚂。RabbitMQ主要是為了實現系統(tǒng)之間的雙向解耦而實現的。當生產者大量產生數據時骏庸，消費者無法快速消費毛甲，那么需要一個中間層。保存這個數據具被。
AMQP玻募，即Advanced Message Queuing Protocol，高級消息隊列協(xié)議一姿，是應用層協(xié)議的一個開放標準七咧，為面向消息的中間件設計。消息中間件主要用于組件之間的解耦叮叹，消息的發(fā)送者無需知道消息使用者的存在艾栋，反之亦然。AMQP的主要特征是面向消息蛉顽、隊列蝗砾、路由（包括點對點和發(fā)布/訂閱）、可靠性携冤、安全悼粮。
RabbitMQ是一個開源的AMQP實現，服務器端用Erlang語言編寫曾棕，支持多種客戶端扣猫，如：Python、Ruby翘地、.NET申尤、Java、JMS子眶、C瀑凝、PHP、ActionScript臭杰、XMPP粤咪、STOMP等，支持AJAX渴杆。用于在分布式系統(tǒng)中存儲轉發(fā)消息寥枝，在易用性宪塔、擴展性、高可用性等方面表現不俗囊拜。
通常我們談到隊列服務, 會有三個概念： 發(fā)消息者某筐、隊列、收消息者冠跷，RabbitMQ 在這個基本概念之上, 多做了一層抽象, 在發(fā)消息者和 隊列之間, 加入了交換器 (Exchange). 這樣發(fā)消息者和隊列就沒有直接聯系, 轉而變成發(fā)消息者把消息給交換器, 交換器根據調度策略再把消息再給隊列南誊。
左側 P 代表 生產者，也就是往 RabbitMQ 發(fā)消息的程序蜜托。
中間即是 RabbitMQ抄囚，其中包括了 交換機 和 隊列。
右側 C 代表 消費者橄务，也就是往 RabbitMQ 拿消息的程序幔托。
那么，其中比較重要的概念有 4 個蜂挪，分別為：虛擬主機重挑，交換機，隊列棠涮，和綁定谬哀。
虛擬主機：一個虛擬主機持有一組交換機、隊列和綁定严肪。為什么需要多個虛擬主機呢玻粪？很簡單，RabbitMQ當中诬垂，用戶只能在虛擬主機的粒度進行權限控制。因此伦仍，如果需要禁止A組訪問B組的交換機/隊列/綁定结窘，必須為A和B分別創(chuàng)建一個虛擬主機。每一個RabbitMQ服務器都有一個默認的虛擬主機“/”充蓝。
交換機：Exchange 用于轉發(fā)消息隧枫，但是它不會做存儲，如果沒有 Queue bind 到 Exchange 的話谓苟，它會直接丟棄掉 Producer 發(fā)送過來的消息官脓。
這里有一個比較重要的概念：路由鍵。消息到交換機的時候涝焙，交互機會轉發(fā)到對應的隊列中卑笨，那么究竟轉發(fā)到哪個隊列，就要根據該路由鍵仑撞。
綁定：也就是交換機需要和隊列相綁定赤兴，這其中如上圖所示妖滔，是多對多的關系。
交換機的功能主要是接收消息并且轉發(fā)到綁定的隊列桶良，交換機不存儲消息座舍，在啟用ack模式后，交換機找不到隊列會返回錯誤陨帆。交換機有四種類型：Direct, topic, Headers and Fanout
Direct：direct 類型的行為是"先匹配, 再投送". 即在綁定時設定一個routing_key, 消息的routing_key匹配時, 才會被交換器投送到綁定的隊列中去.
Topic：按規(guī)則轉發(fā)消息（最靈活）
Headers：設置header attribute參數類型的交換機
Fanout：轉發(fā)消息到所有綁定隊列
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 轉發(fā)消息主要是根據通配符嵌赠。在這種交換機下塑荒，隊列和交換機的綁定會定義一種路由模式，那么姜挺，通配符就要在這種路由模式和路由鍵之間匹配后交換機才能轉發(fā)消息齿税。
在這種交換機模式下：
路由鍵必須是一串字符，用句號（.） 隔開炊豪，比如說 agreements.us凌箕，或者 agreements.eu.stockholm 等。
路由模式必須包含一個 星號（*）词渤，主要用于匹配路由鍵指定位置的一個單詞牵舱，比如說，一個路由模式是這樣子：agreements..b.*缺虐，那么就只能匹配路由鍵是這樣子的：第一個單詞是 agreements芜壁，第四個單詞是 b。 井號（#）就表示相當于一個或者多個單詞，例如一個匹配模式是agreements.eu.berlin.#沿盅，那么把篓，以agreements.eu.berlin開頭的路由鍵都是可以的。
具體代碼發(fā)送的時候還是一樣腰涧，第一個參數表示交換機韧掩，第二個參數表示routing key，第三個參數即消息窖铡。如下：
rabbitTemplate.convertAndSend("testTopicExchange","key1.a.c.key2"," this is? RabbitMQ!");
topic 和 direct 類似, 只是匹配上支持了"模式", 在"點分"的 routing_key 形式中, 可以使用兩個通配符:
*表示一個詞.
#表示零個或多個詞.
Headers Exchange
headers 也是根據規(guī)則匹配, 相較于 direct 和 topic 固定地使用 routing_key , headers 則是一個自定義匹配規(guī)則的類型.
在隊列與交換器綁定時, 會設定一組鍵值對規(guī)則, 消息中也包括一組鍵值對( headers 屬性), 當這些鍵值對有一對, 或全部匹配時, 消息被投送到對應隊列.
Fanout Exchange
Fanout Exchange 消息廣播的模式疗锐，不管路由鍵或者是路由模式，會把消息發(fā)給綁定給它的全部隊列费彼，如果配置了routing_key會被忽略滑臊。
springboot集成RabbitMQ非常簡單，如果只是簡單的使用配置非常少箍铲，springboot提供了spring-boot-starter-amqp項目對消息各種支持雇卷。
1、配置pom包颠猴，主要是添加spring-boot-starter-amqp的支持
org.springframework.bootspring-boot-starter-amqp
2关划、配置文件
配置rabbitmq的安裝地址、端口以及賬戶信息
spring.application.name=spirng-boot-rabbitmqspring.rabbitmq.host=192.168.0.86spring.rabbitmq.port=5672spring.rabbitmq.username=adminspring.rabbitmq.password=123456
3翘瓮、隊列配置
@ConfigurationpublicclassRabbitConfig{@BeanpublicQueueQueue(){returnnewQueue("hello");? ? }}
3贮折、發(fā)送者
rabbitTemplate是springboot 提供的默認實現
publicclassHelloSender{@AutowiredprivateAmqpTemplate rabbitTemplate;publicvoidsend(){? ? ? ? String context ="hello "+newDate();? ? ? ? System.out.println("Sender : "+ context);this.rabbitTemplate.convertAndSend("hello", context);? ? }}
4、接收者
@Component@RabbitListener(queues ="hello")publicclassHelloReceiver{@RabbitHandlerpublicvoidprocess(String hello){? ? ? ? System.out.println("Receiver? : "+ hello);? ? }}
5资盅、測試
@RunWith(SpringRunner.class)@SpringBootTestpublicclassRabbitMqHelloTest{@AutowiredprivateHelloSender helloSender;@Testpublicvoidhello()throwsException{? ? ? ? helloSender.send();? ? }}
注意调榄，發(fā)送者和接收者的queue name必須一致，不然不能接收
一個發(fā)送者呵扛，N個接收者或者N個發(fā)送者和N個接收者會出現什么情況呢每庆？
一對多發(fā)送
對上面的代碼進行了小改造,接收端注冊了兩個Receiver,Receiver1和Receiver2，發(fā)送端加入參數計數今穿，接收端打印接收到的參數扣孟，下面是測試代碼，發(fā)送一百條消息荣赶，來觀察兩個接收端的執(zhí)行效果
@TestpublicvoidoneToMany()throwsException{for(inti=0;i<100;i++){? ? ? ? neoSender.send(i);? ? }}
結果如下：
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
根據返回結果得到以下結論
一個發(fā)送者，N個接受者,經過測試會均勻的將消息發(fā)送到N個接收者中
多對多發(fā)送
復制了一份發(fā)送者鸽斟，加入標記拔创，在一百個循環(huán)中相互交替發(fā)送
@TestpublicvoidmanyToMany()throwsException{for(inti=0;i<100;i++){? ? ? ? ? ? neoSender.send(i);? ? ? ? ? ? neoSender2.send(i);? ? ? ? }}
結果如下：
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
結論：和一對多一樣，接收端仍然會均勻接收到消息
對象的支持
springboot以及完美的支持對象的發(fā)送和接收富蓄，不需要格外的配置剩燥。
//發(fā)送者publicvoidsend(User user){? ? System.out.println("Sender object: "+ user.toString());this.rabbitTemplate.convertAndSend("object", user);}...//接受者@RabbitHandlerpublicvoidprocess(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規(guī)則配置，這里使用兩個隊列來測試
@ConfigurationpublicclassTopicRabbitConfig{finalstaticString message ="topic.message";finalstaticString messages ="topic.messages";@BeanpublicQueuequeueMessage(){returnnewQueue(TopicRabbitConfig.message);? ? }@BeanpublicQueuequeueMessages(){returnnewQueue(TopicRabbitConfig.messages);? ? }@BeanTopicExchangeexchange(){returnnewTopicExchange("exchange");? ? }@BeanBindingbindingExchangeMessage(Queue queueMessage, TopicExchange exchange){returnBindingBuilder.bind(queueMessage).to(exchange).with("topic.message");? ? }@BeanBindingbindingExchangeMessages(Queue queueMessages, TopicExchange exchange){returnBindingBuilder.bind(queueMessages).to(exchange).with("topic.#");? ? }}
使用queueMessages同時匹配兩個隊列灭红，queueMessage只匹配"topic.message"隊列
publicvoidsend1(){? ? String context ="hi, i am message 1";? ? System.out.println("Sender : "+ context);this.rabbitTemplate.convertAndSend("exchange","topic.message", context);}publicvoidsend2(){? ? 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 相關配置
@ConfigurationpublicclassFanoutRabbitConfig{@BeanpublicQueueAMessage(){returnnewQueue("fanout.A");? ? }@BeanpublicQueueBMessage(){returnnewQueue("fanout.B");? ? }@BeanpublicQueueCMessage(){returnnewQueue("fanout.C");? ? }@BeanFanoutExchangefanoutExchange(){returnnewFanoutExchange("fanoutExchange");? ? }@BeanBindingbindingExchangeA(Queue AMessage,FanoutExchange fanoutExchange){returnBindingBuilder.bind(AMessage).to(fanoutExchange);? ? }@BeanBindingbindingExchangeB(Queue BMessage, FanoutExchange fanoutExchange){returnBindingBuilder.bind(BMessage).to(fanoutExchange);? ? }@BeanBindingbindingExchangeC(Queue CMessage, FanoutExchange fanoutExchange){returnBindingBuilder.bind(CMessage).to(fanoutExchange);? ? }}
這里使用了A、B娇斑、C三個隊列綁定到Fanout交換機上面策添，發(fā)送端的routing_key寫任何字符都會被忽略：
publicvoidsend(){? ? ? ? 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
