RabbitMQ是一个开源的消息队列系统，由Erlang语言开发，基于AMQP协议实现。AMQP的主要特点是面向消息、排队、路由(包括点对点和发布/订阅)、可靠性和安全性。AMQP协议更多用于企业系统，对数据的一致性、稳定性和可靠性要求高，对性能和吞吐量的要求次之。

资料来源：https://www.cnblogs.com/haixiang/p/10199754.html.

RabbitMQ简述

RabbitMQ是一个消息代理：它接受和转发消息。你可以把它想象成一个邮局：当你准备把邮件投递到邮箱的时候，你可以确定邮差先生最终会把邮件送到收件人手中。在这个比喻中，RabbitMQ是邮箱、邮局和邮递员，用于接收、存储和转发二进制数据块的消息。

队列就像是在RabbitMQ中扮演邮箱的角色。虽然消息通过RabbitMQ和应用程序，但它们只能存储在队列中。队列只受主机的内存和磁盘限制，本质上是一个很大的消息缓冲区。许多生产者可以向队列发送消息，许多消费者可以尝试从队列接收数据。

Producer是生产者，用于生成消息并将其发送到队列。消费者是消费者，需要读取队列中的消息。生产者、消费者和代理(rabbitMQserver)不必驻留在同一个主机上；事实上，在大多数应用程序中，它们都是这样分布的。

资料来源：https://www.cnblogs.com/haixiang/p/10826710.html.

RabbitMQ支持哪些协议？

RabbitMQ通过使用插件支持多种消息传递协议。

AMQP0-9-1

跺脚

MQTT

AMQP1.0

HTTPandWebSockets

资料来源：https://www.cnblogs.com/haixiang/p/10853467.html.

与其他消息队列相比，RabbitMQ有一系列防止消息丢失的措施，具有很强的高可用性能。它的吞吐量可能不如其他消息队列大，但是它的消息可支持性非常突出，在金融服务中被广泛使用。与其他消息队列的比较以及防止消息丢失的强大功能将在后续文章中介绍。

RabbitMQ核心概念

工作队列(也称为任务队列)的主要思想是避免立即执行资源密集型任务，而必须等待它完成。相反，我们把任务安排在以后完成。我们将任务封装为消息，并将其发送到队列。后台运行的worker进程会弹出任务，最终执行作业。当您运行许多工作线程时，任务将在它们之间共享。

这个概念在Web应用程序中特别有用，因为在Web应用程序中，不可能在很短的HTTP请求窗口内处理复杂的任务。

默认情况下，RabbitMQ依次将每条消息发送给下一个消费者。平均而言，每个消费者都会收到相同数量的消息。这种分发消息的方式称为循环。让三个或更多的工人试一试。

队列

路由

频道。queue _ bind(exchange=exchange _ name，queue=queue_name)绑定是交换和队列之间的关系。

队列对来自此交换的邮件感兴趣。

绑定可以使用附加的routing_key参数。为了避免与basic_publish参数混淆，我们称它为绑定键。这就是我们如何用键创建绑定：

channel . queue _ bind(exchange=exchange _ name，queue=queue_name，routing_key='black ')绑定密钥的含义取决于交换类型。我们之前使用的扇出faudout开关只是忽略了它的值。

绑定

我们采用扇出切换，并没有给我们带来太大的灵活性——只能无意识的播放。

我们将使用直接兑换。直接交换背后的路由算法很简单——邮件进入绑定键与邮件的路由键完全匹配的队列。

为了说明这一点，请考虑以下设置：

在这个设置中，我们可以看到绑定了两个队列的直接交换X。第一个队列由绑定键orange绑定，第二个队列有两个绑定，一个是黑色，另一个是绿色。

在此设置中，发送到带有橙色路由关键字的交换机的消息将被路由到队列Q1。带有黑色或绿色路由关键字的消息将被发送到Q2。所有其他消息都将被丢弃。

直接交换

用同一个绑定键绑定多个队列是完全合法的。在我们的例子中，我们可以在

X和Q1之间添加绑定。在这种情况下，直接交换的行为将类似于扇出，并将消息广播到所有匹配的队列。带有黑色路由键的消息将同时传递给Q1和Q2

消息确认

执行任务可能需要几秒钟。您可能想知道，如果其中一个使用者开始一项漫长的任务并仅部分完成而死掉，会发生什么。使用我们当前的代码，RabbitMQ一旦将消息传递给使用者，便立即将其标记为删除。在这种情况下，如果您杀死一个工人，我们将丢失正在处理的消息。我们还将丢失所有发送给该特定工作人员但尚未处理的消息。

但是我们不想丢失任何任务。如果一个工人死亡，我们希望将任务交付给另一个工人。

为了确保消息永不丢失，RabbitMQ支持消息确认。消费者发送回一个确认（acknowledgement），以告知RabbitMQ已经接收，处理了特定的消息，并且RabbitMQ可以自由删除它。

如果使用者死了（其通道已关闭，连接已关闭或TCP连接丢失）而没有发送确认，RabbitMQ将了解消息未完全处理，并将重新排队。如果同时有其他消费者在线，它将很快将其重新分发给另一个消费者。这样，您可以确保即使工人偶尔死亡也不会丢失任何消息。

