RabbitMQ 相关问题总结--RabbitMQ 如何确保消息发送和消费？

àì夳堔傛蜴生んèń 2022-03-11 02:12 361阅读 0赞

# 转载：[https://blog.51cto.com/4925054/2096599][https_blog.51cto.com_4925054_2096599] #

[![wKiom1LOBNfTtnepAAAOe2gGtLk660_middle.jpg][]][wKiom1LOBNfTtnepAAAOe2gGtLk660_middle.jpg 1]

[yunlielai][wKiom1LOBNfTtnepAAAOe2gGtLk660_middle.jpg 1]关注0人评论13216人阅读2018-04-10 17:13:41

## RabbitMQ 相关问题总结 ##

**HA 的RabbitMQ 集群架构：**

![图片.png][.png]

一、RabbitMQ 如何高可用部署，如何确保集群不宕机？

RabbitMQ可用采用三种方式来部署集群：

1. cluster

a. 不支持跨网段，用于同一个网段内的局域网。

b.可用随意的或者动态的增加 或者减少。

c.节点之间需要运行相同版本的RabbitMQ 和Erlang

2.federation

部署在广域网，允许单台服务器上的交换机或者队列，接收发布到另外一台机器上的消息和队列。

3.sholve

和federation类似，工作在更低层次，可以应用于广域网

二、节点类型：

1.RAM Node

内存节点，将所有的队列，交换机， 绑定，用户 权限与vhost的元数据存储在内存中，可以让队列和交换机声明更加的便捷。

2. Disk Node

将元数据存储在磁盘中，单节点系统，只运行 磁盘类型的节点，防止重启RabbitMQ时，丢失系统的配置信息。

三、ErLang Cookie

ErLang Cookie 是保证不同节点之间的通讯，不同节点之间共享相同的Cookie,集群部署时候，需要copy这个数据到不同的节点,使得cookie一致。

四、RabbitMQ集群模式分类

1.普通模式：

默认的集群模式，假设集群上有两个节点 node1,node2, 消息实体只存在一个节点，如果生产者生产一个消息，丢向node1，但是消费者在node2进行消费，那么就需要node2将node1中的消息取出，并且发送给消费者。

2. 镜像模式：

需要将消费者的队列变成镜像队列，存在于多个节点。实现RabbitMQ的高可用，作用就是，消息实体会在队列之间同步。

RabbitMQ相关操作命令：

1. 单个节点的服务启动 停止

<code class="language-java">rabbitmqctl stop 
 rabbitmq-server -detached</code>

2.查询节点状态

<code class="language-java">rabbitmqctl cluster_status</code>

调研：

<code class="language-java">1. 产品定位（能做什么，不能做什么） 
 2. 产品特性（适合做什么，不适合做什么） 
 3. 产品背景（社区活跃度，团队知名度，维护力度，口碑） 
 4. 产品架构（实现原理） 
 5. 产品安装（用户指南） 
 6. 产品维护（运维管理）</code>

demo 验证

## **RabbitMQ 如何确保消息发送和消费:** ##

**![20180328093337050][]**

从RabbitMQ 结构图来看，有如下几个 过程：

1. 消息从生产者Producer发送到交换机Exchange

2.交换机根据路由规则将消息转发到相应队列

3. 队列将消息进行存储

4.消费者订阅队列消息，并进行消费

第一个过程  消息 从生产者发送到交换机

a.中间网络断开怎么办？

rabbitmq 的解决方案：

1). 设置信道channel 为事务模式

通过channel.txSelect 开启事务，channel.txCommit 提交事务，channel.txRollback 用于事务回滚

如果在还没有提交事务之前，RabbitMQ抛出异常，我们可以 将其捕获，然后进行事务回滚。缺点是 事务模式会极大的消耗RabbitMQ的性能。

2). 设置信道confirm 模式

通过confirm.select开启confirm模式，如果设置了no\_wait为 false的话，那么broker会返回confirm.select\_ok,表示broker同意将信道设置为confirm模式。 但是这个优点是发送方确认是异步的，发送方可以不等到确认就发送下一条消息。当消息被brocker接收，broker 会发送basic.ack，生产者可以通过回调函数确认这个消息，如果因为RabbitMQ本身问题导致消息丢失，broker会发送basic.nack，生产者通过回调方法接收到nack后，可以考虑消息重发

