RabbitMQ
RabbitMQ是一款开源的，使用Erlang语言编写的，基于AMQP协议的消息中间件。
基本概念
特点
- AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全
- AMQP协议适用于对数据一致性、稳定性和可靠性要求很高的场景，对性能吞吐量的要求其次的（可以根据场景选择kafka或RocketMQ）
常用组件
- Connection 连接，应用程序与Server的网络连接，TCP协议
- Channel 信道，消息读写等操作在信道中进行
- Message 消息，应用程序和服务器之间传输的数据
- Virtual Host 虚拟主机，用于逻辑隔离，一个虚拟主机里面可以有若干个Exchange和Queue
- Exchange 交换器，接收消息，按照路由器规则将消息路由到一个或多个队列
- Binding 绑定，交换器和消息队列之间的虚拟连接，绑定中可以包含一个或多个RoutingKey
- RoutingKey 路由键，生产者将消息发送给你交换器时会发一个RoutingKey，用来指定路由规则
- Queue 消息队列，用来保存消息，供消费者消费
- Broker 标识消息队列服务器实体
交换器类型
- Direct Exchange 完全匹配，消息中的路由键（routing key）和Binding中的binding key 一致
- Topic Exchange 模糊匹配，两个通配符：“#”和“*”，#匹配0个或多个单词，*匹配一个单词
- Fanout Exchange 广播模式，不处理路由键，把所有发送到路由器的消息路由到所有绑定的队列中
- Headers Exchange 忽略路由规则，根据消息中的headers属性来匹配，性能低（不常用）
MQ使用场景及优缺点
优点
- 解耦（最终一致性）用户下单后，订单系统需要通知库存系统。传统的做法是，订单系统调用库存系统的接口，假如库存系统无法访问，则订单减库存将失败，从而导致订单失败；引入MQ用户下单后，订单系统完成持久化处理，将消息写入消息队列，返回用户订单下单成功，库存系统订阅下单消息，进行增减库存操作，即使库存系统崩溃了，也不影响正常下单，实现接耦
- 异步（提升效率） 比如用户注册，需要发注册邮件和注册短信。传统的做法有两种 1.串行的方式；2.并行方式 比如发送邮件50ms，发送短信50ms，串行调用的话是耗费时间总和。改用MQ同时异步发送短信和邮件，大大缩短了整体的响应时间，提升效率。
- 削峰（量力而行）比如秒杀场景，高峰期时每秒有5000个请求打到MySQL上，而MySQL最多每秒处理2000个请求，过多的请求会导致MySQL瘫痪。引入MQ将高峰期的请求直接写入MQ，MySQL根据自己的处理能力去MQ拉去请求进行处理，缓解MySQL压力，虽然期间会有短暂的消息积压，不过高峰期过后消息就会被快速消费掉。
缺点
系统的可用性降低
引入了外部依赖越多，系统越容易挂掉；MQ挂了也会导致整个系统不可用
系统的复杂性提高
如何保证消息没有重复消费？幂等性
使用全局唯一ID+指纹码（全局唯一ID：雪花算法生成的业务表的主键。指纹码：时间戳，UUID，订单号）；并发量不高的情况下可以用数据库维护一张消费记录表，并发量很高可以将全局ID写入到redis，利用其原子性setnx；接收到消息后先执行setnx，如果成功则表示没有处理过，可以消费，相反如果执行失败表示该消息已经被消费
- 如何保证消息传到的顺序？顺序性
在MQ里面创建多个queue，使用hash算法将需要排序的数据有序的放入到同一个queue，每个queue对应一个consumer或者就一个queue，对应一个consumer，这个consumer内部用内存队列排序，然后分发给不同的worker来处理
如何保证消息不丢失？可靠性
消息丢失场景
- 生产者发送消息到MQ
1、网络原因；
2、代码/配置 事例1:
一般情况下，生产者使用Confirm模式投递消息，如果方案不够严谨，比如RabbitMQ Server 接收消息失败后会发送nack消息通知生产者，生产者监听消息失败或者没做任何事情，消息存在丢失风险；
事例2:
生产者发送消息到exchange后，发送的路由和queue没有绑定，消息会存在丢失情况，下面会讲到具体的例子，保证意外情况的发生，即使发生，也在可控范围内 - MQ中存储的消息丢失：消息未完全持久化
- 消费者从MQ拉取消息：消费端收到相关消息后，消费端还没来得急处理消息，消费端机器就宕机了
如何避免消息丢失
生产者丢消息
- 事物机制（基于AMQP协议）：吞吐量下降（同步），一般不推荐；发送消息前，开启事务（channel.txSelect()），然后发送消息，如果发送过程中出现什么异常，事务就会回滚（channel.txRollback()），如果发送成功则提交事务（channel.txCommit()）。
- confirm机制（生产者确认机制）：异步回调，效率高；在生产者那里设置开启confirm模式后，每次写的消息都会分配一个唯一的ID，如果成功写入了rabbitmq中，rabbitmq会回调生产者ack接口，说明这个消息成功写入了。如果rabbitmq没能处理这个消息，会回调一个nack接口，说明这个消息接收失败，需要重发。而且还可以在内存里维护每个消息ID的状态，如果超过一定时间还没有接受到这个消息的回调，也可以重发。
MQ丢失消息
设置持久化有两个步骤：
第一个步骤是创建queue的时候将其设置为持久化，（将队列的持久化标识durable设置为true）这样就可以保证rabbitmq持久化queue的元数据，但是不会持久化queue里的消息；
第二个步骤是发送消息的时候将消息的deliveryMode设置为2，就是将消息设置为持久化。此时rabbitmq就会将消息持久化到磁盘了，注意：必须同时设置这两个才可以。这个持久化可以跟生产者那边的confirm机制配合起来，只有消息被持久化到磁盘之后，才会回调生产者的ack接口，所以即使是持久化到磁盘之前rabbitmq挂了，生产者收不到回调，也会重发。
- 使用镜像集群模式保证高可用：1、rabbitMQ友很好的管理控制台，在后台新增一个镜像集群模式的策略，指定同步节点的时候要求数据同步到所有节点（性能有极大影响）；镜像策略：指定最多同步N台机器、只同步到符合指定名称的机器，再次创建queue的时候应用这个策略就会自动将数据同步到其他节点上
消费者丢消息
关闭消费者的自动ack机制，采用手动ack形式，消费者处理完消息后手动ack通知MQ删除消息
数据一致性问题
A系统发送完消息后直接返回成功，但是后续消费系统之间中若有系统写库失败，则会产生数据不一致的问题
常见问题
死信队列（DLX）
消息变成死信队列的原因
- 消息被拒绝（basic.reject/basic.nack），并且不再重新投递requeue=faslse
- TTL（time-to-live）消息超时未消费
- 队列达到最长度
死信队列的设置
- 提前设置好死信队列的Exchange和queue然后进行绑定；
例如：
exchange: dlx.exchange
queue: dlx.queue
routingkey: # - 在普通队列上添加参数：argument.put(“x-dead-letter-exchange”,“dkx,exchange”)
- 这样消息在过期或队列达到最大长度时，消息就会直接到路由到死信队列
延迟队列怎么实现？
-
使用定时任务 非常浪费服务器性能，不建议 -
使用java自带的delayQueue(这种实现方法在内存中，可能面临数据丢失的情况，并且无法支持分布式系统，不能做集群化处理且不易维护) 不建议 -
使用RabbitMQ的消息过期时间（TTL）和死信队列来模拟延时队列
eg：
队列1:延迟队列，负载数据的暂时存储
队列2:真正监听的消息队列
创建交换机
当前队列的消息一旦过期，则进入到死信队列交换机 -
可以用RabbitMQ的插件rabbitmq-delayed-message-exchange来实现延时队列达到可投递时间时并将其通过x-delayed-type类型标记的交换机类型投递到目标队列
|