干货！基于SpringBoot的RabbitMQ多种模式队列

　　环境准备安装RabbitMQ
　　由于RabbitMQ的安装比较简单，这里不再赘述。可自行到官网下载http：www。rabbitmq。comdownload。html依赖
　　SpringBoot项目导入依赖dependencygroupIdorg。springframework。bootgroupIdspringbootstarteramqpartifactIddependency连接配置
　　配置文件添加如下配置（根据自身情况修改配置）spring。rabbitmq。host127。0。0。1spring。rabbitmq。port5672spring。rabbitmq。usernameguestspring。rabbitmq。passwordguestspring。rabbitmq。virtualhostacelin
　　五种队列模式实现1点对点的队列
　　在java配置文件DirectRabbitConfig中先声明一个队列用于接收信息publicstaticfinalStringPEERTOPEERQUEUEpeertopeerqueue；点对点队列PeertopeerBeanpublicQueuepeer2peerQueue（）｛returnnewQueue（PEERTOPEERQUEUE，true）；｝
　　创建一个消费者类Peer2PeerConsumers。用RabbitListener对声明的队列进行监听ComponentpublicclassPeer2PeerConsumersextendsBase｛RabbitListener（queuesDirectRabbitConfig。PEERTOPEERQUEUE）publicvoidconsumer2（ObjecttestMessage）｛logger。debug（peertopeer消费者收到消息：testMessage。toString（））；｝｝
　　创造一个消息生产者。在编码形式上，直接把消息发发送给接收的消息队列【点对点模式】paramtask消息内容PostMapping（peertopeer｛task｝）publicStringpeerToPeer（PathVariable（task）Stringtask）｛rabbitTemplate。convertAndSend（DirectRabbitConfig。PEERTOPEERQUEUE，task）；returnok；｝
　　启动项目。队列绑定到默认交换机
　　调用生产者接口产生消息，可看到的消费者立即接收到信息peertopeer消费者收到消息：（Body：himqMessageProperties〔headers｛｝，contentTypetextplain，contentEncodingUTF8，contentLength0，receivedDeliveryModePERSISTENT，priority0，redeliveredfalse，receivedExchange，receivedRoutingKeypeertopeerqueue，deliveryTag1，consumerTagamq。ctagvuKWCYLNLn3GwRJKJO5Mg，consumerQueuepeertopeerqueue〕）
　　这里要说明一点的是，点对点模式虽然编码形式只与队列交互，但其本质上还是要跟交换机交互的，本质跟下面要介绍的路由模式其实是一样的。
　　查看convertAndSend方法的源码，可以看到我们虽然没有进行交换机和队列的绑定，发送消息是也没指定交换机，但是程序会为我们绑定默认的交换机。Thedefaultexchangeisimplicitlyboundtoeveryqueue，witharoutingkeyequaltothequeuename。Itisnotpossibletoexplicitlybindto，orunbindfromthedefaultexchange。Italsocannotbedeleted。
　　默认交换机会隐式绑定到每个队列，路由键等于队列名称。我们无法明确绑定到默认交换机或从默认交换中解除绑定。它也无法删除。
　　且我们第一个参数传递的是队列的名称，但实际上程序是以这个名字作为路由，将同名队列跟默认交换机做绑定。所以的消息会根据该路由信息，通过默认交换机分发到同名队列上。（我们通过接收的信息receivedRoutingKeypeertopeerqueue和consumerQueuepeertopeerqueue也可以看的出来）
　　2工作队列模式WorkQueue
　　在java配置文件DirectRabbitConfig中先声明一个工作队列publicstaticfinalStringWORKQUEUEworkqueue；工作队列WorkQueueBeanpublicQueueworkQueue（）｛returnnewQueue（WORKQUEUE，true）；｝
　　创建一个消费者类WorkConsumers。同样用RabbitListener对声明的队列进行监听ComponentpublicclassWorkConsumersextendsBase｛RabbitListener（queuesDirectRabbitConfig。WORKQUEUE）publicvoidconsumer1（ObjecttestMessage）｛logger。debug（work消费者〔1〕收到消息：testMessage。toString（））；｝RabbitListener（queuesDirectRabbitConfig。WORKQUEUE）publicvoidconsumer2（ObjecttestMessage）｛logger。debug（work消费者〔2〕收到消息：testMessage。toString（））；｝｝
　　创造一个消息生产者。在编码形式上，直接把消息发发送给接收的消息队列【工作队列模式】paramtask消息内容PostMapping（work｛task｝）publicStringsendWorkMessage（PathVariable（task）Stringtask）｛rabbitTemplate。convertAndSend（DirectRabbitConfig。WORKQUEUE，task）；returnok；｝
　　启动项目，同样的，工作队列也是绑定到默认交换机。
　　调用生产者接口连续发送几次消息，可看到两个消费者竞争对队列消息进行消费，一条消息只被一个消费者消费，不会出现重复消费的情况，因此工作队列模式也被称为竞争消费者模式。work消费者〔1〕收到消息：（Body：task1MessageProperties〔headers｛｝，contentTypetextplain，contentEncodingUTF8，contentLength0，receivedDeliveryModePERSISTENT，priority0，redeliveredfalse，receivedExchange，receivedRoutingKeyworkqueue，deliveryTag1，consumerTagamq。ctagPUYjfVq56aEn7a9DzLNzQ，consumerQueueworkqueue〕）work消费者〔2〕收到消息：（Body：task2MessageProperties〔headers｛｝，contentTypetextplain，contentEncodingUTF8，contentLength0，receivedDeliveryModePERSISTENT，priority0，redeliveredfalse，receivedExchange，receivedRoutingKeyworkqueue，deliveryTag1，consumerTagamq。ctag1IVtDalFUCKVvYpFrGF8A，consumerQueueworkqueue〕）work消费者〔1〕收到消息：（Body：task3MessageProperties〔headers｛｝，contentTypetextplain，contentEncodingUTF8，contentLength0，receivedDeliveryModePERSISTENT，priority0，redeliveredfalse，receivedExchange，receivedRoutingKeyworkqueue，deliveryTag2，consumerTagamq。ctagPUYjfVq56aEn7a9DzLNzQ，consumerQueueworkqueue〕）work消费者〔2〕收到消息：（Body：task4MessageProperties〔headers｛｝，contentTypetextplain，contentEncodingUTF8，contentLength0，receivedDeliveryModePERSISTENT，priority0，redeliveredfalse，receivedExchange，receivedRoutingKeyworkqueue，deliveryTag2，consumerTagamq。ctag1IVtDalFUCKVvYpFrGF8A，consumerQueueworkqueue〕）
　　事实上，竞争消费者模式本质就是多个消费者对同一个队列消息进行消费。另外，与点对点模式一样，工作队列模式的也是用到了默认交换机进行消息分发。因此于基于的Direct交换机的路由模式的原理本质上都是一样的，因此，某种程度上，我们也可以用路由模式实现工作队列模式，这点我们下面介绍路由模式再进行展开
　　3路由模式Routing
　　在java配置文件DirectRabbitConfig中先声明2个队列和一个direct类型的交换机，然后将队列1和与交换机用一个路由键1进行绑定，队列2用路由键2与队列进行绑定publicstaticfinalStringDIRECTQUEUEONEdirectQueue1；Direct队列名称1publicstaticfinalStringDIRECTQUEUETWOdirectQueue2；Direct队列名称2publicstaticfinalStringMYDIRECTEXCHANGEmyDirectExchange；Direct交换机名称publicstaticfinalStringROUTINGKEYONEdirect。routingkey1；direct路由标识1publicstaticfinalStringROUTINGKEYONEdirect。routingkey2；direct路由标识2DirectBeanpublicQueuedirectQueueOne（）｛returnnewQueue（DIRECTQUEUEONE，true）；｝BeanpublicQueuedirectQueueTwo（）｛returnnewQueue（DIRECTQUEUETWO，true）；｝BeanpublicDirectExchangedirectExchange（）｛returnnewDirectExchange（MYDIRECTEXCHANGE，true，false）；｝BeanpublicBindingbindingDirectOne（）｛returnBindingBuilder。bind（directQueueOne（））。to（directExchange（））。with（ROUTINGKEYONE）；｝BeanpublicBindingbindingDirectTwo（）｛returnBindingBuilder。bind（directQueueTwo（））。to（directExchange（））。with（ROUTINGKEYTWO）；｝
　　创建一个消费者类DirectConsumers。在每个消费者上，我们用3个消费者注解RabbitListener对声明的队列进行监听。消费者1和3监听队列1，消费者2监听队列2ComponentpublicclassDirectConsumersextendsBase｛RabbitListener（queuesDirectRabbitConfig。DIRECTQUEUEONE）publicvoidconsumer1（ObjecttestMessage）｛logger。debug（Direct消费者〔1〕收到消息：testMessage。toString（））；｝RabbitListener（queuesDirectRabbitConfig。DIRECTQUEUETWO）publicvoidconsumer2（ObjecttestMessage）｛logger。debug（Direct消费者〔2〕收到消息：testMessage。toString（））；｝RabbitListener（queuesDirectRabbitConfig。DIRECTQUEUEONE）publicvoidconsumer3（ObjecttestMessage）｛logger。debug（Direct消费者〔3〕收到消息：testMessage。toString（））；｝｝
　　创造一个消息生产者。发送消息时，带上路由键1信息【Direct路由模式】parammessage消息内容PostMapping（direct｛message｝）publicStringsendDirectMessage（PathVariable（message）Stringmessage）｛MapString，ObjectmapnewHashMap（）；map。put（messageId，String。valueOf（UUID。randomUUID（）））；map。put（messageData，message）；设置路由标识MYROUTINGKEY，发送到交换机MYDIRECTEXCHANGErabbitTemplate。convertAndSend（DirectRabbitConfig。MYDIRECTEXCHANGE，DirectRabbitConfig。ROUTINGKEYONE，map）；returnok；｝
　　启动项目，查看该交换机的绑定情况
　　发送多条信息，可以看到，由于队列2没有通过路由键1跟交换机进行绑定，所以对于监控队列2的消费者2，其无法结束到的带有路由键1的消息，而消费者1和3则竞争消费队列1的消息Direct消费者〔3〕收到消息：（Body：｛messageId54682b16014246afbe0c1156df1f27a7，messageDatamsg1｝MessageProperties〔headers｛｝，contentTypeapplicationxjavaserializedobject，contentLength0，receivedDeliveryModePERSISTENT，priority0，redeliveredfalse，receivedExchangemyDirectExchange，receivedRoutingKeydirect。routingkey1，deliveryTag15，consumerTagamq。ctagCsuZL9KKByH9IDtqTKefg，consumerQueuedirectQueue1〕）Direct消费者〔1〕收到消息：（Body：｛messageId66cd296a9a6044588e8772ed13f9964b，messageDatamsg2｝MessageProperties〔headers｛｝，contentTypeapplicationxjavaserializedobject，contentLength0，receivedDeliveryModePERSISTENT，priority0，redeliveredfalse，receivedExchangemyDirectExchange，receivedRoutingKeydirect。routingkey1，deliveryTag2，consumerTagamq。ctaghWmdY04YuLL0O2rgeSlxsw，consumerQueuedirectQueue1〕）Direct消费者〔3〕收到消息：（Body：｛messageId48c0830e220747ecbd3ea958fec48118，messageDatamsg3｝MessageProperties〔headers｛｝，contentTypeapplicationxjavaserializedobject，contentLength0，receivedDeliveryModePERSISTENT，priority0，redeliveredfalse，receivedExchangemyDirectExchange，receivedRoutingKeydirect。routingkey1，deliveryTag16，consumerTagamq。ctagCsuZL9KKByH9IDtqTKefg，consumerQueuedirectQueue1〕）
　　我们如果对新增一个队列3，通过路由键1与交换机进行绑定，消费者独立监听队列3，那么我们不难猜到，队列3将和队列1同样拿到一条消息，相当于广播的概念，但我们会发现如果要这么做，似乎路由键无足轻重，因此rabbitmq提供了一种特殊的交换机来处理这种场景，不需要路由键的参与。我们接着往下看
　　4发布订阅模式PublishSubscribe
　　在java配置文件DirectRabbitConfig中先声明Fanout交换机和两队列，并将两个队列与该交换机进行绑定publicstaticfinalStringMYFANOUTEXCHANGEmyFanoutExchange；Fanout交换机名称publicstaticfinalStringFANOUTQUEUEONEfanoutqueue1；Fanout队列名称1publicstaticfinalStringFANOUTQUEUETWOfanoutqueue2；Fanout队列名称2FanoutBeanpublicQueuefanoutQueueOne（）｛returnnewQueue（FANOUTQUEUEONE，true）；｝BeanpublicQueuefanoutQueueTwo（）｛returnnewQueue（FANOUTQUEUETWO，true）；｝BeanpublicFanoutExchangefanoutExchange（）｛returnnewFanoutExchange（MYFANOUTEXCHANGE，true，false）；｝BeanpublicBindingbindingFanoutOne（）｛returnBindingBuilder。bind（fanoutQueueOne（））。to（fanoutExchange（））；｝BeanpublicBindingbindingFanoutTwo（）｛returnBindingBuilder。bind（fanoutQueueTwo（））。to（fanoutExchange（））；｝
　　创建一个消费者类FanoutConsumers。创建两个消费者，分表对两个队列进行监听ComponentpublicclassFanoutConsumersextendsBase｛RabbitListener（queuesDirectRabbitConfig。FANOUTQUEUEONE）publicvoidconsumer1（ObjecttestMessage）｛logger。debug（FANOUT消费者〔1〕收到消息：testMessage。toString（））；｝RabbitListener（queuesDirectRabbitConfig。FANOUTQUEUETWO）publicvoidconsumer2（ObjecttestMessage）｛logger。debug（FANOUT消费者〔2〕收到消息：testMessage。toString（））；｝｝
　　创造一个消息生产者。将消息发送给Fanout交换机【工作队列模式】paramtask消息内容PostMapping（work｛task｝）publicStringsendWorkMessage（PathVariable（task）Stringtask）｛rabbitTemplate。convertAndSend（DirectRabbitConfig。WORKQUEUE，task）；returnok；｝
　　启动项目，我们可以看到交换机与两个队列进行了绑定，但是路由键那一栏是空的。
　　发送两条消息。【Fanout发布订阅模式】parammessage消息内容PostMapping（fanout｛message｝）publicStringsendFanoutMessage（PathVariable（message）Stringmessage）｛MapString，ObjectmapnewHashMap（）；map。put（messageId，String。valueOf（UUID。randomUUID（）））；map。put（messageData，message）；直接跟交换机MYFANOUTEXCHANGE交互rabbitTemplate。setExchange（DirectRabbitConfig。MYFANOUTEXCHANGE）；rabbitTemplate。convertAndSend（map）；returnok；｝
　　可以看到，两个消费者都拿到了同样的数据，达到了广播的效果。FANOUT消费者〔2〕收到消息：（Body：｛messageIda4bf19311db84cb98b01397f43a82660，messageDataHiFanout｝MessageProperties〔headers｛｝，contentTypeapplicationxjavaserializedobject，contentLength0，receivedDeliveryModePERSISTENT，priority0，redeliveredfalse，receivedExchangemyFanoutExchange，receivedRoutingKey，deliveryTag1，consumerTagamq。ctagncVmsRM7xHLZ0iAJT2tSTg，consumerQueuefanoutqueue2〕）FANOUT消费者〔1〕收到消息：（Body：｛messageIda4bf19311db84cb98b01397f43a82660，messageDataHiFanout｝MessageProperties〔headers｛｝，contentTypeapplicationxjavaserializedobject，contentLength0，receivedDeliveryModePERSISTENT，priority0，redeliveredfalse，receivedExchangemyFanoutExchange，receivedRoutingKey，deliveryTag1，consumerTagamq。ctagzR3Oi0MVESq8qushlAMa3Q，consumerQueuefanoutqueue1〕）FANOUT消费者〔1〕收到消息：（Body：｛messageId51f6672035dd4abf9d3324acf7786ed8，messageData666｝MessageProperties〔headers｛｝，contentTypeapplicationxjavaserializedobject，contentLength0，receivedDeliveryModePERSISTENT，priority0，redeliveredfalse，receivedExchangemyFanoutExchange，receivedRoutingKey，deliveryTag2，consumerTagamq。ctagzR3Oi0MVESq8qushlAMa3Q，consumerQueuefanoutqueue1〕）FANOUT消费者〔2〕收到消息：（Body：｛messageId51f6672035dd4abf9d3324acf7786ed8，messageData666｝MessageProperties〔headers｛｝，contentTypeapplicationxjavaserializedobject，contentLength0，receivedDeliveryModePERSISTENT，priority0，redeliveredfalse，receivedExchangemyFanoutExchange，receivedRoutingKey，deliveryTag2，consumerTagamq。ctagncVmsRM7xHLZ0iAJT2tSTg，consumerQueuefanoutqueue2〕）
　　5通配符模式Topics
　　在java配置文件DirectRabbitConfig中先声明一个Topic交换机、两个工作队列和三个通配绑定键，其中一个队列通过两个不同通配绑定键与交换机绑定，另外一个队列用第三个绑定键进行绑定。publicstaticfinalStringWORKQUEUEworkqueue；工作队列WorkQueueBeanpublicQueueworkQueue（）｛returnnewQueue（WORKQUEUE，true）；｝
　　通过rabbitmq管理页面我们可以看到交换机与队列的绑定变化，可以看到队列1车工绑定了两个通配键
　　创建一个消费者类TopicConsumers。创建两个消费者分别对两个队列做监听。ComponentpublicclassWorkConsumersextendsBase｛RabbitListener（queuesDirectRabbitConfig。WORKQUEUE）publicvoidconsumer1（ObjecttestMessage）｛logger。debug（work消费者〔1〕收到消息：testMessage。toString（））；｝RabbitListener（queuesDirectRabbitConfig。WORKQUEUE）publicvoidconsumer2（ObjecttestMessage）｛logger。debug（work消费者〔2〕收到消息：testMessage。toString（））；｝｝
　　创造一个消息生产者。发送3条不同的消息，分别带上三个不同的路由键【Topic通配符模式】parammessage消息内容PostMapping（topic｛message｝）publicStringsendTopicMessage（PathVariable（message）Stringmessage）｛MapString，ObjectmapnewHashMap（）；直接跟交换机MYFANOUTEXCHANGE交互rabbitTemplate。setExchange（DirectRabbitConfig。MYTOPICEXCHANGE）；map。put（messageId，String。valueOf（UUID。randomUUID（）））；map。put（messageData，messageTEST1）；rabbitTemplate。convertAndSend（DirectRabbitConfig。TOPICROUTINGKEYONE，map）；map。put（messageId，String。valueOf（UUID。randomUUID（）））；map。put（messageData，messageTEST2）；rabbitTemplate。convertAndSend（DirectRabbitConfig。TOPICROUTINGKEYTWO，map）；map。put（messageId，String。valueOf（UUID。randomUUID（）））；map。put（messageData，messageTEST3）；rabbitTemplate。convertAndSend（DirectRabbitConfig。TOPICROUTINGKEYTHREE，map）；returnok；｝
　　路由键声明如下：publicstaticfinalStringTOPICROUTINGKEYONEtopic。a1。b1。c1；topic路由键1publicstaticfinalStringTOPICROUTINGKEYTWOtopic。a1。b1；topic路由键2publicstaticfinalStringTOPICROUTINGKEYTHREEtopic。a2。b1；topic路由键3
　　启动项目，调用生产者的接口，查看两个消费者的消费情况。TOPIC消费者〔2〕收到消息：（Body：｛messageId82abd28211104f1ab09eae9a43c560c3，messageDatahitopic！TEST1｝MessageProperties〔headers｛｝，contentTypeapplicationxjavaserializedobject，contentLength0，receivedDeliveryModePERSISTENT，priority0，redeliveredfalse，receivedExchangemyTopicExchange，receivedRoutingKeytopic。a1。b1。c1，deliveryTag1，consumerTagamq。ctagwlRVC5xWiN8glrtA2i6uA，consumerQueuetopicqueue2〕）TOPIC消费者〔1〕收到消息：（Body：｛messageIdb203955775d847d593a02a03a38fabc7，messageDatahitopic！TEST2｝MessageProperties〔headers｛｝，contentTypeapplicationxjavaserializedobject，contentLength0，receivedDeliveryModePERSISTENT，priority0，redeliveredfalse，receivedExchangemyTopicExchange，receivedRoutingKeytopic。a1。b1，deliveryTag1，consumerTagamq。ctagF6ByjknEnCjh7XVolNfmcg，consumerQueuetopicqueue1〕）TOPIC消费者〔2〕收到消息：（Body：｛messageIdb203955775d847d593a02a03a38fabc7，messageDatahitopic！TEST2｝MessageProperties〔headers｛｝，contentTypeapplicationxjavaserializedobject，contentLength0，receivedDeliveryModePERSISTENT，priority0，redeliveredfalse，receivedExchangemyTopicExchange，receivedRoutingKeytopic。a1。b1，deliveryTag2，consumerTagamq。ctagwlRVC5xWiN8glrtA2i6uA，consumerQueuetopicqueue2〕）TOPIC消费者〔1〕收到消息：（Body：｛messageId3a8f3164706f4523bd2a4fee73595fbb，messageDatahitopic！TEST3｝MessageProperties〔headers｛｝，contentTypeapplicationxjavaserializedobject，contentLength0，receivedDeliveryModePERSISTENT，priority0，redeliveredfalse，receivedExchangemyTopicExchange，receivedRoutingKeytopic。a2。b1，deliveryTag2，consumerTagamq。ctagF6ByjknEnCjh7XVolNfmcg，consumerQueuetopicqueue1〕）
　　可以看到，路由键前缀为topic。a1的信息都可以被绑定了topic。a1。的队列接收到，而绑定了topic。a1。的队列只能接收到topic。a1后面带一个单词的信息，由于队列1还通过topic。。b1绑定交换机，因此携带路由键topic。a2。b1的信息同样也被队列1接收topic交换机是direct交换机做的改造的。两者的区别主要体现在路由键和绑定键格式上的限制不同。
　　路由键：必须是由点分隔的单词列表。单词形式不限。比如一个主题建：主题1。主题2。主题3
　　绑定键：格式上和路由键一致，但多了两个通配符和，代表任意数量的单词，包括0个。标识一个单词。
　　使用上，一个绑定键，我们可以看成是对一类具有多个特征的物体的一个抽象，由点分割的每个单词，我们可以看成一个主题或是一个特征。因此只要做好消息特征的归纳抽象，加上通配符的使用，我们就有很高的自由度去处理任意类型的消息
　　总结
　　以上就是关于RabbitMQ五种队列模式的实战演练，关于RabbitMQ其它实战与知识理解后续会相继分享，感兴趣的同学欢迎留言讨论