没有任何消息超时；消费者死亡时，RabbitMQ将重新传递消息。即使处理一条消息花费非常非常长的时间也没关系。

默认情况下，手动消息确认处于打开状态。在前面的示例中，我们通过auto_ack=True标志显式关闭了它们。完成任务后，是时候删除此标志并从工作人员发送适当的确认了。

确认必须在收到交货的同一通道上发送。

尝试使用其他通道进行确认将导致通道级协议异常。

讯息持久性

我们已经学会了如何确保即使消费者死亡，任务也不会丢失。但是，如果RabbitMQ服务器停止，我们的任务仍然会丢失。

RabbitMQ退出或崩溃时，它将忘记队列和消息，除非您告知不要这样做。要确保消息不会丢失，需要做两件事：我们需要将队列和消息都标记为持久。

公平派遣

您可能已经注意到，调度仍然无法完全按照我们的要求进行。例如，在有两名工人的情况下，当所有奇怪的消息都很重，甚至消息很轻时，一位工人将一直忙碌而另一位工人将几乎不做任何工作。好吧，RabbitMQ对此一无所知，并且仍将平均分配消息。

发生这种情况是因为RabbitMQ在消息进入队列时才调度消息。它不会查看使用者的未确认消息数。它只是盲目地将每第n条消息发送给第n个使用者。

为了解决这个问题，我们可以将Channel＃basic_qos通道方法与prefetch_count=1设置一起使用。这使用basic.qos协议方法来告诉RabbitMQ一次不向工作人员发送多条消息。换句话说，在处理并确认上一条消息之前，不要将新消息发送给工作人员。而是将其分派给不忙的下一个工作程序。

有关消息持久性的说明

将消息标记为持久性并不能完全保证不会丢失消息。尽管它告诉RabbitMQ将消息保存到磁盘，但是RabbitMQ接受消息并且尚未保存消息时，还有很短的时间。而且，RabbitMQ不会对每条消息都执行fsync（2）－它可能只是保存到缓存中，而没有真正写入磁盘。持久性保证并不强，但是对于我们的简单任务队列而言，这已经绰绰有余了。如果您需要更强有力的保证，则可以使用发布者确认。

AMQP协议

AMQP:AdvancedMessageQueuingProtocol高级消息队列协议

AMQP定义:是具有现代特征的二进制协议。是一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。

Erlang语言最初在于交换机领域的架构模式，这样使得RabbitMQ在Broker之间进行数据交互的性能是非常优秀的Erlang的优点:Erlang有着和原生Socket一样的延迟。

RabbitMQ是一个开源的消息代理和队列服务器，用来通过普通协议在完全不同的应用之间共享数据,RabbitMQ是使用Erlang语言来编写的，并且RabbitMQ是基于AMQP协议的。

RabbitMQ消息传递机制

生产者发送消息到指定的Exchange，Exchange依据自身的类型(direct、topic等)，根据routingkey将消息发送给0-n个队列，队列再将消息转发给了消费者。

Server:又称Broker,接受客户端的连接，实现AMQP实体服务，这里指RabbitMQ服务器

Connection:连接，应用程序与Broker的网络连接。

Channel:网络信道，几乎所有的操作都在Channel中进行，Channel是进行消息读写的通道。客户端可建立多个Channel:，每个Channel代表一个会话任务。

Virtualhost:虚似地址，用于迸行逻辑隔离，是最上层的消息路由。一个VirtualHost里面可以有若干个Exchange和Queue，同一个VirtualHost里面不能有相同名称的Exchange或Queue。权限控制的最小粒度是VirtualHost。

Binding:Exchange和Queue之间的虚拟连接，binding中可以包含routingkey。

Routingkey:一个路由规则，虚拟机可用它来确定如何路由一个特定消息，即交换机绑定到Queue的键。

Queue:也称为MessageQueue,消息队列，保存消息并将它们转发给消费者。

Message

消息，服务器和应用程序之间传送的数据，由Properties和Body组成。Properties可以对消息进行修饰，比如消息的优先级、延迟等高级特性;，Body则就是消息体内容。

properties中我们可以设置消息过期时间以及是否持久化等，也可以传入自定义的map属性，这些在消费端也都可以获取到。

生产者