要注意的是 事务模式 和confirm模式不能共存，是互斥的。

第二个过程  消息由交换机路由到消息队列

对于不能路由到消息队列的消息，如果设置了mandatory的话，basic.return 会在basic.ack或者basic.nack之前返回。

对于可以路由到消息队列的消息，在confirm模式下，如果返回basic.ack的话说明如下：

1.消息被接收到所有的队列中了 2.如果是镜像队列，说明被所有的镜像队列接收 3.如果是持久化的消息到持久化队列已经持久化到看硬盘。

<code class="language-java">@Service("returnCallBackListener") 
 public class ReturnCallBackListener implements ReturnCallback { 
 
 @Override 
 public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) { 
 String msgId = ""; 
 if (message.getMessageProperties().getCorrelationId() != null) { 
 msgId = new String(message.getMessageProperties().getCorrelationId()); 
 } 
 System.out.println("return--message: msgId:" + msgId + ",msgBody:" + new String(message.getBody()) 
 + ",replyCode:" + replyCode + ",replyText:" + replyText + ",exchange:" + exchange + ",routingKey:" 
 + routingKey); 
 } 
 
 }</code>

当然如果不想在生产者实行return 函数，可以采用备份交换机的方式，当交换机没有路由到正确队列，会将消息转发到这个备份交换机上，前提是这个备份交换机绑定了正确队列，这样就可以被消费了。channel.exchangeDeclare方法的时候添加alternate-exchange参数声明备份交换机

第三个阶段， 队列消息持久化阶段

如果队列不设置持久化，即使消息是持久化的，消息依然不能保存，皮之不存，毛将焉附需要队列和消息都是持久化的，队列持久化只是保存其元数据，在重启，宕机后不丢失，但是要消息也持久化，需要保证消息是持久化的。并且如果是设置了confim模式的话，basic.ack 之前是会将数据进行落盘的， 并且RabbitMQ采用镜像队列，多个副本方式，Master 宕机，依然可以使用Slave 不影响 集群正常使用，保证高可用

第4个阶段  消费者消费消息

为了保证消息正常的到达消费者，RabbitMQ 提供了消息 acknowledgement来确认消息。

默认autoAck=None   isAutoAck=false 来确认消息， autoAck = false 是表示，需要等到消费者发送显示的回复确认信号之后，消息才从内存中移除，但是如果acknowledgeMode设置了Auto ,那么isAutoAck= true  ，这个其实是不安全的，fire-And-forget, 消息发送出去之后，但是可能还没到消费者，TCP连接就断了（由于配置了auto，只要消费发出去了，就删掉了）,那么TCP连接断开了，这部分消息就丢失了，是不安全的，但是对应能一直keepup连接的，是可以提高吞吐量的。

还可以设置autoAck = manual ，isAutoack= false,那么就是 队列每次向消费者发送消息之后，需要消费者手动确认basc.ack,basic,nack等，rabbitmq才可以将消息删除或者重新入队。 isAutoack=false 情况下 ，一致没有收到  消费者的 basic.ack, RabbitMQ如果检测到 和消费者端口连接 端口，会重新发这条消息。

还有注意如果basic.nack basic.reject 只是简单的拒绝 ，而不是重新 requeue的话，那么消息是不会重新入队的

查看下basic.Nack说明

<code class="language-java">/** 
 * Reject one or several received messages. 
 * 
 * Supply the <code>deliveryTag</code> from the {@link com.rabbitmq.client.AMQP.Basic.GetOk} 
 * or {@link com.rabbitmq.client.AMQP.Basic.GetOk} method containing the message to be rejected. 
 * @see com.rabbitmq.client.AMQP.Basic.Nack 
 * @param deliveryTag the tag from the received {@link com.rabbitmq.client.AMQP.Basic.GetOk} or {@link com.rabbitmq.client.AMQP.Basic.Deliver} 
 * @param multiple true to reject all messages up to and including 
 * the supplied delivery tag; false to reject just the supplied 
 * delivery tag. 
 * @param requeue true if the rejected message(s) should be requeued rather 
 * than discarded/dead-lettered 
 * @throws java.io.IOException if an error is encountered 
 */ 
 void basicNack(long deliveryTag, boolean multiple, boolean requeue) 
 throws IOException;</code>

