轉(zhuǎn)載?2017年11月01日 09:54:03
2595
RabbitMQ 即一個(gè)消息隊(duì)列潜的，主要是用來(lái)實(shí)現(xiàn)應(yīng)用程序的異步和解耦骚揍，同時(shí)也能起到消息緩沖，消息分發(fā)的作用啰挪。
消息中間件在互聯(lián)網(wǎng)公司的使用中越來(lái)越多信不，剛才還看到新聞阿里將RocketMQ捐獻(xiàn)給了apache，當(dāng)然了今天的主角還是講RabbitMQ亡呵。消息中間件最主要的作用是解耦抽活，中間件最標(biāo)準(zhǔn)的用法是生產(chǎn)者生產(chǎn)消息傳送到隊(duì)列，消費(fèi)者從隊(duì)列中拿取消息并處理政己，生產(chǎn)者不用關(guān)心是誰(shuí)來(lái)消費(fèi)酌壕，消費(fèi)者不用關(guān)心誰(shuí)在生產(chǎn)消息掏愁，從而達(dá)到解耦的目的。在分布式的系統(tǒng)中卵牍，消息隊(duì)列也會(huì)被用在很多其它的方面果港，比如：分布式事務(wù)的支持，RPC的調(diào)用等等糊昙。
以前一直使用的是ActiveMQ辛掠，在實(shí)際的生產(chǎn)使用中也出現(xiàn)了一些小問(wèn)題，在網(wǎng)絡(luò)查閱了很多的資料后释牺，決定嘗試使用RabbitMQ來(lái)替換ActiveMQ萝衩，RabbitMQ的高可用性、高性能没咙、靈活性等一些特點(diǎn)吸引了我們猩谊，查閱了一些資料整理出此文。
RabbitMQ是實(shí)現(xiàn)AMQP（高級(jí)消息隊(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)者無(wú)法快速消費(fèi)，那么需要一個(gè)中間層棒口。保存這個(gè)數(shù)據(jù)寄月。
AMQP，即Advanced Message Queuing Protocol陌凳，高級(jí)消息隊(duì)列協(xié)議剥懒，是應(yīng)用層協(xié)議的一個(gè)開放標(biāo)準(zhǔn)，為面向消息的中間件設(shè)計(jì)合敦。消息中間件主要用于組件之間的解耦，消息的發(fā)送者無(wú)需知道消息使用者的存在验游，反之亦然充岛。AMQP的主要特征是面向消息、隊(duì)列耕蝉、路由（包括點(diǎn)對(duì)點(diǎn)和發(fā)布/訂閱）崔梗、可靠性、安全垒在。
RabbitMQ是一個(gè)開源的AMQP實(shí)現(xiàn)蒜魄，服務(wù)器端用Erlang語(yǔ)言編寫，支持多種客戶端，如：Python谈为、Ruby旅挤、.NET、Java伞鲫、JMS粘茄、C、PHP秕脓、ActionScript柒瓣、XMPP、STOMP等吠架，支持AJAX芙贫。用于在分布式系統(tǒng)中存儲(chǔ)轉(zhuǎn)發(fā)消息，在易用性傍药、擴(kuò)展性屹培、高可用性等方面表現(xiàn)不俗。
通常我們談到隊(duì)列服務(wù), 會(huì)有三個(gè)概念： 發(fā)消息者怔檩、隊(duì)列褪秀、收消息者，RabbitMQ 在這個(gè)基本概念之上, 多做了一層抽象, 在發(fā)消息者和 隊(duì)列之間, 加入了交換器 (Exchange). 這樣發(fā)消息者和隊(duì)列就沒(méi)有直接聯(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ī)呢渐排？很簡(jiǎn)單炬太，RabbitMQ當(dāng)中，用戶只能在虛擬主機(jī)的粒度進(jìn)行權(quán)限控制驯耻。?因此亲族，如果需要禁止A組訪問(wèn)B組的交換機(jī)/隊(duì)列/綁定炒考，必須為A和B分別創(chuàng)建一個(gè)虛擬主機(jī)。每一個(gè)RabbitMQ服務(wù)器都有一個(gè)默認(rèn)的虛擬主機(jī)“/”霎迫。
交換機(jī)：Exchange 用于轉(zhuǎn)發(fā)消息斋枢，但是它不會(huì)做存儲(chǔ)?，如果沒(méi)有 Queue bind 到 Exchange 的話女气，它會(huì)直接丟棄掉 Producer 發(fā)送過(guò)來(lái)的消息杏慰。
這里有一個(gè)比較重要的概念：路由鍵?炼鞠。消息到交換機(jī)的時(shí)候，交互機(jī)會(huì)轉(zhuǎn)發(fā)到對(duì)應(yīng)的隊(duì)列中谒主，那么究竟轉(zhuǎn)發(fā)到哪個(gè)隊(duì)列，就要根據(jù)該路由鍵霎肯。
綁定：也就是交換機(jī)需要和隊(duì)列相綁定，這其中如上圖所示观游，是多對(duì)多的關(guān)系。
交換機(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ī)模式，也是最簡(jiǎn)單的模式工碾，根據(jù)key全文匹配去尋找隊(duì)列弱睦。
第一個(gè) X - Q1 就有一個(gè) binding key，名字為 orange渊额； X - Q2 就有 2 個(gè) binding key况木，名字為 black 和 green。當(dāng)消息中的 路由鍵 和 這個(gè) binding key 對(duì)應(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ī)模式下：
路由鍵必須是一串字符嗜浮，用句號(hào)（.） 隔開羡亩，比如說(shuō) agreements.us，或者 agreements.eu.stockholm 等危融。
路由模式必須包含一個(gè) 星號(hào)（*）畏铆，主要用于匹配路由鍵指定位置的一個(gè)單詞，比如說(shuō)吉殃，一個(gè)路由模式是這樣子：agreements..b.*辞居，那么就只能匹配路由鍵是這樣子的：第一個(gè)單詞是 agreements，第四個(gè)單詞是 b蛋勺。 井號(hào)（#）就表示相當(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è)定一組鍵值對(duì)規(guī)則, 消息中也包括一組鍵值對(duì)( headers 屬性), 當(dāng)這些鍵值對(duì)有一對(duì), 或全部匹配時(shí), 消息被投送到對(duì)應(yīng)隊(duì)列.
Fanout Exchange
Fanout Exchange 消息廣播的模式趴樱，不管路由鍵或者是路由模式酪捡，會(huì)把消息發(fā)給綁定給它的全部隊(duì)列，如果配置了routing_key會(huì)被忽略捺疼。
springboot集成RabbitMQ非常簡(jiǎn)單永罚，如果只是簡(jiǎn)單的使用配置非常少卧秘，springboot提供了spring-boot-starter-amqp項(xiàng)目對(duì)消息各種支持翅敌。
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鸥滨、隊(duì)列配置
@ConfigurationpublicclassRabbitConfig{@BeanpublicQueueQueue(){returnnewQueue("hello");? ? }}
3、發(fā)送者
rabbitTemplate是springboot 提供的默認(rèn)實(shí)現(xiàn)
publicclassHelloSender{@AutowiredprivateAmqpTemplate rabbitTemplate;public void send(){? ? ? ? String context ="hello "+newDate();? ? ? ? System.out.println("Sender : "+ context);this.rabbitTemplate.convertAndSend("hello", context);? ? }}
4老速、接收者
@Component@RabbitListener(queues ="hello")publicclassHelloReceiver{@RabbitHandlerpublic void process(String hello){? ? ? ? System.out.println("Receiver? : "+ hello);? ? }}
5凸主、測(cè)試
@RunWith(SpringRunner.class)@SpringBootTestpublicclassRabbitMqHelloTest{@AutowiredprivateHelloSender helloSender;@Testpublic void hello()throws Exception {? ? ? ? helloSender.send();? ? }}
注意，發(fā)送者和接收者的queue name必須一致旁舰，不然不能接收
一個(gè)發(fā)送者嗡官，N個(gè)接收者或者N個(gè)發(fā)送者和N個(gè)接收者會(huì)出現(xiàn)什么情況呢？
一對(duì)多發(fā)送
對(duì)上面的代碼進(jìn)行了小改造,接收端注冊(cè)了兩個(gè)Receiver,Receiver1和Receiver2磺樱，發(fā)送端加入?yún)?shù)計(jì)數(shù)婆咸，接收端打印接收到的參數(shù)，下面是測(cè)試代碼块差，發(fā)送一百條消息，來(lái)觀察兩個(gè)接收端的執(zhí)行效果
@Testpublic void oneToMany()throws Exception {for(inti=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)過(guò)測(cè)試會(huì)均勻的將消息發(fā)送到N個(gè)接收者中
多對(duì)多發(fā)送
復(fù)制了一份發(fā)送者起趾，加入標(biāo)記训裆，在一百個(gè)循環(huán)中相互交替發(fā)送
@Testpublic void manyToMany()throws Exception {for(inti=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é)論：和一對(duì)多一樣边琉，接收端仍然會(huì)均勻接收到消息
對(duì)象的支持
springboot以及完美的支持對(duì)象的發(fā)送和接收记劝，不需要格外的配置。
//發(fā)送者public void send(User user){? ? System.out.println("Sender object: "+ user.toString());this.rabbitTemplate.convertAndSend("object", user);}...//接受者@RabbitHandlerpublic 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ì)列
首先對(duì)topic規(guī)則配置怒竿，這里使用兩個(gè)隊(duì)列來(lái)測(cè)試
@ConfigurationpublicclassTopicRabbitConfig{finalstaticString message ="topic.message";finalstaticString messages ="topic.messages";@Beanpublic Queue queueMessage(){returnnewQueue(TopicRabbitConfig.message);? ? }@Beanpublic Queue queueMessages(){returnnewQueue(TopicRabbitConfig.messages);? ? }@BeanTopicExchange exchange(){returnnewTopicExchange("exchange");? ? }@BeanBinding bindingExchangeMessage(Queue queueMessage, TopicExchange exchange){returnBindingBuilder.bind(queueMessage).to(exchange).with("topic.message");? ? }@BeanBinding bindingExchangeMessages(Queue queueMessages, TopicExchange exchange){returnBindingBuilder.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)聽(tīng)到消息
Fanout Exchange
Fanout 就是我們熟悉的廣播模式或者訂閱模式饭弓，給Fanout交換機(jī)發(fā)送消息媒抠，綁定了這個(gè)交換機(jī)的所有隊(duì)列都收到這個(gè)消息。
Fanout 相關(guān)配置
@ConfigurationpublicclassFanoutRabbitConfig{@Beanpublic Queue AMessage(){returnnewQueue("fanout.A");? ? }@Beanpublic Queue BMessage(){returnnewQueue("fanout.B");? ? }@Beanpublic Queue CMessage(){returnnewQueue("fanout.C");? ? }@BeanFanoutExchange fanoutExchange(){returnnewFanoutExchange("fanoutExchange");? ? }@BeanBinding bindingExchangeA(Queue AMessage,FanoutExchange fanoutExchange){returnBindingBuilder.bind(AMessage).to(fanoutExchange);? ? }@BeanBinding bindingExchangeB(Queue BMessage, FanoutExchange fanoutExchange){returnBindingBuilder.bind(BMessage).to(fanoutExchange);? ? }@BeanBinding bindingExchangeC(Queue CMessage, FanoutExchange fanoutExchange){returnBindingBuilder.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é)果說(shuō)明排霉，綁定到fanout交換機(jī)上面的隊(duì)列都收到了消息
RabbitMQ：Spring 集成 RabbitMQ 與其概念，消息持久化瑰钮，ACK機(jī)制
作者：純潔的微笑
版權(quán)所有，歡迎保留原文鏈接進(jìn)行轉(zhuǎn)載：)
