一线大厂的MQ组件实现思路和架构设计方案

MQ组件需要实现功能点

• 支持消息高性能序列化转换、异步化发送消息
• 支持消息生产实例与消费实例的链接池化、缓存化，提升性能
• 支持可靠性投递消息，保障消息的100%不丢失
• 支持消费端的幂等操作，避免消费端重复消费的问题
• 支持迅速消息发送模式，在一些日志收集、统计分析等需求下可以保证高性能，超高吞吐量(可忽略100%投递）
• 支持延迟消息模式，消息可以延迟发送，指定延迟时间，用于某些延迟检查、服务限流场景
• 支持事务消息，且100%保障可靠性投递，在金融行业单笔大金额操作是会有此类需求
• 支持顺序消息，保证消费送达消费端的前后顺序，例如下订单，再送积分、优惠券等复合性操作
• 支持消息补偿，重试，以及快速定位异常/失败消息
• 支持集群消息负载均衡，保障消息落到具体SET集群的负责均衡
• 支持消息路由策略，指定某些消息路由到指定的SET集群

消息发送模式 - 迅速消息发送

• 迅速消息是指消息不进行落库存储，不做可靠性的保障
• 在一些非核心消息、日志数据、或者统计分析等场景下比较合适

• 迅速消息的特点就是性能最高，吞吐量最大

  
  

消息发送模式 - 确认消息发送

消息发送模式 - 批量消息发送

• 批量消息是指我们把消息放到一个集合里统一进行提交，这种方案设计思路是期望消息在一个会话里，比如投掷到threadlocal里的集合，然后拥有相同会话ID，并且带有这次提交消息的size等相关属性，最重要的一点是要把这一批消息进行合并。对应Channel而言，就是发送一次消息。这种方式希望消费端在消费的时候，可以进行批量化的消费，针对于某一个原子业务的操作去处理，但是不保障可靠性，需要进行补偿机制

  
  

消息发送模式 - 延迟消息发送

• 延迟消息相对简单，就是我们在Message封装的时候添加delayTime属性即可，使得我们的消息可以进行延迟发送，根据具体的业务场景也可以很好的使用得到。
• 场景举例：
  1. 比如你在电商平台买到商品签收后，不点击确定支付，那么系统自动会在7天（一定时间）去进行支付操作。
  2. 还要一些自动超时作废场景，你的优惠券/红包有使用时间限制，也可以用延迟消息机制。

消息发送模式 - 顺序消息发送

• 顺序消息，比较类似于批量消息的实现机制，但是有些不同。

• 我们要保障以下几点：

  1. 发送的顺序消息，必须保障消息投递到同一个队列，且这个消费者只能有一个（独占模式）

  2. 然后需要统一提交（可能合并成一个大消息，也可能拆分成多个消息），并且所有消息的会话ID一致。

  3. 添加消息属性：顺序标记的序号、和本次顺序消息的SIZE属性，进行落库操作。

  4. 并行进行发送给自身的延迟消息（注意带上关键属性：会话ID、SIZE)进行后续处理消费（接收到第一个消息后，创建延迟消息队列，等几分钟后，所有的顺序消息都落库后）

  5. 当收到延迟消息后，根据会话ID、SIZE抽取数据库数据进行处理即可。

  6. 定时轮询补偿机制，对于异常情况

  备注：比如生产端消息没有完全投递成功，或者消费端落库异常导致消费端落库缺少消息条目的情况

消息发送模式 - 事务消息发送

• 事务消息，相对使用比较少见，但是本身在早期做互联网行业中，面对单笔大额现金流交易时遇到遇到过：比如单笔转账超过一个上限的时候，我们就希望这个消息优先级最高，并且可靠性要求达到100%，当然我们的系统和银行端系统都要兼顾才行，所有也会有一些补偿机制，主动发起银行端查询指令机制等。

• 为了保障性能的同时，也支持事务。我们并没有选择传统的RabbitMQ事务和Spring集成的机制，因为在性能测试过程中，效果并不理想，非常消耗系统资源且会出现阻塞等情况，在高峰期也是一定程度上影响MQ集群的性能

• 解决方案：
  我们采用类似可靠性投递的机制，也就是补偿机制。
  但是我们的数据源必须是同一个，也就是业务操作DB1数据和消息记录BD2数据库必须使用同一个数据源
  然后利用重写Spring DatasourceTransactionManage，在本地事务提交的时候进行发送消息，但是也有可能事务提交成功但是消息发送失败，这个时候就需要进行补偿了。

  • DatasourceTransactionManager核心代码：

    
    

消息幂等性的重要性

• 保障消息的幂等性，这也是我们在使用MQ中至关重要的环节
• 可能导致消息出现非幂等性的原因：
  1. 可靠性消息投递机制（消息失败时多次发送）
  2. MQ Broker服务与消费端传输消息过程中出现网络抖动

  3. 消费端的故障或异常
     以上这些问题都会导致消费端重复消息问题。