标签：02 消费 Rabbitmq 生产者 流程 确认 RabbitMQ 消息 channel

Rabbitmq工作流程详解

生产者发送消息的流程

1. 生产者连接RabbitMQ，建立TCP连接( Connection)，开启信道（Channel） 2. 生产者声明一个Exchange（交换器），并设置相关属性，比如交换器类型、是否持久化等
2. 生产者声明一个队列井设置相关属性，比如是否排他、是否持久化、是否自动删除等
3. 生产者通过 bindingKey （绑定Key）将交换器和队列绑定（ binding ）起来
4. 生产者发送消息至RabbitMQ Broker，其中包含 routingKey （路由键）、交换器等信息
5. 相应的交换器根据接收到的 routingKey 查找相匹配的队列。
6. 如果找到，则将从生产者发送过来的消息存入相应的队列中。
7. 如果没有找到，则根据生产者配置的属性选择丢弃还是回退给生产者
8. 关闭信道。
9. 关闭连接。

消费者接收消息的流程

1. 消费者连接到RabbitMQ Broker ，建立一个连接(Connection ) ，开启一个信道(Channel) 。 2. 消费者向RabbitMQ Broker 请求消费相应队列中的消息，可能会设置相应的回调函数， 以及
   做一些准备工作
2. 等待RabbitMQ Broker 回应并投递相应队列中的消息， 消费者接收消息。
3. 消费者确认( ack) 接收到的消息。
4. RabbitMQ 从队列中删除相应己经被确认的消息。
5. 关闭信道。
6. 关闭连接。

消息可靠性（重要理解）

场景： 一般我们使用支付宝或微信转账支付的时候，都是扫码，支付，然后立刻得到结果，说你支付了多少钱如果你绑定的是银行卡，可能这个时候你并没有收到支付的确认消息。往往是在一段时间之后，你会收到银行卡发来的短信，告诉你支付的信息。

支付平台如何保证这笔帐不出问题？

这种情况下必须保证数据正确性，保证数据并发安全性，保证数据一致性。

保证数据一致性的两种方式：

1. 分布式锁
   这个比较容易理解，就是在操作某条数据时先锁定，可以用redis或zookeeper等常用框架来实现。 比如我们在修改账单时，先锁定该账单，如果该账单有并发操作，后面的操作只能等待上一个操作的锁释放后再依次执行。

优点：能够保证数据强一致性。
缺点：高并发场景下可能有性能问题。

2. 消息队列
   消息队列是为了保证最终一致性，我们需要确保消息队列有ack机制，客户端收到消息并消费处理完成后，客户端发送ack消息给消息中间件 如果消息中间件超过指定时间还没收到ack消息，则定时去重发消息。

优点：异步、高并发
缺点：有一定延时、数据弱一致性，并且必须能够确保该业务操作肯定能够成功完成，不可能失败。

对于消息的可靠性，我们可以通过以下几个方面来保证：

1. 客户端代码中的异常捕获，包括生产者和消费者
2. AMQP/RabbitMQ的事务机制
3. 发送端确认机制
4. 消息持久化机制
5. Broker端的高可用集群
6. 消费者确认机制
7. 消费端限流
8. 消息幂等性

异常捕获机制

先执行行业务操作，业务操作成功后执行消息发送，消息发送过程通过try catch方式捕获异常，在异常处理的代码块中执行回滚业务操作或者执行重发操作等。这是一种最大努力确保的方式，并无法保证100%绝对可靠，因为这里没有异常并不代表消息就一定投递成功。

AMQP/RabbitMQ的事务机制

没有捕获到异常并不能代表消息就一定投递成功了。
一直到事务提交后都没有异常，确实就说明消息是投递成功了。但是，这种方式在性能方面的开销比较大，一般也不推荐使用。

发送端确认机制

RabbitMQ后来引入了一种轻量量级的方式，叫发送方确认(publisher confirm)机制。生产者将信道设置成confirm(确认)模式，一旦信道进入confirm 模式，所有在该信道上⾯面发布的消息都会被指派一个唯一的ID(从1 开始)，一旦消息被投递到所有匹配的队列之后（如果消息和队列是持久化的，那么确认消息会在消息持久化后发出），RabbitMQ 就会发送一个确认(Basic.Ack)给生产者(包含消息的唯一ID)，这样生产者就知道消息已经正确送达了。