importcom.rabbitmq.client.AMQP;importcom.rabbitmq.client.Channel;importcom.rabbitmq.client.Connection;importcom.rabbitmq.client.ConnectionFactory;importjava.util.HashMap;importjava.util.Map;publicclassMessageProducer{publicstaticvoidmain(String[]args)throwsException{//1.创建一个ConnectionFactory并进行设置ConnectionFactoryfactory=newConnectionFactory();factory.setHost("localhost");factory.setVirtualHost("/");factory.setUsername("guest");factory.setPassword("guest");//2.通过连接工厂来创建连接Connectionconnection=factory.newConnection();//3.通过Connection来创建ChannelChannelchannel=connection.createChannel();//4.声明使用默认交换机以队列名作为routingkeyStringqueueName="msg_queue";/***deliverMode设置为2的时候代表持久化消息*expiration意思是设置消息的有效期，超过10秒没有被消费者接收后会被自动删除*headers自定义的一些属性**///5.发送Map<String,Object>headers=newHashMap<String,Object>();headers.put("myhead1","111");headers.put("myhead2","222");AMQP.BasicPropertiesproperties=newAMQP.BasicProperties().builder().deliveryMode(2).contentEncoding("UTF-8").expiration("100000").headers(headers).build();Stringmsg="testmessage";channel.basicPublish("",queueName,properties,msg.getBytes());System.out.println("Sendmessage:"+msg);//6.关闭连接channel.close();connection.close();}}

消费者

importcom.rabbitmq.client.*;importjava.io.IOException;importjava.util.Map;publicclassMessageConsumer{publicstaticvoidmain(String[]args)throwsException{//1.创建一个ConnectionFactory并进行设置ConnectionFactoryfactory=newConnectionFactory();factory.setHost("localhost");factory.setVirtualHost("/");factory.setUsername("guest");factory.setPassword("guest");factory.setAutomaticRecoveryEnabled(true);factory.setNetworkRecoveryInterval(3000);//2.通过连接工厂来创建连接Connectionconnection=factory.newConnection();//3.通过Connection来创建ChannelChannelchannel=connection.createChannel();//4.声明StringqueueName="msg_queue";channel.queueDeclare(queueName,false,false,false,null);//5.创建消费者并接收消息Consumerconsumer=newDefaultConsumer(channel){@OverridepublicvoidhandleDelivery(StringconsumerTag,Envelopeenvelope,AMQP.BasicPropertiesproperties,byte[]body)throwsIOException{Stringmessage=newString(body,"UTF-8");Map<String,Object>headers=properties.getHeaders();System.out.println("head:"+headers.get("myhead1"));System.out.println("[x]Received'"+message+"'");System.out.println("expiration:"+properties.getExpiration());}};//6.设置Channel消费者绑定队列channel.basicConsume(queueName,true,consumer);}}Sendmessage:testmessagehead:111[x]Received'testmessage'100000

Exchange

1.简介

Exchange就是交换机，接收消息，根据路由键转发消息到绑定的队列。有很多的Message进入到Exchange中，Exchange根据Routingkey将Message分发到不同的Queue中。

2.类型

RabbitMQ中的Exchange有多种类型，类型不同，Message的分发机制不同，如下：

fanout：广播模式。这种类型的Exchange会将Message分发到绑定到该Exchange的所有Queue。direct：这种类型的Exchange会根据Routingkey（精确匹配，将Message分发到指定的Queue。Topic：这种类型的Exchange会根据Routingkey（模糊匹配，将Message分发到指定的Queue。headers:主题交换机有点相似，但是不同于主题交换机的路由是基于路由键，头交换机的路由值基于消息的header数据。主题交换机路由键只有是字符串,而头交换机可以是整型和哈希值.

3.属性

/***Declareanexchange,viaaninterfacethatallowsthecompletesetof*arguments.*@seecom.rabbitmq.client.AMQP.Exchange.Declare*@seecom.rabbitmq.client.AMQP.Exchange.DeclareOk*@paramexchangethenameoftheexchange交换机名称*@paramtypetheexchangetype交换机类型direct、topic、fanout、headers*@paramdurabletrueifwearedeclaringadurableexchange(theexchangewillsurviveaserverrestart)是否需要持久化，true为持久化*@paramautoDeletetrueiftheservershoulddeletetheexchangewhenitisnolongerinuse当最后一个绑定到Exchange.上的队列删除后，自动删除该Exchange*@paraminternaltrueiftheexchangeisinternal,i.e.can'tbedirectly当前Exchange是否用于RabbitMQ内部使用，默认为False*publishedtobyaclient.*@paramargumentsotherproperties(constructionarguments)fortheexchange扩展参数，用于扩展AMQP协议自制定化使用*@returnadeclaration-confirmmethodtoindicatetheexchangewassuccessfullydeclared*@throwsjava.io.IOExceptionifanerrorisencountered*/Exchange.DeclareOkexchangeDeclare(Stringexchange,Stringtype,booleandurable,booleanautoDelete,booleaninternal,Map<String,Object>arguments)throwsIOException;

RabbitMQ模式

简单队列模式

简单队列是最简单的一种模式，由生产者、队列、消费者组成。生产者将消息发送给队列，消费者从队列中读取消息完成消费。

在下图中，“P”是我们的生产者，“C”是我们的消费者。中间的框是队列-RabbitMQ代表消费者的消息缓冲区。

java方式

生产者

packagecom.anqi.mq.nat;importcom.rabbitmq.client.Channel;importcom.rabbitmq.client.Connection;importcom.rabbitmq.client.ConnectionFactory;publicclassMyProducer{privatestaticfinalStringQUEUE_NAME="ITEM_QUEUE";publicstaticvoidmain(String[]args)throwsException{//1.创建一个ConnectionFactory并进行设置ConnectionFactoryfactory=newConnectionFactory();factory.setHost("localhost");factory.setVirtualHost("/");factory.setUsername("guest");factory.setPassword("guest");//2.通过连接工厂来创建连接Connectionconnection=factory.newConnection();//3.通过Connection来创建ChannelChannelchannel=connection.createChannel();//实际场景中，消息多为json格式的对象Stringmsg="hello";//4.发送三条数据for(inti=1;i<=3;i++){channel.basicPublish("",QUEUE_NAME,null,msg.getBytes());System.out.println("Sendmessage"+i+":"+msg);}//5.关闭连接channel.close();connection.close();}}

消费者

packagecom.anqi.mq.nat;importcom.rabbitmq.client.*;importjava.io.IOException;publicclassMyConsumer{privatestaticfinalStringQUEUE_NAME="ITEM_QUEUE";publicstaticvoidmain(String[]args)throwsException{//1.创建一个ConnectionFactory并进行设置ConnectionFactoryfactory=newConnectionFactory();factory.setHost("localhost");factory.setVirtualHost("/");factory.setUsername("guest");factory.setPassword("guest");//2.通过连接工厂来创建连接Connectionconnection=factory.newConnection();//3.通过Connection来创建ChannelChannelchannel=connection.createChannel();//4.声明一个队列channel.queueDeclare(QUEUE_NAME,true,false,false,null);System.out.println("[*]Waitingformessages.ToexitpressCTRL+C");/*true:表示自动确认，只要消息从队列中获取，无论消费者获取到消息后是否成功消费，都会认为消息已经成功消费false:表示手动确认，消费者获取消息后，服务器会将该消息标记为不可用状态，等待消费者的反馈，如果消费者一直没有反馈，那么该消息将一直处于不可用状态，并且服务器会认为该消费者已经挂掉，不会再给其发送消息，直到该消费者反馈。*///5.创建消费者并接收消息Consumerconsumer=newDefaultConsumer(channel){@OverridepublicvoidhandleDelivery(StringconsumerTag,Envelopeenvelope,AMQP.BasicPropertiesproperties,byte[]body)throwsIOException{Stringmessage=newString(body,"UTF-8");System.out.println("[x]Received'"+message+"'");}};//6.设置Channel消费者绑定队列channel.basicConsume(QUEUE_NAME,true,consumer);}}Sendmessage1:helloSendmessage2:helloSendmessage3:hello[*]Waitingformessages.ToexitpressCTRL+C[x]Received'hello'[x]Received'hello'[x]Received'hello'

当我们启动生产者之后查看RabbitMQ管理后台可以看到有一条消息正在等待被消费。

当我们启动生产者之后查看RabbitMQ管理后台可以看到有一条消息正在等待被消费。当我们启动消费者之后再次查看，可以看到积压的一条消息已经被消费。

总结

/***Declareaqueue*@paramqueuethenameofthequeue第一个参数表示队列名称*@paramdurabletrueifwearedeclaringadurablequeue(thequeuewillsurviveaserverrestart)第二个参数为是否持久化（true表示是，队列将在服务器重启时生存）*@paramexclusivetrueifwearedeclaringanexclusivequeue(restrictedtothisconnection)第三个参数为是否是独占队列（创建者可以使用的私有队列，断开后自动删除）*@paramautoDeletetrueifwearedeclaringanautodeletequeue(serverwilldeleteitwhennolongerinuse)第四个参数为当所有消费者客户端连接断开时是否自动删除队列*@paramargumentsotherproperties(constructionarguments)forthequeue第五个参数为队列的其他参数*@returnadeclaration-confirmmethodtoindicatethequeuewassuccessfullydeclared*@throwsjava.io.IOExceptionifanerrorisencountered*/Queue.DeclareOkqueueDeclare(Stringqueue,booleandurable,booleanexclusive,booleanautoDelete,Map<String,Object>arguments)throwsIOException;/***Publishamessage*@seecom.rabbitmq.client.AMQP.Basic.Publish*@paramexchangetheexchangetopublishthemessageto*@paramroutingKeytheroutingkey*@parampropsotherpropertiesforthemessage-routingheadersetc*@parambodythemessagebody*@throwsjava.io.IOExceptionifanerrorisencountered*/voidbasicPublish(Stringexchange,StringroutingKey,BasicPropertiesprops,byte[]body)throwsIOException;/***Startanon-nolocal,non-exclusiveconsumer,with*aserver-generatedconsumerTag.*@paramqueuethenameofthequeue*@paramautoAcktrueiftheservershouldconsidermessages*acknowledgedoncedelivered;falseiftheservershouldexpect*explicitacknowledgements第二个参数autoAck:应答模式，true：自动应答，即消费者获取到消息，该消息就会从队列中删除掉，false：手动应答，当从队列中取出消息后，需要程序员手动调用方法应答，如果没有应答，该消息还会再放进队列中，就会出现该消息一直没有被消费掉的现象。*@paramcallbackaninterfacetotheconsumerobject*@returntheconsumerTaggeneratedbytheserver*@throwsjava.io.IOExceptionifanerrorisencountered*@seecom.rabbitmq.client.AMQP.Basic.Consume*@seecom.rabbitmq.client.AMQP.Basic.ConsumeOk*@see#basicConsume(String,boolean,String,boolean,boolean,Map,Consumer)*/StringbasicConsume(Stringqueue,booleanautoAck,Consumercallback)throwsIOException;

这种简单队列的模式，系统会为每个队列隐式地绑定一个默认交换机，交换机名称为"(AMQPdefault)"，类型为直连direct，当你手动创建一个队列时，系统会自动将这个队列绑定到一个名称为空的Direct类型的交换机上，绑定的路由键routingkey与队列名称相同，相当于channel.queueBind(queue:"QUEUE_NAME",exchange:"(AMQPdefault)“,routingKey:"QUEUE_NAME");虽然实例没有显式声明交换机，但是当路由键和队列名称一样时，就会将消息发送到这个默认的交换机中。这种方式比较简单，但是无法满足复杂的业务需求，所以通常在生产环境中很少使用这种方式。Thedefaultexchangeisimplicitlyboundtoeveryqueue,witharoutingkeyequaltothequeuename.Itisnotpossibletoexplicitlybindto,orunbindfromthedefaultexchange.Italsocannotbedeleted.默认交换机隐式绑定到每个队列，其中路由键等于队列名称。不可能显式绑定到，或从缺省交换中解除绑定。它也不能被删除。——引自RabbitMQ官方文档

最常用的三大模式-Direct模式

所有发送到DirectExchange的消息被转发到RouteKey中指定的Queue。Direct模式可以使用RabbitMQ自带的Exchange:defaultExchange，所以不需要将Exchange进行任何绑定(binding)操作。消息传递时，RouteKey必须完全匹配才会被队列接收，否则该消息会被抛弃，

importcom.rabbitmq.client.Channel;importcom.rabbitmq.client.Connection;importcom.rabbitmq.client.ConnectionFactory;publicclassDirectProducer{publicstaticvoidmain(String[]args)throwsException{//1.创建一个ConnectionFactory并进行设置ConnectionFactoryfactory=newConnectionFactory();factory.setHost("localhost");factory.setVirtualHost("/");factory.setUsername("guest");factory.setPassword("guest");//2.通过连接工厂来创建连接Connectionconnection=factory.newConnection();//3.通过Connection来创建ChannelChannelchannel=connection.createChannel();//4.声明StringexchangeName="test_direct_exchange";StringroutingKey="item.direct";//5.发送Stringmsg="thisisdirectmsg";channel.basicPublish(exchangeName,routingKey,null,msg.getBytes());System.out.println("Sendmessage:"+msg);//6.关闭连接channel.close();connection.close();}}

importcom.rabbitmq.client.*;importjava.io.IOException;publicclassDirectConsumer{publicstaticvoidmain(String[]args)throwsException{//1.创建一个ConnectionFactory并进行设置ConnectionFactoryfactory=newConnectionFactory();factory.setHost("localhost");factory.setVirtualHost("/");factory.setUsername("guest");factory.setPassword("guest");factory.setAutomaticRecoveryEnabled(true);factory.setNetworkRecoveryInterval(3000);//2.通过连接工厂来创建连接Connectionconnection=factory.newConnection();//3.通过Connection来创建ChannelChannelchannel=connection.createChannel();//4.声明StringexchangeName="test_direct_exchange";StringqueueName="test_direct_queue";StringroutingKey="item.direct";channel.exchangeDeclare(exchangeName,"direct",true,false,null);channel.queueDeclare(queueName,false,false,false,null);//一般不用代码绑定，在管理界面手动绑定channel.queueBind(queueName,exchangeName,routingKey);//5.创建消费者并接收消息Consumerconsumer=newDefaultConsumer(channel){@OverridepublicvoidhandleDelivery(StringconsumerTag,Envelopeenvelope,AMQP.BasicPropertiesproperties,byte[]body)throwsIOException{Stringmessage=newString(body,"UTF-8");System.out.println("[x]Received'"+message+"'");}};//6.设置Channel消费者绑定队列channel.basicConsume(queueName,true,consumer);}}

最常用的三大模式-Topic模式

可以使用通配符进行模糊匹配

符号'#"匹配一个或多个词符号"*”匹配不多不少一个词

例如:

'log.#"能够匹配到'log.info.oa""log.*"只会匹配到"log.erro“

importcom.rabbitmq.client.Channel;importcom.rabbitmq.client.Connection;importcom.rabbitmq.client.ConnectionFactory;publicclassTopicProducer{publicstaticvoidmain(String[]args)throwsException{//1.创建一个ConnectionFactory并进行设置ConnectionFactoryfactory=newConnectionFactory();factory.setHost("localhost");factory.setVirtualHost("/");factory.setUsername("guest");factory.setPassword("guest");//2.通过连接工厂来创建连接Connectionconnection=factory.newConnection();//3.通过Connection来创建ChannelChannelchannel=connection.createChannel();//4.声明StringexchangeName="test_topic_exchange";StringroutingKey1="item.update";StringroutingKey2="item.delete";StringroutingKey3="user.add";//5.发送Stringmsg="thisistopicmsg";channel.basicPublish(exchangeName,routingKey1,null,msg.getBytes());channel.basicPublish(exchangeName,routingKey2,null,msg.getBytes());channel.basicPublish(exchangeName,routingKey3,null,msg.getBytes());System.out.println("Sendmessage:"+msg);//6.关闭连接channel.close();connection.close();}}

importcom.rabbitmq.client.*;importjava.io.IOException;publicclassTopicConsumer{publicstaticvoidmain(String[]args)throwsException{//1.创建一个ConnectionFactory并进行设置ConnectionFactoryfactory=newConnectionFactory();factory.setHost("localhost");factory.setVirtualHost("/");factory.setUsername("guest");factory.setPassword("guest");factory.setAutomaticRecoveryEnabled(true);factory.setNetworkRecoveryInterval(3000);//2.通过连接工厂来创建连接Connectionconnection=factory.newConnection();//3.通过Connection来创建ChannelChannelchannel=connection.createChannel();//4.声明StringexchangeName="test_topic_exchange";StringqueueName="test_topic_queue";StringroutingKey="item.#";channel.exchangeDeclare(exchangeName,"topic",true,false,null);channel.queueDeclare(queueName,false,false,false,null);//一般不用代码绑定，在管理界面手动绑定channel.queueBind(queueName,exchangeName,routingKey);//5.创建消费者并接收消息Consumerconsumer=newDefaultConsumer(channel){@OverridepublicvoidhandleDelivery(StringconsumerTag,Envelopeenvelope,AMQP.BasicPropertiesproperties,byte[]body)throwsIOException{Stringmessage=newString(body,"UTF-8");System.out.println("[x]Received'"+message+"'");}};//6.设置Channel消费者绑定队列channel.basicConsume(queueName,true,consumer);}}

最常用的三大模式-Fanout模式

不处理路由键，只需要简单的将队列绑定到交换机上发送到交换机的消息都会被转发到与该交换机绑定的所有队列上。Fanout交换机转发消息是最快的。

importcom.rabbitmq.client.*;importjava.io.IOException;publicclassFanoutConsumer{publicstaticvoidmain(String[]args)throwsException{//1.创建一个ConnectionFactory并进行设置ConnectionFactoryfactory=newConnectionFactory();factory.setHost("localhost");factory.setVirtualHost("/");factory.setUsername("guest");factory.setPassword("guest");factory.setAutomaticRecoveryEnabled(true);factory.setNetworkRecoveryInterval(3000);//2.通过连接工厂来创建连接Connectionconnection=factory.newConnection();//3.通过Connection来创建ChannelChannelchannel=connection.createChannel();//4.声明StringexchangeName="test_fanout_exchange";StringqueueName="test_fanout_queue";StringroutingKey="item.#";channel.exchangeDeclare(exchangeName,"fanout",true,false,null);channel.queueDeclare(queueName,false,false,false,null);//一般不用代码绑定，在管理界面手动绑定channel.queueBind(queueName,exchangeName,routingKey);//5.创建消费者并接收消息Consumerconsumer=newDefaultConsumer(channel){@OverridepublicvoidhandleDelivery(StringconsumerTag,Envelopeenvelope,AMQP.BasicPropertiesproperties,byte[]body)throwsIOException{Stringmessage=newString(body,"UTF-8");System.out.println("[x]Received'"+message+"'");}};//6.设置Channel消费者绑定队列channel.basicConsume(queueName,true,consumer);}}

importcom.rabbitmq.client.Channel;importcom.rabbitmq.client.Connection;importcom.rabbitmq.client.ConnectionFactory;publicclassFanoutProducer{publicstaticvoidmain(String[]args)throwsException{//1.创建一个ConnectionFactory并进行设置ConnectionFactoryfactory=newConnectionFactory();factory.setHost("localhost");factory.setVirtualHost("/");factory.setUsername("guest");factory.setPassword("guest");//2.通过连接工厂来创建连接Connectionconnection=factory.newConnection();//3.通过Connection来创建ChannelChannelchannel=connection.createChannel();//4.声明StringexchangeName="test_fanout_exchange";StringroutingKey1="item.update";StringroutingKey2="";StringroutingKey3="ookjkjjkhjhk";//任意routingkey//5.发送Stringmsg="thisisfanoutmsg";channel.basicPublish(exchangeName,routingKey1,null,msg.getBytes());channel.basicPublish(exchangeName,routingKey2,null,msg.getBytes());channel.basicPublish(exchangeName,routingKey3,null,msg.getBytes());System.out.println("Sendmessage:"+msg);//6.关闭连接channel.close();connection.close();}}

spring-amqp方式

引入Maven依赖

<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.6.0</version></dependency><dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit</artifactId><version>2.1.5.RELEASE</version></dependency>

spring配置文件

<beansxmlns="http://www.springframework.org/schema/beans"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xmlns:rabbit="http://www.springframework.org/schema/rabbit"xsi:schemaLocation="http://www.springframework.org/schema/rabbithttps://www.springframework.org/schema/rabbit/spring-rabbit.xsdhttp://www.springframework.org/schema/beanshttps://www.springframework.org/schema/beans/spring-beans.xsd"><rabbit:connection-factoryid="connectionFactory"host="localhost"virtual-host="/"username="guest"password="guest"/><rabbit:templateid="amqpTemplate"connection-factory="connectionFactory"/><rabbit:adminconnection-factory="connectionFactory"/><rabbit:queuename="MY-QUEUE"/></beans>

使用测试

Copyimportorg.springframework.amqp.core.AmqpTemplate;importorg.springframework.context.ApplicationContext;importorg.springframework.context.support.ClassPathXmlApplicationContext;publicclassMain{publicstaticvoidmain(String[]args){ApplicationContextapp=newClassPathXmlApplicationContext("spring/rabbit-context.xml");AmqpTemplateamqpTemplate=app.getBean(AmqpTemplate.class);amqpTemplate.convertAndSend("MY-QUEUE","Item");Stringmsg=(String)amqpTemplate.receiveAndConvert("MY-QUEUE");System.out.println(msg);}}

参考方法

/***ConvertaJavaobjecttoanAmqp{@linkMessage}andsendittoaspecificexchange*withaspecificroutingkey.**@paramexchangethenameoftheexchange*@paramroutingKeytheroutingkey*@parammessageamessagetosend*@throwsAmqpExceptionifthereisaproblem*/voidconvertAndSend(Stringexchange,StringroutingKey,Objectmessage)throwsAmqpException;/***ReceiveamessageifthereisonefromaspecificqueueandconvertittoaJava*object.Returnsimmediately,possiblywithanullvalue.**@paramqueueNamethenameofthequeuetopoll*@returnamessageornullifthereisnonewaiting*@throwsAmqpExceptionifthereisaproblem*/@NullableObjectreceiveAndConvert(StringqueueName)throwsAmqpException;

可靠性指南

本指南概述了RabbitMQ，AMQP0-9-1和其他与数据安全有关的受支持协议。它们帮助应用程序开发人员和操作员实现可靠的传递，即确保始终传递消息，即使遇到各种故障也是如此。

数据安全是RabbitMQ节点，发布者和消费者的共同责任

什么情况会失败？

基于消息的系统按定义分布，并且可能以不同的（有时甚至是微妙的）方式失败。

网络连接问题和拥塞可能是最常见的故障类别。不仅网络会发生故障，防火墙还会中断他们认为是空闲的连接，并且网络故障需要时间来检测。

除了连接故障之外，服务器和客户端应用程序还可以随时经历硬件故障（或软件可能崩溃）。此外，即使客户端应用程序继续运行，逻辑错误也可能导致通道或连接错误，从而迫使客户端建立新的通道或连接并从问题中恢复。

当然，此故障清单并非详尽无遗。它不包括更细微的故障，例如遗漏故障（无法在可预测的时间内响应），性能下降，耗尽系统资源的恶意或错误的应用程序等。可以通过监视，指标和运行状况检查来检测那些故障。

连接失败

如果客户端和RabbitMQ节点之间的网络连接失败，则客户端将需要建立与代理的新连接。在上一个连接上打开的所有通道将被自动关闭，这些通道也需要重新打开。

通常，当连接失败时，连接会抛出异常（或类似的语言构造）来通知客户端。

大多数客户端库提供的功能可从连接失败时自动恢复。对于这种认为无法恢复的情况，应用程序开发人员可以通过定义连接失败事件处理程序来实现自己的恢复。java客户端API：https://www.rabbitmq.com/api-guide.html

确认

当连接失败时，消息可能会在客户端和服务器之间传输-它们可能在任一侧被解码或编码，位于TCP堆栈缓冲区中或正在传输中。在这种情况下，传输中的消息将不会传递-它们将需要重新传输。确认使服务器和客户端知道何时执行此操作。

确认可以在两个方向上使用-允许消费者向服务器指示它已经接收和/或处理了交付，并允许服务器向发布者指示同一件事。两者都称为消费者确认和发布者确认。

尽管TCP确保将数据包已传递到连接对等方，并且将重新传输数据包直到连接对等方，但这只能处理网络层的故障。确认和确认表示消息已被对等应用程序接收并采取措施。确认既表示接收消息，也表示接收者对消息承担全部责任的所有权转移。

因此，确认具有语义。消费应用程序在完成对消息的任何处理之前，不应确认消息：将消息记录在数据存储中，转发消息或执行任何其他操作。一旦这样做，经纪人就可以自由标记要删除的交货。

同样，代理对消息负责后将确认消息。有关详细信息，请参见“确认和确认”指南。

确认的使用至少保证一次交货。如果没有确认，则在发布和使用操作期间可能会丢失消息，并且最多只能保证一次传递。

使用心跳检测死的TCP连接

在某些类型的网络故障中，数据包丢失可能意味着操作系统检测到较长的时间（例如，在Linux上为默认配置，大约需要11分钟），否则TCP连接中断。AMQP0-9-1提供心跳功能，以确保应用程序层迅速发现中断的连接（以及完全无响应的对等体）。心跳还可以防御某些可能终止“空闲”TCP连接的网络设备。有关详细信息，请参见心跳指南。

broker侧的数据安全

为了避免在代理中丢失消息，队列和消息必须能够应对代理重新启动，代理硬件故障以及极端甚至代理崩溃的情况。

为确保消息和代理定义在重新启动后仍然有效，我们需要确保它们在磁盘上。AMQP标准具有交换，队列和持久性消息的持久性概念，要求持久性对象或持久性消息在重新启动后仍可生存。有关持久性和持久性的特定标志的更多详细信息，可以在“队列”指南中找到。

集群和消息复制

节点集群提供冗余，并且可以容忍单个节点的故障。在RabbitMQ集群中，所有定义（交换，绑定，用户等）都在整个集群中复制。队列的行为有所不同，默认情况下，队列仅驻留在单个节点上，但可以配置为在多个节点之间复制（镜像）。队列保持可见并且可从所有节点访问，无论其主副本位于哪个节点。

镜像队列在许多已配置的群集节点上复制其内容。当某个节点发生故障时，队列中的主副本将托管在该节点下，并进行升级（新的主选举）。在这种情况下，关键的可靠性标准是是否存在符合升级条件的副本（队列镜像）。

互斥队列与其连接的生命周期相关联，因此从不对其进行镜像，并且从定义上讲，它在节点重启后将无法生存。

连接到故障节点的使用者必须照常恢复。当为队列选择新的主副本时，连接到其他节点的使用者将由RabbitMQ自动重新注册。这些使用者不需要执行恢复（例如，重新连接或重新订阅）。

生产者侧的数据安全

使用确认时，从通道或连接故障中恢复的生产者应重新传输尚未收到来自代理的确认的任何消息。由于代理可能发送了从未到达生产者的确认消息（由于网络故障等），因此此处可能存在消息重复。因此，消费者应用程序将需要执行重复数据删除或以幂等方式处理传入消息。

确保消息被路由

在某些情况下，对于生产者来说，确保他们的消息被路由到队列可能很重要（尽管并非总是如此-对于pub-sub系统，生产者将只发布消息，并且如果没有消费者感兴趣，则将消息正确发送是很重要的。删除）。

为了确保将消息路由到单个已知队列，生产者只需声明一个目标队列并直接发布到该队列即可。如果消息可能以更复杂的方式路由，但生产者仍然需要知道它们是否已到达至少一个队列，则可以在basic.publish上设置强制标志，以确保basic.return（包含回复代码和一些文本消息）如果没有适当绑定任何队列，则会将其发送回客户端。有关详细信息，请参见发布者指南。

生产者还应注意，发布到群集节点时，如果绑定到交换的一个或多个目标队列在群集中具有镜像，则由于节点之间的流量控制，面对节点之间的网络故障，可能会导致延迟。副本和队列主副本。有关更多详细信息，请参见节点间心跳指南。

消费者侧的数据安全

如果出现网络故障（或节点故障），则可以重新传递消息，并且消费者必须准备好处理过去看到的传递。建议将使用者实现设计为幂等的，而不是明确执行重复数据删除。

如果消息被传递给使用者，然后由RabbitMQ或由相同或不同的使用者自动重新排队，则RabbitMQ将在其再次传递时在其上设置重新传递的标志。这暗示消费者可能以前已经看过此消息。由于原始交付可能由于网络或消费者应用程序故障而未交付给任何消费者，因此无法保证。

如果未设置重新传递的标志，则可以确保该消息之前从未被看到过。因此，如果消费者发现对消息进行重复数据删除或以幂等方式处理消息更为昂贵，则只能对设置了重新传递标志集的消息进行此操作。

无法处理的交货

如果消费者决定了它不能处理的消息则可以使用它的拒绝basic.reject或basic.nack方法，无论是要求该服务器重新排队，或者不（在这种情况下，服务器可以被配置为死信它代替。

消费者取消通知

当消费者正在消费的队列被删除时，RabbitMQ将通知消费者。无论是从另一个队列中消费，还是从安全和适当的角度重新声明其最初消费的消费者，此类消费者都必须采取行动以进行恢复。

节点联合Federation和铲Shovel

RabbitMQ提供了两个插件来协助分布式节点在不可靠的网络（例如，广域网）上：Federation和Shovel。两者都将从网络故障中恢复并在必要时重新传输消息。默认情况下，两者都使用确认acknowledgements和证实confirms。

当使用Federation或Shovel连接群集时，需要确保联盟链接和Shovels可以从节点故障中恢复，包括永久性（故障停止）方案。

联盟将自动在整个下游群集中分配链接，并在下游节点发生故障时迁移它们。为了在上游节点发生故障时连接到新的上游，必须为上游指定多个上游URI，否则必须通过具有足够可用性特性的负载均衡器进行连接。

铲子可以使用多个源端点和目标端点。将使用第一个可达端点。可配置的延迟后，失败的铲将重新启动并重试。

监控和健康检查

有些故障情况很细微，很难观察或发现。例如，缓慢的连接泄漏可能会随着时间的流逝而累积，并且像慢性病一样，在一段时间内不会引起注意。监视和指标是检测多种类型故障的方法。使用Prometheus等工具收集的长期度量数据可以帮助发现系统行为中的不规则性和问题模式。

除了监视外，运行状况检查是另一个可用于检测时间点问题（即当前可观察到的问题）的工具。广泛的健康检查覆盖范围可能会导致误报，因此，进行更多的检查不一定更好。

专门的指南中涵盖了监视和健康检查。