day11商城项目中集成rabbitmq实现消息处理46队列

JBossMessaging、JORAM、ActiveMQ、SunOpenMessageQueue、IBMMQ、ApacheQpidHTTPSQS当前使用较多的消息队列有RabbitMQMQActiveMQKafkaZeroMQ以及中间件服务的一个标准和API定义，而MQ则是遵循了AMQP协议的具体实现和产品。AMQP，即AdvancedMessageQueuingProtocol,一个提供统一消息服务的应用JMS，Java消息服务（JavaMessageService）Java平如果将来D系统接入，系统A还需要修改代码，过于麻烦！ AMQP，即AdvancedMessageQueuingProtocol，高级消息队列协议，是应用层协持AJAX。用于在分布式系统中转发消息，在易用性、扩展性、高可用性等方面表现不还是64位，以下的安装以window10(64位)和CentOS7-1611(64位)举例。官网地址输入 yumcleanallyumyumcleanallyumyum-yyum-yinstalltar-xvfmkdir-p-p创建上层 tar-xvfmkdir-p./configure--./configure--用于检测报错说找不到ncurses-develerlang，检测后会报很多依赖缺失，我们需要一个个都安装以后才可以继续安装erlang)yumyum-yinstallmakegccgcc-c++kernel-develm4ncurses-developenssl-develunixODBC-devel使用命令：makemake&&make预编译、编译与安装合并：makemake&&makesourcesourcevimvimexportPATH=$PATH:$ERLANG_HOME/bin检查erlang官网地址 运行 cdD:\RabbitMQServer\rabbitmq_server-rabbitmqctl.batstatus，如果出现以下的图，说明安装是成功的，并且说明 cdD:\RabbitMQServer\rabbitmq_server- 使用修改yum源的方式，修改方式和erlang方法一的一yumyum-yinstallrabbitmq-server-3.7.12-rabbitmq-pluginsenablerabbitmq-pluginsenablesystemctlsystemctlstartrabbitmq-CentOS6:servicerabbitmq-serverstatusCentOS7:systemctlCentOS6:servicerabbitmq-serverstatusCentOS7:systemctlstatusrabbitmq-server.service-A-AINPUT-ptcp-mstate--stateNEW-mtcp--dport15672-j-AINPUT-ptcp-mstate--stateNEW-mtcp--dport5672-jsystemctlrestartsystemctlrestartrpmrpm-qa|greprpm-qlrabbitmq-server-3.7.12-rpm-qlrabbitmq-server-3.7.12-{loopback_users,{loopback_users,systemctlrestartrabbitmq-systemctlrestartrabbitmq-guest角色为管理员，一般开发环境下会创建新的用户并对权限ego用户登录 不能managementplugin（管理插件imAMQPvirtualvirtualhostsqueuesexchanges和channels和virtualhosts的“全局”的统计信息，包含其他用户在这些virtualhosts中的活动。virtualhostsvirtualvirtualhosts的全局的统计信息创建和删除virtualhostsvirtual删除或更改。可通过将配置文件中loopback_users置空来取消其本地的限制：[{rabbit,[{loopback_users,用户仅能对其所能的virtualhosts中的资源进行操作。这里的资源指的是中的exchangequeuedeclaredelete分别需要exchangequeuequeuebindunbindqueueexchange的读权限发消息(publish)需exchange的写权限set_permissions[-p<vhostpath>]<user><conf><write>其中，<conf<write><read>的位置分别用正则表达式来匹配特定的资源，如可以匹配生成的和默认的RabbitMQconnectionchannel的权限验证结果、因此权 QueueRabbitMQ里。虽然消息流经RabbitMQ和Queue里才能被保存。Queue没有任何边界的限制，你想存多少消息都可以，它本质上是一个无限的缓存。许多生产者都可以向一个Queue里发送消息，许多消费者都可以从一个Queue里接收消息。在这部分的使用指南中，我们要用Java写两个程序；一个是生产者，他发送一个消细节，而是将注意要放在要做的这件事上，这件事就是发送一个"oWorld"消在下面的图中，"P"C"代表消费者。中间的就是一个Queue，一个 我们把消息发送者叫Send，消息接收者叫Recv。消息发送者连接RabbitMQ，发送packagepackageimportimportimportimportimportpublicclassSendprivatefinalstaticStringQUEUE_NAME=publicstaticvoidmain(String[]args)ConnectionFactoryfactory=newConnectionFactory();Connectionconnection=null;Channelchannel=null;tryconnection=channel=第三个参数Exclusive：排他队列，如果一个队列被 1.排他队列是基于连接可见的，同一连接的不同通道是可以同时 2."首次"，如果 channel.queueDeclare(QUEUE_NAME,false,false,false,Stringmessage= ochannel.basicPublish("",QUEUE_NAME,null,message.getBytes("UTF-8"));System.out.println("[x]Sent'"+message+"'");}catch(IOException{}catch(TimeoutException{}finallytryif(null!=channel&&channel.isOpen())if(null!=connection&&connection.isOpen())}catch(TimeoutExceptione)}catch(IOException{}}}}packagepackageimportimportimportpublicclassRecvprivatefinalstaticStringQUEUE_NAME= publicstaticvoidmain(String[]args)ConnectionFactoryfactory=newConnectionFactory();tryConnectionconnection=Channelchannel=channel.queueDeclare(QUEUE_NAME,false,false,false,null);System.out.println("[*]Waitingformessages.ToexitpressCTRL+C");

Consumerconsumer=new{publicvoidhandleDelivery(StringconsumerTag,EnvelopeAMQP.BasicPropertiesproperties,byte[]body)throwsIOException{Stringmessage=newString(body,"utf-8");System.out.println("[x]Received'"+message+"'");} channel.basicConsume(QUEUE_NAME,true, DeliverCallbackdeliverCallback=(consumerTag,delivery)-{Stringmessage=newString(delivery.getBody(),"UTF-8");System.out.println("[x]Received'"+message+"'");channel.basicConsume(QUEUE_NAME,true,deliverCallback,consumerTag->}catch(IOException{}catch(TimeoutException{}}}运行与，这时Message在实际应用中，可能会发生消费者收到Queue中的消息，但没有处理完成就宕机（或可以要求消费者在消费完消息后发送一个回执给RabbitMQ，RabbitMQ收到消息回执（Messageacknowledgment）QueueRabbitMQ没有收到回执并检测到消费者的RabbitMQ连接断开，则RabbitMQ会将该消息发送给其他消费者（如果存在多个消费者）timeout概念，一个消费者处理消息时间再长也不会导致该消息被发送给其他消费者，除非它的RabbitMQ连接断开。RabbitMQ，这将会导致严重的bug——Queue中堆积的消息会越来越多；消费者重启后会MessageMessage都设置为可持久化的（durable），这样可以保证绝大部分情况下我们的RabbitMQ服务器就断电了果我们需要对这种小概率也要管理起来，那么我们要用到事务。PrefetchQueue中的消息，Queue中的消息会prefetchCount来限制Queue每次发送给每个消费者的消息数，比如我们设置prefetchCount=1Queue每次给每个消费者发送一条消息；消费者处理完这条消息后Queue会再给该消费者发送一条消息。QueueRabbitMQ中这种事情都不会发生。实际的情况是，生产者将消息发送到Exchange（交换器，下图中的ExchangeQueue中（或者丢弃中的routingExchange的时候，一般会指定一个routingkey，来指定这及ExchangeType与bindingkey固定的情况下（在正常使用时一般这些内容都是固定配置好的ExchangeroutingkeyRabbitMQroutingkey255bytes关联起来，这样如何正确地将消息路由到指定的Queue了。BindingkeyQueueExchangeTypes章节会列bindingkey并不是在所有情况下都生效，它依赖于ExchangeType，比如fanoutExchange就会无视bindingkey，而是将消息路由到所有绑定到该Exchange的Exchange有ExchangeTypesystem与自定义，这里不予以描述路由到所有与它绑定的Queue中。Queue，并最终被两个消费者（C1与C2）消费。routingkeyQueue以上图的配置为例，我们以routingKey=”error”发送消息到Exchange，则消息会路由到Queue1（amqp.gen-S9b…，这是由RabbitMQ自动生成的Queue名称）和Queue2”或Queue2routingKey发送消息，则消息不会路由到这两个Queue中。前面讲到direct类型的Exchange路由规则是完全匹配bindingkey与routing类型的key与routingkey相匹配的Queue中，但这里的匹配规则有些不同，它约定：routingkey为一个句点号“.”分隔的字符串（被句点号“.”分隔开的每一段独立的字符串称为一个单词），如“stock.usd.nyse”、“nyse.vmw”、bindingkeyroutingkey一样也是句点号“以上图中的配置为例，routingKey=”quick.orange.rabbitQ1Q2routingKeylazy.orange.fox”的消息会路由到Q1，routingKey”Q2（只会投递给Q2一次，虽然这个routingKey与Q2的两个bindingKey都匹配；routingKey=”quick.brown.fox”、routingKey=”orange”、routingKey”quick.orange.male.rabbit”的消息将会被丢弃，因为它们没有匹配任何bindingKey。headersExchangeroutingkey与bindingkey的匹配规则来路由消息，而是根据发送的消息内容中的headers属性进行匹配。header（到该Queue。该类型的MQ本身是基于异步的消息处理，前面的示例中所有的生产者（P）将消息发送到在RabbitMQ中也支持RPC。定义了14种properties，这些属性会随着消息一起发送）中设置两个值replyTo（一个Queue名称，用于告诉服务器处理完成后将通知我的消息发送到这个Queue客户端将根据这个id了解哪条请求被成功执行了或执行失败）correlationId属性的correlationId属性分析哪条请求被执行了，根据执行结果进行后续业务处理案例 或者app应用，这种功能在大部分项目开发中都比较常见，而对于oworld的应用中虽然能够对消息进行消费，但是有很大问题:消带来很大压力，导致消息堆积越来越多。对于这种情况，RabbitMQ提供了工作队列模式，通过工作队列提供做个消费者，对MQ产生的消息进行消费，提高MQ消息的吞吐率，降低packagepackageimportimportimportimportimportpublicclassSendprivatefinalstaticStringQUEUE_NAME=publicstaticvoidmain(String[]args)ConnectionFactoryfactory=newConnectionFactory();Connectionconnection=null;Channelchannel=null;tryconnection=channel= channel.queueDeclare(QUEUE_NAME,false,false,false,for(inti=1;i<=20;i++)Stringmessage= oWorld!-----"+channel.basicPublish("",QUEUE_NAME,null,message.getBytes("UTF-8"));System.out.println("[x]Sent'"+message+"'");}}catch(IOException{}catch(TimeoutException{}finallytryif(null!=channel&&channel.isOpen())if(null!=connection&&connection.isOpen())}catch(TimeoutException{}catch(IOExceptione)}}}}packagepackageimportimportimportimportimportimportpublicclassRecv01privatefinalstaticStringQUEUE_NAME=publicstaticvoidmain(String[]args)ConnectionFactoryfactory=newConnectionFactory();tryConnectionconnection=Channelchannel=channel.queueDeclare(QUEUE_NAME,false,false,false,System.System.out.println("[*]Waitingformessages.ToexitpressDeliverCallbackdeliverCallback=(consumerTag,delivery)-{Stringmessage=newString(delivery.getBody(),"UTF-8");System.out.println("[x]Received01'"+message+"'");try}catch(InterruptedException{} channel.basicConsume(QUEUE_NAME,true,deliverCallback,consumerTag->}catch(IOException{}catch(TimeoutException{}}}packagepackageimportimportimportimportpublicclassRecv02privatefinalstaticStringQUEUE_NAME=publicstaticvoidmain(String[]args)ConnectionFactoryfactory=newConnectionFactory();tryConnectionconnection=Channelchannel=channel.queueDeclare(QUEUE_NAME,false,false,false,null);System.out.println("[*]Waitingformessages.ToexitpressCTRL+C");DeliverCallbackdeliverCallback=(consumerTag,delivery)-{Stringmessage=newString(delivery.getBody(),"UTF-8");System.out.println("[x]Received02'"+message+"'");try}catch(InterruptedException{} channel.basicConsume(QUEUE_NAME,true,deliverCallback,consumerTag->}catch(IOException{}catch(TimeoutException{}}}率，创建多个消费者来对消息进行处理。这里RabitMQ采用轮询来对消息进行分发时保证了消息被平均分配到每个消费方，但是引入新的问题:真正的生产环境下，对于消息的处理原因，机器cpu,内存等硬件问题，消费方处理消息时同类消息不同机器进行处理时消耗时111秒，215秒，121号消费者多收到消息进行处理，也即是我们通常所说的”能者多劳”,同样Rabbitmq对于这种消息分配模式的将第n条消息发给第n个消费者。packagepackageimportimportimportimportimportpublicclassSendprivatefinalstaticStringQUEUE_NAME=publicstaticvoidmain(String[]args)ConnectionFactoryfactory=newConnectionFactory();Connectionconnection=Channelchannel=null;try{connection=channel=channel.queueDeclare(QUEUE_NAME,false,false,false,for(inti=1;i<=20;i++)Stringmessage= oWorld!-----"+channel.basicPublish("",QUEUE_NAME,null,message.getBytes("UTF-8"));System.out.println("[x]Sent'"+message+"'");}}catch(IOException{}catch(TimeoutException{}finallytryif(null!=channel&&channel.isOpen())if(null!=connection&&connection.isOpen())}catch(TimeoutException{}catch(IOException{}}}}packagepackageimportimportimportimportpublicclassRecv01privatefinalstaticStringQUEUE_NAME=publicstaticvoidmain(String[]args)ConnectionFactoryfactory=newConnectionFactory();tryfinalConnectionconnection=finalChannelchannel=channel.queueDeclare(QUEUE_NAME,false,false,false,intprefetchCount=1;System.out.println("[*]Waitingformessages.ToexitpressDeliverCallbackdeliverCallback=(consumerTag,delivery)-{Stringmessage=newString(delivery.getBody(),"UTF-8");System.out.println("[x]Received01'"+message+"'");try}catch(InterruptedException{}} booleanautoAck=channel.basicConsume(QUEUE_NAME,autoAck,deliverCallback,consumerTag->}catch(IOException{}catch(TimeoutException{}}}packagepackageimportimportimportimportpublicclassRecv02privatefinalstaticStringQUEUE_NAME=publicstaticvoidmain(String[]args)ConnectionFactoryfactory=newtryfinalConnectionconnection=finalChannelchannel=channel.queueDeclare(QUEUE_NAME,false,false,false,intprefetchCount=1;System.out.println("[*]Waitingformessages.ToexitpressDeliverCallbackdeliverCallback=(consumerTag,delivery)-{Stringmessage=newString(delivery.getBody(),"UTF-8");System.out.println("[x]Received02'"+message+"'");try}catch(InterruptedException{} booleanautoAck=channel.basicConsume(QUEUE_NAME,autoAck,deliverCallback,consumerTag->}catch(IOException{}}catch(TimeoutException{}}}12fair机制被公对于，相信每个人都订阅过，当发送新的消息后，对于订阅过该公众见的如用户操作。模型图如下:息经过Exchange交换机投递到相关队列packagepackageimportimportimportimportpublicclassSend //privatefinalstaticStringQUEUE_NAME=privatestaticfinalStringEXCHANGE_NAME=publicstaticvoidmain(String[]args)ConnectionFactoryfactory=newConnectionFactory();Connectionconnection=null;Channelchannel=null;tryconnection=channel= //channel.queueDeclare(QUEUE_NAME,false,false,false, Stringmessage= channel.basicPublish(EXCHANGE_NAME,"",null,message.getBytes("UTF-8"));System.out.println("[x]Sent'"+message+"'");}catch(IOException{}catch(TimeoutException{}finallytryif(null!=channel&&channel.isOpen())if(null!=connection&&connection.isOpen())}}catch(TimeoutException{}catch(IOException{}}}}packagepackageimportimportimportpublicclassRecv01privatestaticfinalStringEXCHANGE_NAME=publicstaticvoidmain(String[]args)ConnectionFactoryfactory=newConnectionFactory();tryfinalConnectionconnection=finalChannelchannel=StringqueueName=channel.queueBind(queueName,EXCHANGE_NAME,intprefetchCount=1;System.out.println("[*]Waitingformessages.Toexitpress DeliverCallbackdeliverCallback=(consumerTag,delivery)-{Stringmessage=newString(delivery.getBody(),"UTF-8");System.out.println("[x]Received01'"+message+"'"); booleanautoAck=channel.basicConsume(queueName,autoAck,deliverCallback,consumerTag->}catch(IOException{}catch(TimeoutException{}}}packagepackageimportimportimportpublicclassRecv02privatestaticfinalStringEXCHANGE_NAME=publicstaticvoidmain(String[]args)ConnectionFactoryfactory=newConnectionFactory();tryfinalConnectionconnection=finalChannelchannel=StringqueueName=channel.queueBind(queueName,EXCHANGE_NAME,intprefetchCount=1;System.out.println("[*]Waitingformessages.Toexitpress DeliverCallbackdeliverCallback=(consumerTag,delivery)-{Stringmessage=newString(delivery.getBody(),"UTF-8");System.out.println("[x]Received01'"+message+"'");booleanautoAck=channel.basicConsume(queueName,autoAck,deliverCallback,consumerTag->}catch(IOException{}catch(TimeoutException{}}}解决：采用direct路由模式。通过案例时会遇到这种情况:在对项目信息输出日志进行收集时，会把日志(errorwarning,info)ExchangeTypes中的direct类型就可以实现，针对不同的消息，在对消息进行消费时，通过Exchangetypes以及Routingkey设置的规则，便可RoutingKey被路由到指定的队列进行后续消费packagepackageimportimportimportimportimportimportpublicclassSend privatestaticfinalStringEXCHANGE_NAME=publicstaticvoidmain(String[]args)ConnectionFactoryfactory=newConnectionFactory();Connectionconnection=null;Channelchannel=null;tryconnection=channel=Stringmessage"INFO消息//Stringmessage"WARNING消息//Stringmessage"ERROR消息StringroutingKey=//StringroutingKey=channel.basicPublish(EXCHANGE_NAME,routingKey,null,System.out.println("[x]Sent'"+message+}catch(IOException{}catch(TimeoutException{}finallytryif(null!=channel&&channel.isOpen())if(null!=connection&&connection.isOpen())}}catch(TimeoutException{}catch(IOException{}}}}消费者01routingKey=info和warning，对应级别日志消息均会路由到该队列中。packagepackageimportimportimportpublicclassRecv01privatestaticfinalStringEXCHANGE_NAME=publicstaticvoidmain(String[]args)ConnectionFactoryfactory=newConnectionFactory();tryfinalConnectionconnection=finalChannelchannel=StringqueueName=StringroutingKeyInfo="info";StringroutingKeyWarning="warning";channel.queueBind(queueName,EXCHANGE_NAME,routingKeyInfo);channel.queueBind(queueName,EXCHANGE_NAME,routingKeyWarning);intprefetchCount=1;System.out.println("[*]Waitingformessages.Toexitpress DeliverCallbackdeliverCallback=(consumerTag,delivery)-{Stringmessage=newString(delivery.getBody(),"UTF-8");System.out.println("[x]Received01'"+message+"'"); booleanautoAck=channel.basicConsume(queueName,autoAck,deliverCallback,consumerTag->}catch(IOException{}catch(TimeoutException{}}}packageimport importimportpublicclassRecv02privatestaticfinalStringEXCHANGE_NAME=publicstaticvoidmain(String[]args)ConnectionFactoryfactory=newConnectionFactory();tryfinalConnectionconnection=finalChannelchannel=StringqueueName=StringroutingKey=channel.queueBind(queueName,EXCHANGE_NAME,intprefetchCount=1;System.System.out.println("[*]Waitingformessages.ToexitpressDeliverCallbackdeliverCallback=(consumerTag,delivery)-{Stringmessage=newString(delivery.getBody(),"UTF-8");System.out.println("[x]Received02'"+message+"'"); booleanautoAck=channel.basicConsume(queueName,autoAck,deliverCallback,consumerTag->}catch(IOException{}catch(TimeoutException{}}}解决：采用topic模式。息路由到指定的队列，也符合在工作中的场景去使用的式，对于RabbitMq除了direct模式外，Mq同样还提供了topics模式来对消息进行匹配路由，比如在项目处理，此时采用direct模式可以实现，但对于的队列可能就不太容易进行，如果涉及模块很多，此时对应队列数量就很多，此时我们就可以通过topic模式来对消息routingkey匹配模式定义规则举例如下routingkey为一个句点号“.”分隔的字符串（被句点号“.”分隔开的每一段独立的字符串称为一个单词），如“stock.usd.nyse”、“nyse.vmw”、routingKey=”lazy.orange.fox”的消息会路由到Q1，Q2,routingKey=”lazy.brown.fox”的消息会路由到Q2，routingKey=”lazy.pink.rabbit”的消息会路由到Q2；任何bindingKey。packagepackageimportimportimportimport* publicclassSend privatestaticfinalStringEXCHANGE_NAME=publicstaticvoidmain(String[]args)ConnectionFactoryfactory=newConnectionFactory();Connectionconnection=null;Channelchannel=null;tryconnection=channel= Stringmessage"商品查询操作StringroutingKey=//StringroutingKey=channel.basicPublish(EXCHANGE_NAME,routingKey,null,System.out.println("[x]Sent'"+message+}catch(IOException{}catch(TimeoutException{}}finallytryif(null!=channel&&channel.isOpen())if(null!=connection&&connection.isOpen())}catch(TimeoutException{}catch(IOException{}}}}packagepackageimportimportimport* publicclassRecv01privatestaticfinalStringEXCHANGE_NAME=publicstaticvoidmain(String[]args)ConnectionFactoryfactory=newConnectionFactory();tryfinalConnectionconnection=finalChannelchannel= StringqueueName=StringroutingKey=channel.queueBind(queueName,EXCHANGE_NAME,intprefetchCount=1;System.out.println("[*]Waitingformessages.Toexitpress DeliverCallbackdeliverCallback=(consumerTag,delivery)-{Stringmessage=newString(delivery.getBody(),"UTF-8");System.out.println("[x]Received01'"+message+"'"); booleanautoAck=channel.basicConsume(queueName,autoAck,deliverCallback,consumerTag->}catch(IOException{}catch(TimeoutException{}}}}packageimport importimport* publicclassRecv02privatestaticfinalStringEXCHANGE_NAME=publicstaticvoidmain(String[]args)ConnectionFactoryfactory=newConnectionFactory();tryfinalConnectionconnection=finalChannelchannel= StringqueueName=StringroutingKey=channel.queueBind(queueName,EXCHANGE_NAME,intintprefetchCount=1;System.out.println("[*]Waitingformessages.ToexitpressDeliverCallbackdeliverCallback=(consumerTag,delivery)-{Stringmessage=newString(delivery.getBody(),"UTF-8");System.out.println("[x]Received02'"+message+"'"); booleanautoAck=channel.basicConsume(queueName,autoAck,deliverCallback,consumerTag->}catch(IOException{}catch(TimeoutException{}}}从结果可以看出生产者发送了多条设置了路由匹配规则()的消息，根据不同的路由匹配规则()，可以将消息根据指定的routingkey路由到匹配到的队列中，也是在解决：采用RPC模式。MQ本身是基于异步的消息处理，前面的示例中所有的生产者（P）将消息发送到在RabbitMQ中也支持RPC。客户端发送请求（消息）时，在消息的属性（MessageProperties，在AMQP协议中定义了14种properties，这些属性会随着消息一起发送）中设置两个值replyTo（一个Queue名称，用于告诉服务器处理完成后将通知我的消息发送到将此属性返还，客户端将根据这个id了解哪条请求被成功执行了或执行失败）带correlationId属性的correlationId属性分析哪条请求被执行了，根据执行结果进行后续业务处理。packagepackageimportimportimportpublicclassRPCServerprivatestaticfinalStringRPC_QUEUE_NAME= *@paramprivatestaticintfib(intn)if(n==0)returnif(n==1)returnreturnfib(n-1)+fib(n-}publicstaticvoidmain(String[]args)ConnectionFactoryfactory=newConnectionFactory();tryfinalConnectionconnection=finalChannelchannel= channel.queueDeclare(RPC_QUEUE_NAME,false,false,false,null);intprefetchCount=1;System.out.println("[x]AwaitingRPCrequests");Objectmonitor=newObject();DeliverCallbackdeliverCallback=(consumerTag,delivery)->AMQP.BasicPropertiesreplyProps=newStringresponse="";try{Stringmessage=newString(delivery.getBody(),"UTF-8");intn=Integer.parseInt(message);System.out.println("[.]fib("+message+response+=}catch(RuntimeException{System.out.println("[.]"+}finallychannel.basicPublish("",delivery.getProperties().getReplyTo(),replyProps,//RabbitMqconsumerworkerthreadnotifiestheRPCserverownersynchronized{}} booleanautoAck=channel.basicConsume(RPC_QUEUE_NAME,autoAck,deliverCallback,consumerTag->//Waitandbepreparedtoconsumethemessagefrom while(true)synchronized(monitor)try}catch(InterruptedException{}}}}}catch(IOException{}catch(TimeoutException{}}}packagepackage;importimportimportimportimportimportimportpublicclass implementsAutoCloseableprivateConnectionprivateChannelprivateStringrequestQueueName=public()throwsIOException,{ConnectionFactoryfactory=newconnection=channel=}publicstaticvoidmain(String[]args)try fibonacciRpc=new ())for(inti=0;i<10;i++)Stringi_str=System.out.println("[x]Requestingfib("+i_str+Stringresponse=fibonacciRpc.call(i_str);System.out.println("[.]Got'"+response+"'");}}catch(IOException|TimeoutException|InterruptedException{}}publicStringcall(Stringmessage)throwsIOException,InterruptedExceptionfinalStringcorrId=StringreplyQueueName=AMQP.BasicPropertiesprops=newfinalBlockingQueue<String>response=newStringctag=channel.basicConsume(replyQueueName,true,(consumerTag,delivery)->responseresponse.offer(newString(delivery.getBody(),"UTF-}},consumerTag->Stringresult=//从消息队列中丢弃该值returnresult;}publicvoidclose()throwsIOException}}底有没有正确到达broker服务器呢？如果不进行特殊配置的话，默认情况下发布操作brokerbroker之前已经丢失的话，持久化操作也解决不了这个问RabbitMQ中与事务机制有关的方法有三个：txSelect(),mit()以及txRollback(),txSelect用于将当前channel设置成transaction模式，mit()用于提交事务，txRollback()用于回滚事务，在通过txSelect()开启事务之后，我们便可以发布消息给broker服务器了，如果 broker了，如果在 个时候我们便可以捕获异常通过txRollback()回滚事务。trytryconnection=channel=channel.queueDeclare(QUEUE_NAME,false,false,false,Stringmessage= ochannel.basicPublish("",QUEUE_NAME,null,message.getBytes("UTF-8"));System.out.println("[x]Sent'"+message+"'");inti=1/}catch(IOException|TimeoutException{try}catch(IOException{}}与使用事务机制的话会降低RabbitMQ的性能，那么有没有更好的方法既能保障producer知道消息已经正确送到，又能基本上不带来性能上的损失呢？从AMQP协议的层面看是没有更好的方法，但是RabbitMQ提供了一个更好的方案，即将channel信道设置成confirm模式。有第二种方式即:Confirm模式。的消息都会被指派一个唯一的ID(从1开始)，一旦消息被投递到所有匹配的队列之后，broker就会发送一个确认给生产者（ID）,这就使得生产者知道消息已经发送一条nack消息，生产者应用程序同样可以在回调方法中处理该nack消息。在channel被设置成confirm模式之后，所有被publish的后续消息都将被confirm（即ack）或者被nack一次。但是没有对消息被confirm的快慢做任何保证，并且同一条消息