RabbitMQ 回传给生产者的确认消息中的deliveryTag 字段包含了确认消息的序号，另外，通过设置channel.basicAck方法中的multiple参数，表示到这个序号之前的所有消息是否都已经得到了处理了。生产者投递消息后并不需要一直阻塞着，可以继续投递下一条消息并通过回调方式处理理ACK响应。如果 RabbitMQ 因为自身内部错误导致消息丢失等异常情况发生，就会响应一条nack(Basic.Nack)命令，生产者应用程序同样可以在回调方法中处理理该 nack 命令。

Connection connection = factory.newConnection(); 
Channel channel = connection.createChannel(); 
// Publisher Confirms 
channel.confirmSelect();

channel.exchangeDeclare(EX_PUBLISHER_CONFIRMS, BuiltinExchangeType.DIRECT); 

channel.queueDeclare(QUEUE_PUBLISHER_CONFIRMS, false, false, false, null); 

channel.queueBind(QUEUE_PUBLISHER_CONFIRMS, EX_PUBLISHER_CONFIRMS, QUEUE_PUBLISHER_CONFIRMS);

 String message = "hello"; 
 
 channel.basicPublish(EX_PUBLISHER_CONFIRMS, QUEUE_PUBLISHER_CONFIRMS, null, message.getBytes());
 
  try {
  	channel.waitForConfirmsOrDie(5_000);
  	System.out.println("消息被确认：message = " + message);
} catch (IOException e) { 
  	 	e.printStackTrace(); 
  	 	System.err.println("消息被拒绝！ message = " + message); 
} catch (InterruptedException e) { 
		e.printStackTrace(); 
		System.err.println("在不是Publisher Confirms的通道上使用该方法"); 
} catch (TimeoutException e) { 
		e.printStackTrace(); 
		System.err.println("等待消息确认超时！ message = " + message); 
}

持久化存储机制

持久化是提高RabbitMQ可靠性的基础，否则当RabbitMQ遇到异常时（如：重启、断电、停机等）数据将会丢失。主要从以下几个方面来保障消息的持久性：

1. Exchange的持久化。通过定义时设置durable 参数为ture来保证Exchange相关的元数据不不
   丢失。
2. Queue的持久化。也是通过定义时设置durable 参数为ture来保证Queue相关的元数据不不
   丢失。
3. 消息的持久化。通过将消息的投递模式 (BasicProperties 中的 deliveryMode 属性)设置为 2
   即可实现消息的持久化，保证消息自身不丢失。

RabbitMQ中的持久化消息都需要写入磁盘（当系统内存不不足时，非持久化的消息也会被刷盘处理），这些处理动作都是在“持久层”中完成的。持久层是一个逻辑上的概念，实际包含两个部分：

1. 队列索引(rabbit_queue_index)，rabbit_queue_index 负责维护Queue中消息的信息，包括
   消息的存储位置、是否已交给消费者、是否已被消费及Ack确认等，每个Queue都有与之对应
   的rabbit_queue_index。
2. 消息存储(rabbit_msg_store)，rabbit_msg_store 以键值对的形式存储消息，它被所有队列列
   共享，在每个节点中有且只有一个。

Consumer ACK

如何保证消息被消费者成功消费？

前面我们讲了生产者发送确认机制和消息的持久化存储机制，然而这依然无法完全保证整个过程的可靠性，因为如果消息被消费过程中业务处理失败了但是消息却已经出列了（被标记为已消费了），我们又没有任何重试，那结果跟消息丢失没什么分别。

RabbitMQ在消费端会有Ack机制，即消费端消费消息后需要发送Ack确认报文给Broker端，告知自己是否已消费完成，否则可能会一直重发消息直到消息过期（AUTO模式）。

一般来说，我们有三种处理手段：

1. 采用NONE模式，消费的过程中自行捕获异常，引发异常后直接记录日志并落到异常恢复表，
   再通过后台定时任务扫描异常恢复表尝试做重试动作。如果业务不自行处理则有丢失数据的风
   险
2. 采用AUTO（自动Ack）模式，不主动捕获异常，当消费过程中出现异常时会将消息放回
   Queue中，然后消息会被重新分配到其他消费者节点（如果没有则还是选择当前节点）重新
   被消费，默认会一直重发消息并直到消费完成返回Ack或者一直到过期

但是采用AUTO，有可能消费者自动返回了ACK，但是消息还未消费但是消费者宕机了，同样有可能导致消息丢失。

3. 采用MANUAL（手动Ack）模式，消费者自行控制流程并手动调用channel相关的方法返回
   Ack

