RabbitMQ消息的可靠投递

RabbitMQ提供了两种方式来解决消息投递的可靠性，避免消息丢失或消息投递失败的情况

• confirm确认模式
• return退回模式

SpringBoot中需要开启响应配置才能使用，如果不设置消息投递失败后，默认丢弃消息，不会进行任何回调。 

spring:
  rabbitmq:
    publisher-confirm-type: correlated
    publisher-returns: true

RabbitMQ整个消息投递的过程：
producer→RabbitMQ broker→exchange→queue→consumer 【其中消息实际是存放在queue中】

工作机制

消息从producer投递到exchange，会返回一个confirmCallback。无论是否投递成功，都会返回。当ack=true，表示成功，当ack=fase，表示失败
消息从exchange投递到queue，投递失败则会返回一个returnCallback

确认模式：

设置ConnectionFactory的publisher-confirm-type=correlated，开启confirm确认模式
使用rabbitTemplate.setConfirmCallback设置回调函数，当消息发送到Exchange后回调confirm方法，在方法中判断ack，如果为true，则发送成功。如果为false，则发送失败
回退模式：

设置ConnectionFactory的publisher-returns=true，开启return确认模式
使用rabbitTemplate.setReturnCallback设置spring.rabbitmq.publisher-returns=true，当消息由Exchange路由到queue失败后，如果设置了 spring.rabbitmq.template.mandatory=true，则消息回回退给消息生产者producer，并执行回调函数

ConsumerASK

ack指Acknowledge，确认。 表示消费端收到消息后的确认方式。
三种确认方式：
acknowledge=“none”
acknowledge=“manual”
根据异常情况确认：acknowledge=“auto”，（使用麻烦,不演示）
自动确认是指，当消息一旦被Consumer接收到，则自动确认收到，并将相应 message 从 RabbitMQ 的消息缓存中移除。但是在实际业务处理中，很可能消息接收到，业务处理出现异常，那么该消息就会丢失。
手动确认方式，则需要在业务处理成功后，调用channel.basicAck()，手动签收，如果出现异常，则调用channel.basicNack()方法，让其自动重新发送消息。

 消费端限流

 设置每次处理的请求数，达到限流的作用。mq能够削峰填谷

prefetch属性是一次去拉取多少，消费端的确认模式一定要为手动。

TTL time to live：过期时间

rabbitmq可以设置队列的过期时间和消息单独的过期时间，假如都设置了以时间短的为主，

可以应用的场景，例如 下单后 24小时以为完成支付。

Rabbitmq的过期消除方法是当消息在队列的顶端等待被消费的时候才会激活，

但是队列的过期时间到达就会移除里面全部的消息。

 

死信队列

什么是死信，什么是死信队列？

一般来说，producer将消息投递到queue中，consumer从queue取出消息进行消费，但某些时候由于特定的原因导致queue中的某些消息无法被消费，这样的消息如果没有后续的处理，就变成了死信(Dead Letter)，所有的死信都会放到死信队列中。“死信”消息会被RabbitMQ进行特殊处理，如果配置了死信队列信息，那么该消息将会被丢进死信队列中，如果没有配置，则该消息将会被丢弃

• 消息被拒绝（basic.reject或basic.nack）并且requeue=false.
• 消息TTL过期
• 队列达到最大长度（队列满了，无法再添加数据到mq中）

 延迟队列（RabbitMQ并未提供延迟队列） 

​ 延迟队列存储的对象肯定是对应的延时消息，所谓"延时消息"是指当消息被发送以后，并不想让消费者立即拿到消息，而是等待指定时间后，消费者才拿到这个消息进行消费。

普通队列中的元素总是等着希望被早点取出处理，而延时队列中的元素则是希望被在指定时间得到取出和处理 。使用场景如订单系统：

但是RabbitMQ可以用TTL+死信队列实现延迟队列。

 

 日志监控以及消息追踪

(47条消息) RabbitMQ高级特性之日志监控和消息追踪_悠然予夏的博客-CSDN博客_rabbitmq监控

1、RabbitMQ日志
RabbitMQ默认日志存放路径： /var/log/rabbitmq/rabbit@xxx.log

日志包含了RabbitMQ的版本号、Erlang的版本号、RabbitMQ服务节点名称、cookie的hash值、RabbitMQ配置文件地址、内存限制、磁盘限制、默认账户guest的创建以及权限配置等等

还以通过web控制台来管理监控

2、rabbitmqctl管理和监控（Linux命令）
查看队列

# rabbitmqctl list_queues

查看exchanges

# rabbitmqctl list_exchanges

查看用户

# rabbitmqctl list_users

查看连接

# rabbitmqctl list_connections

查看消费者信息

# rabbitmqctl list_consumers

查看环境变量

# rabbitmqctl environment

查看未被确认的队列

# rabbitmqctl list_queues  name messages_unacknowledged

查看单个队列的内存使用

# rabbitmqctl list_queues name memory

查看准备就绪的队列

# rabbitmqctl list_queues name messages_ready

3、消息追踪
        在使用任何消息中间件的过程中，难免会出现某条消息异常丢失的情况。对于RabbitMQ而言，可能是因为生产者或消费者与RabbitMQ断开了连接，而它们与RabbitMQ又采用了不同的确认机制；也有可能是因为交换器与队列之间不同的转发策略；甚至是交换器并没有与任何队列进行绑定，生产者又不感知或者没有采取相应的措施；另外RabbitMQ本身的集群策略也可能导致消息的丢失。这个时候就需要有一个较好的机制跟踪记录消息的投递过程，以此协助开发和运维人员进行问题的定位。

在RabbitMQ中可以使用Firehose和rabbitmq_tracing插件功能来实现消息追踪

3.1、Firehose
        firehose的机制是将生产者投递给rabbitmq的消息，rabbitmq投递给消费者的消息按照指定的格式发送到默认的exchange上。这个默认的exchange的名称为amq.rabbitmq.trace，它是一个topic类型的exchange。发送到这个exchange上的消息的routing key为 publish.exchangename 和 deliver.queuename。其中exchangename和queuename为实际exchange和queue的名称，分别对应生产者投递到exchange的消息，和消费者从queue上获取的消息。

注意：打开 trace 会影响消息写入功能，适当打开后请关闭。

rabbitmqctl trace_on：开启Firehose命令

rabbitmqctl trace_off：关闭Firehose命令

 3.2、rabbitmq_tracing
        rabbitmq_tracing和Firehose在实现上如出一辙，只不过rabbitmq_tracing的方式比Firehose多了一层GUI的包装，更容易使用和管理。

启用插件：rabbitmq-plugins enable rabbitmq_tracing