最后一个参数设置true的话才会重新入队

<code class="language-java">channel.basicNack(message.getMessageProperties().getDeliveryTag(), false,true);</code>

查看源码：

<code class="language-java">private void consumeFromQueue(String queue) throws IOException { 
 this.channel.basicConsume(queue, this.acknowledgeMode.isAutoAck(), "", false, this.exclusive, 
 this.consumerArgs, this.consumer); 
 if (logger.isDebugEnabled()) { 
 logger.debug("Started on queue '" + queue + "': " + this); 
 } 
 }</code>

还有个问题，消费者啥时候发送basic.ack呢？

<code class="language-java">Object[] listenerArguments = buildListenerArguments(convertedMessage); 
 Object result = invokeListenerMethod(methodName, listenerArguments, message); 
 if (result != null) { 
 handleResult(result, message, channel); 
 } else { 
 logger.trace("No result object given - no result to handle"); 
 }</code>

、、、、、、、、、、、、、、

<code class="language-java">rotected void doInvokeListener(ChannelAwareMessageListener listener, Channel channel, Message message) 
 throws Exception { 
 
 RabbitResourceHolder resourceHolder = null; 
 Channel channelToUse = channel; 
 boolean boundHere = false; 
 try { 
 if (!isExposeListenerChannel()) { 
 // We need to expose a separate Channel. 
 resourceHolder = getTransactionalResourceHolder(); 
 channelToUse = resourceHolder.getChannel(); 
 /* 
 * If there is a real transaction, the resource will have been bound; otherwise 
 * we need to bind it temporarily here. Any work done on this channel 
 * will be committed in the finally block. 
 */ 
 if (isChannelLocallyTransacted(channelToUse) && 
 !TransactionSynchronizationManager.isActualTransactionActive()) { 
 resourceHolder.setSynchronizedWithTransaction(true); 
 TransactionSynchronizationManager.bindResource(this.getConnectionFactory(), 
 resourceHolder); 
 boundHere = true; 
 } 
 } 
 else { 
 // if locally transacted, bind the current channel to make it available to RabbitTemplate 
 if (isChannelLocallyTransacted(channel)) { 
 RabbitResourceHolder localResourceHolder = new RabbitResourceHolder(channelToUse, false); 
 localResourceHolder.setSynchronizedWithTransaction(true); 
 TransactionSynchronizationManager.bindResource(this.getConnectionFactory(), 
 localResourceHolder); 
 boundHere = true; 
 } 
 } 
 // Actually invoke the message listener... 
 try { 
 listener.onMessage(message, channelToUse); 
 } 
 catch (Exception e) { 
 throw wrapToListenerExecutionFailedExceptionIfNeeded(e, message); 
 } 
 } 
 finally { 
 if (resourceHolder != null && boundHere) { 
 // so the channel exposed (because exposeListenerChannel is false) will be closed 
 resourceHolder.setSynchronizedWithTransaction(false); 
 } 
 ConnectionFactoryUtils.releaseResources(resourceHolder); 
 if (boundHere) { 
 // unbind if we bound 
 TransactionSynchronizationManager.unbindResource(this.getConnectionFactory()); 
 if (!isExposeListenerChannel() && isChannelLocallyTransacted(channelToUse)) { 
 /* 
 * commit the temporary channel we exposed; the consumer's channel 
 * will be committed later. Note that when exposing a different channel 
 * when there's no transaction manager, the exposed channel is committed 
 * on each message, and not based on txSize. 
 */ 
 RabbitUtils.commitIfNecessary(channelToUse); 
 } 
 } 
 } 
 }</code>

参考：https://blog.csdn.net/wangming520liwei/article/details/79523130

[https_blog.51cto.com_4925054_2096599]: https://blog.51cto.com/4925054/2096599
[wKiom1LOBNfTtnepAAAOe2gGtLk660_middle.jpg]: /images/20220311/64c12bf50a8040ef9eede30707a64492.png
[wKiom1LOBNfTtnepAAAOe2gGtLk660_middle.jpg 1]: https://blog.51cto.com/4925054
[.png]: /images/20220311/f4fb44f06d0146c98ff87347eee14c64.png
[20180328093337050]: /images/20220311/7d65075601ec40768d582045255d1ac8.png