/** NONE模式，则只要收到消息后就立即确认（消息出列，标记已消费），有丢失数据的风险 
	 * AUTO模式，看情况确认，如果此时消费者抛出异常则消息会返回到队列中 
	 * MANUAL模式，需要显式的调用当前channel的basicAck方法 
	 * @param channel 
	 * @param deliveryTag 
	 * @param message 
	 */ 
	 @RabbitListener(queues = "lagou.topic.queue", ackMode = "AUTO") 
	 public void handleMessageTopic(Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, @Payload byte[] message) {
	 
	 System.out.println("RabbitListener消费消息，消息内容：" + new String((message))); 
	 
	 try {
	 
	 // 手动ack，deliveryTag表示消息的唯一标志，multiple表示是否是批量确认 
	 channel.basicAck(deliveryTag, false); 
	 
	 // 手动nack，告诉broker消费者处理失败，最后一个参数表示是否需要将消息重新 入列 
	 channel.basicNack(deliveryTag, false, true); 
	 
	 // 手动拒绝消息。第二个参数表示是否重新入列 
	 channel.basicReject(deliveryTag, true); 
	 
	} catch (IOException e) { 
		e.printStackTrace();
	 } 

}

消费端限流

消息可靠性保障

1. 消息传输保障
2. 各种限流、应急手段
3. 业务层面的一些容错、补偿、异常重试等手段

消息可靠传输一般是业务系统接入消息中间件时首要考虑的问题，一般消息中间件的消息传输保障分为三个层级：

1. At most once：最多一次。消息可能会丢失，但绝不会重复传输
2. At least once：最少一次。消息绝不会丢失，但可能会重复传输
3. Exactly once：恰好一次。每条消息肯定会被传输一次且仅传输一次

RabbitMQ 支持其中的“最多一次”和“最少一次”。

其中“最少一次”投递实现需要考虑以下这个几个方面的内容：

1. 消息生产者需要开启事务机制或者publisher confirm 机制，以确保消息可以可靠地传输到
   RabbitMQ 中。
2. 消息生产者需要配合使用 mandatory 参数或者备份交换器来确保消息能够从交换器路由到队
   列中，进而能够保存下来而不会被丢弃。
3. 消息和队列都需要进行持久化处理，以确保RabbitMQ 服务器在遇到异常情况时不会造成消息
   丢失。
4. 消费者在消费消息的同时需要将autoAck 设置为false，然后通过手动确认的方式去确认已经
   正确消费的消息，以避免在消费端引起不必要的消息丢失。

“最多一次”的方式就无须考虑以上那些方面，生产者随意发送，消费者随意消费，不过这样很难确保消息不会丢失。

“恰好一次”是RabbitMQ 目前无法保障的。

考虑这样一种情况，消费者在消费完一条消息之后向RabbitMQ 发送确认Basic.Ack 命令，此时由于网络断开或者其他原因造成RabbitMQ 并没有收到这个确认命令，那么RabbitMQ 不会将此条消息标记删除。在重新建立连接之后，消费者还是会消费到这一条消息，这就造成了重复消费。

再考虑一种情况，生产者在使用publisher confirm机制的时候，发送完一条消息等待RabbitMQ返回确认通知，此时网络断开，生产者捕获到异常情况，为了确保消息可靠性选择重新发送，这样RabbitMQ 中就有两条同样的消息，在消费的时候消费者就会重复消费。

消息幂等性处理

追求高性能就无法保证消息的顺序，而追求可靠性那么就可能产生重复消息，从而导致重复消费。

一般解决重复消息的办法是，在消费端让我们消费消息的操作具备幂等性。

一个幂等操作的特点是，其任意多次执行所产生的影响均与一次执行的影响相同。一个幂等的方法，使用同样的参数，对它进行多次调用和一次调用，对系统产生的影响是一样的。

对于幂等的方法，不用担心重复执行会对系统造成任何改变。

业界对于幂等性的一些常见做法：

1. 借助数据库唯一索引，重复插入直接报错，事务回滚。
2. 前置检查机制，比如乐观锁。
3. 唯一Id机制，比较通用的方式。对于每条消息我们都可以生成唯一Id，消费前判断Tair中是否
   存在（MsgId做Tair排他锁的key），消费成功后将状态写入Tair中，这样就可以防止重复消费
   了。

标签：02,消费,Rabbitmq,生产者,流程,确认,RabbitMQ,消息,channel
来源： https://blog.csdn.net/weixin_43743650/article/details/114436315

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。