RabbitMQ

队列对比

使用docker进行安装

• 下载management的版本，带web界面
• 命令：

docker run -d --name rabbitmq3.7.7 -p 5672:5672 -p 15672:15672 -v `pwd`/data:/var/lib/rabbitmq --hostname myRabbit -e RABBITMQ_DEFAULT_VHOST=my_vhost  -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=123456 rabbitmq:3.7.13-management

• 参数：

  • -d：后台运行
  • --name : 指定容器名
  • -p ： 端口映射
  • --hostname : rabbitmq主机名
  • -e RABBITMQ_DEFAULT_VHOST=my_vhost 虚拟机名
  • -e RABBITMQ_DEFAULT_USER=admin 用户名
  • -e RABBITMQ_DEFAULT_PASS=123456 密码

• 相关概念：

  • 发送消息 生产者
  • 队列、接收消息 消费者
  • 交换机：做了一层抽象。发送消息，和队列通过交换机来做转发。交换机会根据分发策略把消息转给队列。
  • 虚拟主机：支持权限控制，最小粒度为虚拟主机，一个虚拟主机可以包含多个交换机，队列，绑定
  • 队列：缓存消息的容器
  • 绑定： 设置交换机与队列的关系

• 交换机

  • 扇形交换机：Fanout exchange
  • 直连交换机：Direct exchange
  • 主题交换机：Topic exchange
  • 首部交换机：Headers exchange

• 应用场景（优点）

  • 解耦
  • 异步
  • 削峰

• 缺点：

  • 增加了系统的复杂性
  • 增加了维护的难度
  • 降低系统的稳定性：引入依赖越多，越容易挂掉

问题

• 消息的重复消费：

  • 发送端

    • 发送端发送消息给中间件，中间件收到消息并成功存储，此时的中间件出现了问题，导致发送端没有收到发送成功的返回进行重试从而产生重复
    • 中间件由于负载过高，响应变慢，成功把消息储存到中间件后，返回成功时间超时进行重试从而产生重复
    • 消息中间件成功写入消息存储，在返回结果时，出现网络问题，导致超时，而重试时，网络恢复出现重复

  • 消费端

    • 消费者收到消息进行处理，处理完毕程序出现问题，中间件不知道处理结果，会再次投递
    • 消费者收到消息进行处理，处理完毕后网络出现问题，中间件没有收到处理结果，再次投递
    • 消费端处理消息花费时间过长，中间件由于消息超时会再次投递。
    • 收到消费者的处理结果，中间件出现问题，未及时更改状态，再次投递
    • 处理完毕后，消息中间件收到结果，但是遇到消息存储故障，没能更新状态，在次投递

发送端产生消息重复的主要原因是：消息成功写入消息存储后，因为各种原因使得消息发送端没有收到成功的返回结果，并且进行重试，因而导致的重复。
消息接收端消息重复的主要原因是：消息接收者成功处理完消息后，消息中间件不能及时更新投递状态造成。

• 如何解决重复消费：
  • 幂等性控制：多次调用得到相同的结果
  • 消费者发送消息进行数据更新时，需要带上数据的版本号。
  • 去重表：利用数据库的特性实现幂等。常用的思路就是在表上构建一个唯一性的索引，保证某类数据一旦执行完毕，后续同样的请求不在处理了。或者创建一个去重表，每次操作时都去该表中查看一下id是否已经存在了，如果存在则直接返回。

如何保证数据的可靠性传输

• 生产者弄丢了数据

  • 选择用rabbitmq提供的事务功能。生产者在发送数据的之前开启RabbitMQ事务 ，channel.txSelect . 然后发送消息。如果消息没有被成功的接收到，那么生产者会收到异常报错。此时可以回滚事务，然后重试发送。如果收到了消息可以提交事务。缺点 就是吞吐量会下降，太耗性能。
  • 开启confirm模式。开启之后，每次都会分配一个唯一的id,如果写入了队列中，rabbitmq会回传一个ack,如果没有处理这个消息，会回调你的nack接口。告诉你接收失败，可以重试。
  • 事务机制和confirm最大的区别是 事务时同步的。你提交一个事务后悔阻塞在哪里。confirm是异步的。你发送消息之后会发送下一个消息，然后那个消息rabbitmq接收后会异步回调你的接口

• rabbitmq 自己弄丢了数据

  • ，必须开启rabbitmq的持久化，就是消息会写入之后持久化到硬盘上，哪怕是rabbitmq挂了重启之后，数据还是能恢复的。
    • 设置持久化的两个步骤：
      • 创建queue和交换器的时候将其设置为持久化。这样可以保证rabbitmq持久化相关的元数据。但是不会持久化队列中的数据
      • 发送消息的时候将消息的deliveryMode设置为2，就是将消息设置为持久化（默认为2，可不设置）。

• 消费端弄丢了数据：

  • rabbitmq如果丢失了数据，主要是因为你消费的时候，刚消费到，还没有处理，结果进程挂了，比如重启
    • 用rabbitmq提供的ack机制。关闭自动ack,进行手动ack.

实际项目中遇到的问题：

• 之前遇到过这样的一个问题， 由于发送方与消费方的速度不匹配，在压测时，导致大量数据在队列中积压，大约有10万条，因为是测试数据，当时采用的方式比较暴力，直接删除队列，在新建队列，完成消息的删除。但是在生产上是不能这么干滴！！！

• 大量数据持续挤压怎么解决？

  • 先修复consumer的问题
  • 新建几个队列，具体数量自己把控。
  • 新建一个程序，不做耗时处理，将现在积压的数据分发到新建的队列中。
  • 每个队列绑定一个consumer,进行业务处理。
  • 处理完毕后，恢复之前的程序部署

消费方的业务处理时间可能比较长，最好采用多线程和多consumer进行消费。

程序实现

发送方

• 配置文件：
  关于rabbitmq的主机，端口等我直接写死在程序中，可通过properties文件注入，方便配置，基于rabbitTemplate进行代码封装

  
  配置.png

• 初始化

  
  初始化.png

此处说下callback和returncallbakc的执行时机：
消息没有到达exchange,则confirm回调，ack=false
消息到达exchange,则confirm回调，ack=false
exchange到queue成功，则不回调returnCallBack。
exchange到queue失败，则回调returnCallBack,在回调confirmCallBack.需设置mandatory=true,否则不回回调,消息就丢了

• 消息发送

  
  消息发送.png

上面为封装后对外开放的接口，可为此方法编写重载方法。对外只暴露 泛型参数Ｔ

• 消费方：两种实现方式

  • 使用注解：@RabbitListener(queues = "notify.payment")

    
    手动ack.png

  • 实现ChannelAwareMessageListener，并在container中注入

    
    image.png
    image.png

消费方推荐使用多线程进行处理。