RabbitMQ消息的可靠性投递

　　RabbitMQ是实现了高级消息队列协议（AMQP）的开源消息代理软件（亦称面向消息的中间件），广泛微服务系统之间，服务之间消息异步投递。如何保证消息在投递过程的中可靠性呢。下面的文章我这里探讨一下。服务端确认Transaction模式。
　　1。将chanel设置成事务模式。
　　基于Java，api的设置方式。try｛channel。txSelect（）；channel。queueDeclare（simplequeue，true，false，false，null）；创建消息。Stringmessagehellokeelon；发送消息。1。消息要发送的交换机对象，不写使用默认defaultexchange2。当前的路由地址。路由地址，可以写成队列地址3。附加数据可以不写。4。消息channel。basicPublish（，simplequeue，null，message。getBytes（StandardCharsets。UTF8））；channel。txCommit（）；connection。close（）；｝catch（Exceptione）｛｝
　　使用springBoot集成的方式。TestvoidTestSendMessage（）｛rabbitTemplate。setChannelTransacted（true）；rabbitTemplate。convertAndSend（itemtopicexchange，item。add，hellospringBootmq）；｝
　　服务端确认的事务方式，这个是阻塞的一条消息没有发送完毕。不能发送下一条消息，影响系统的性能。不建议使用。服务端确认Confirm模式。
　　开启发送方确认的模式channel。confirmSelect（）；channel。basicPublish（fanoutexchange，，null，message。getBytes（StandardCharsets。UTF8））；if（channel。waitForConfirms（））｛System。out。print（消息发送成功）；｝channel。close（）；connection。close（）；
　　2。异步确认的方式。channel。addConfirmListener（newConfirmListener（）｛OverridepublicvoidhandleAck（longdeliveryTag，booleanmultiple）throwsIOException｛System。out。print（Brooker未确认消息，标识deliveryTag）；if（multiple）｛headerSet表示后面参数之前的参数全部删除confirmset。headSet（deliveryTag1L）。clear（）；｝else｛confirmset。remove（deliveryTag）；｝｝OverridepublicvoidhandleNack（longdeliveryTag，booleanmultiple）throwsIOException｛System。out。print（String。format（服务端已经确认消息，标识：d，多个消息b，deliveryTag））；if（multiple）｛headerSet表示后面参数之前的参数全部删除confirmset。headSet（deliveryTag1L）。clear（）；｝else｛confirmset。remove（deliveryTag）；｝｝｝）；
　　springBoot的集成的实现。rabbitTemplate。setConfirmCallback（newRabbitTemplate。ConfirmCallback（）｛Overridepublicvoidconfirm（CorrelationDatacorrelationData，booleanack，Stringcause）｛if（！ack）｛System。out。println（发送消息失败：cause）；thrownewRuntimeException（发送消息失败：cause）；｝｝｝）；rabbitTemplate。convertAndSend（itemtopicexchange，item。add，hellospringBootmq）；｝路由保证
　　在channel中添加listenner监听无法路由的消息。channel。addReturnListener（newReturnListener（）｛OverridepublicvoidhandleReturn（intreplyCode，StringreplyText，Stringexchange，StringroutingKey，AMQP。BasicPropertiesbasicProperties，byte〔〕bytes）throwsIOException｛System。out。println（监听器收到无法路由，被返回的消息）；｝｝）；消息的持久化
　　消息，交换机，队列的持久化处理。AMQP。BasicPropertiespropertiesnewAMQP。BasicProperties。Builder（）。deliveryMode（2）2代表持久化。contentEncoding（UTF8）编码。expiration（10000）过期时间。build（）；RabbitMQ集群
　　RabbitMQ最优秀的功能之一就是内建集群，这个功能设计的目的是允许消费者和生产者在节点崩溃的情况下继续运行，以及通过添加更多的节点来线性扩展消息通信吞吐量。RabbitMQ内部利用Erlang提供的分布式通信框架OTP来满足上述需求，使客户端在失去一个RabbitMQ节点连接的情况下，还是能够重新连接到集群中的任何其他节点继续生产、消费消息。RabbitMQ集群中的一些概念
　　RabbitMQ会始终记录以下四种类型的内部元数据：队列元数据
　　包括队列名称和它们的属性，比如是否可持久化，是否自动删除交换器元数据
　　交换器名称、类型、属性绑定元数据
　　内部是一张表格记录如何将消息路由到队列vhost元数据
　　为vhost内部的队列、交换器、绑定提供命名空间和安全属性
　　在单一节点中，RabbitMQ会将所有这些信息存储在内存中，同时将标记为可持久化的队列、交换器、绑定存储到硬盘上。存到硬盘上可以确保队列和交换器在节点重启后能够重建。而在集群模式下同样也提供两种选择：存到硬盘上（独立节点的默认设置），存在内存中。
　　如果在集群中创建队列，集群只会在单个节点而不是所有节点上创建完整的队列信息（元数据、状态、内容）。结果是只有队列的所有者节点知道有关队列的所有信息，因此当集群节点崩溃时，该节点的队列和绑定就消失了，并且任何匹配该队列的绑定的新消息也丢失了。还好RabbitMQ2。6。0之后提供了镜像队列以避免集群节点故障导致的队列内容不可用。
　　RabbitMQ集群中可以共享user、vhost、exchange等，所有的数据和状态都是必须在所有节点上复制的，例外就是上面所说的消息队列。RabbitMQ节点可以动态的加入到集群中。
　　当在集群中声明队列、交换器、绑定的时候，这些操作会直到所有集群节点都成功提交元数据变更后才返回。集群中有内存节点和磁盘节点两种类型，内存节点虽然不写入磁盘，但是它的执行比磁盘节点要好。内存节点可以提供出色的性能，磁盘节点能保障配置信息在节点重启后仍然可用，那集群中如何平衡这两者呢？
　　RabbitMQ只要求集群中至少有一个磁盘节点，所有其他节点可以是内存节点，当节点加入火离开集群时，它们必须要将该变更通知到至少一个磁盘节点。如果只有一个磁盘节点，刚好又是该节点崩溃了，那么集群可以继续路由消息，但不能创建队列、创建交换器、创建绑定、添加用户、更改权限、添加或删除集群节点。换句话说集群中的唯一磁盘节点崩溃的话，集群仍然可以运行，但知道该节点恢复，否则无法更改任何东西。RabbitMQ集群配置和启动
　　如果是在一台机器上同时启动多个RabbitMQ节点来组建集群的话，只用上面介绍的方式启动第二、第三个节点将会因为节点名称和端口冲突导致启动失败。所以在每次调用rabbitmqserver命令前，设置环境变量RABBITMQNODENAME和RABBITMQNODEPORT来明确指定唯一的节点名称和端口。下面的例子端口号从5672开始，每个新启动的节点都加1，节点也分别命名为testrabbit1、testrabbit2、testrabbit3。
　　启动第1个节点：RABBITMQNODENAMEtestrabbit1RABBITMQNODEPORT5672。sbinrabbitmqserverdetached
　　启动第2个节点：RABBITMQNODENAMEtestrabbit2RABBITMQNODEPORT5673。sbinrabbitmqserverdetached
　　启动第2个节点前建议将RabbitMQ默认激活的插件关掉，否则会存在使用了某个插件的端口号冲突，导致节点启动不成功。
　　现在第2个节点和第1个节点都是独立节点，它们并不知道其他节点的存在。集群中除第一个节点外后加入的节点需要获取集群中的元数据，所以要先停止Erlang节点上运行的RabbitMQ应用程序，并重置该节点元数据，再加入并且获取集群的元数据，最后重新启动RabbitMQ应用程序。
　　停止第2个节点的应用程序：。sbinrabbitmqctlntestrabbit2stopapp
　　重置第2个节点元数据：。sbinrabbitmqctlntestrabbit2reset
　　第2节点加入第1个节点组成的集群：。sbinrabbitmqctlntestrabbit2joinclustertestrabbit1localhost
　　启动第2个节点的应用程序。sbinrabbitmqctlntestrabbit2startapp
　　第3个节点的配置过程和第2个节点类似：RABBITMQNODENAMEtestrabbit3RABBITMQNODEPORT5674。sbinrabbitmqserverdetached。sbinrabbitmqctlntestrabbit3stopapp。sbinrabbitmqctlntestrabbit3reset。sbinrabbitmqctlntestrabbit3joinclustertestrabbit1localhost。sbinrabbitmqctlntestrabbit3startappRabbitMQ集群运维
　　停止某个指定的节点，比如停止第2个节点：RABBITMQNODENAMEtestrabbit2。sbinrabbitmqctlstop
　　查看节点3的集群状态：。sbinrabbitmqctlntestrabbit3clusterstatus
　　作者：预流
　　链接：https：www。jianshu。comp79ca08116d57
　　来源：简书
　　著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。