RocketMQ高级特性

  |   0 评论   |   0 浏览

RocketMQ高级特性

1.消息的存储

image-20220912084350071

  1. 消息生成者发送消息到MQ
  2. MQ返回ACK给生产者
  3. MQ push 消息给对应的消费者
  4. 消息消费者返回ACK给MQ
    说明：ACK（Acknowledge character）

image-20220912084509937

  1. 消息生成者发送消息到MQ
  2. MQ收到消息，将消息进行持久化，存储该消息
  3. MQ返回ACK给生产者
  4. MQ push 消息给对应的消费者
  5. 消息消费者返回ACK给MQ

​ 6. MQ删除消息

注意：

  1. 第⑤步MQ在指定时间内接到消息消费者返回ACK，MQ认定消息消费成功，执行⑥
  2. 第⑤步MQ在指定时间内未接到消息消费者返回ACK，MQ认定消息消费失败，重新执行④⑤⑥

2.消息的存储介质

数据库

​ 1.ActiveMQ

​ 2.缺点：数据库瓶颈将成为MQ瓶颈

文件系统

​ 1.RocketMQ/Kafka/RabbitMQ

​ 2.解决方案：采用消息刷盘机制进行数据存储

​ 3.缺点：硬盘损坏的问题无法避免

image-20220912084816047

3.高效的消息存储与读写方式

  1. SSD（Solid State Disk）

image-20220912084852802

  1. 随机写（100KB/s）

image-20220912084905777

  1. 顺序写 （600MB/s）1秒1部电影

image-20220912084941613

image-20220912084950165

  1. Linux系统发送数据的方式

  2. “零拷贝”技术

    ​ 1.数据传输由传统的4次复制简化成3次复制，减少1次复制过程

    ​ 2.Java语言中使用MappedByteBuffer类实现了该技术

    ​ 3.要求：预留存储空间，用于保存数据（1G存储空间起步）

image-20220912085135913

4.消息存储结构

  1. MQ数据存储区域包含如下内容

    1. 消息数据存储区域
      1. topic
        1. queueId
        2. message

  2. 消费逻辑队列

    1. minOffset
    2. maxOffset
    3. consumerOffset

  3. 索引

    1. key索引
    2. 创建时间索引

image-20220912085337793

5.刷盘机制

  1. 同步刷盘
    1. 生产者发送消息到MQ，MQ接到消息数据
    2. MQ挂起生产者发送消息的线程
    3. MQ将消息数据写入内存
    4. 内存数据写入硬盘
    5. 磁盘存储后返回SUCCESS
    6. MQ恢复挂起的生产者线程
    7. 发送ACK到生产者

image-20220912085420448

  1. 异步刷盘

    1. 生产者发送消息到MQ，MQ接到消息数据
    2. MQ将消息数据写入内存
    3. 发送ACK到生产者

  2. 同步刷盘：安全性高，效率低，速度慢（适用于对数据安全要求较高的业务）

  3. 异步刷盘：安全性低，效率高，速度快（适用于对数据处理速度要求较高的业务）

配置方式

在rocketmq配置文件中进行配置

#刷盘方式
#- ASYNC_FLUSH 异步刷盘
#- SYNC_FLUSH 同步刷盘 
flushDiskType=SYNC_FLUSH

6.高可用性

  1. nameserver
    1. 无状态+全服务器注册
  2. 消息服务器
    1. 主从架构（2M-2S）
  3. 消息生产
    1. 生产者将相同的topic绑定到多个group组，保障master挂掉后，其他master仍可正常进行消息接收
  4. 消息消费
    1. RocketMQ自身会根据master的压力确认是否由master承担消息读取的功能，当master繁忙时候，自动切换由slave承担数据读取的工作

7.主从数据复制

  1. 同步复制

    1. master接到消息后，先复制到slave，然后反馈给生产者写操作成功
    2. 优点：数据安全，不丢数据，出现故障容易恢复
    3. 缺点：影响数据吞吐量，整体性能低

  2. 异步复制

    1. master接到消息后，立即返回给生产者写操作成功，当消息达到一定量后再异步复制到slave

    2. 优点：数据吞吐量大，操作延迟低，性能高

    3. 缺点：数据不安全，会出现数据丢失的现象，一旦master出现故障，从上次数据同步到故障

      时间的数据将丢失

  3. 配置方式

#Broker 的角色 
#- ASYNC_MASTER 异步复制Master 
#- SYNC_MASTER 同步双写Master
#- SLAVE 
brokerRole=SYNC_MASTER

8.负载均衡

  1. Producer负载均衡

    1. 内部实现了不同broker集群中对同一topic对应消息队列的负载均衡
    2. Consumer负载均衡
      1. 平均分配
      2. 循环平均分配
    3. 广播模式（不参与负载均衡）

    image-20220912090127860

image-20220912090143337

image-20220912090148859

9.消息重试

  1. 当消息消费后未正常返回消费成功的信息将启动消息重试机制
  2. 消息重试机制
    1. 顺序消息
    2. 无序消息

​

1.顺序消息重试

image-20220912090229072

image-20220912090234479

  1. 当消费者消费消息失败后，RocketMQ会自动进行消息重试（每次间隔时间为 1 秒）
  2. 注意：应用会出现消息消费被阻塞的情况，因此，要对顺序消息的消费情况进行监控，避免阻塞现
    象的发生

image-20220912090253119

image-20220912090258381

2.无序消息重试

  1. 无序消息包括普通消息、定时消息、延时消息、事务消息
  2. 无序消息重试仅适用于负载均衡（集群）模型下的消息消费，不适用于广播模式下的消息消费
  3. 为保障无序消息的消费，MQ设定了合理的消息重试间隔时长

image-20220912090346255

10.死信队列

  1. 当消息消费重试到达了指定次数（默认16次）后，MQ将无法被正常消费的消息称为死信消息
    （Dead-Letter Message）
  2. 死信消息不会被直接抛弃，而是保存到了一个全新的队列中，该队列称为死信队列（Dead-Letter
    Queue）
  3. 死信队列特征
    1. 归属某一个组（Gourp Id），而不归属Topic，也不归属消费者
    2. 一个死信队列中可以包含同一个组下的多个Topic中的死信消息
    3. 死信队列不会进行默认初始化，当第一个死信出现后，此队列首次初始化
  4. 死信队列中消息特征
    1. 不会被再次重复消费
    2. 死信队列中的消息有效期为3天，达到时限后将被清除

死信处理

  1. 在监控平台中，通过查找死信，获取死信的messageId，然后通过id对死信进行精准消费

11.消息重复消费

  1. 消息重复消费原因

    1. 网络闪断
    2. 生产者宕机

  2. 消息服务器投递了重复的消息

    1. 网络闪断

  3. 动态的负载均衡过程

    1. 网络闪断/抖动
    2. broker重启
    3. 订阅方应用重启（消费者）
    4. 客户端扩容
    5. 客户端缩容

    image-20220912090657766

1.如何处理重复消费

消息队列出现重复消费的情况，在日常开发不可避免会遇到，而出现这个问题应该怎么去解决呢？这就涉及到消息幂等性的概念了

消息幂等性：对同一条消息，无论消费多少次，结果保持一致，称为消息幂等性

注意：messageId由RocketMQ产生，messageId并不具有唯一性，不能作用幂等判定条件

解决方案

1.设置唯一的幂等令牌（业务流水号），对处理成功的流水号进行缓存，在下一次消费时看看缓存是否命中，如果没有则进入下一步，如果命中则时重复消费

2.缓存失效的情况，先在数据库中查询幂等令牌作为索引的数据是否存在，如果存在则说明是重复性操作，如果不存在则进入下一步

3.在同一事务（分布式事务，rocketmq支持事务消息）中完成三项操作：唯一性处理后，将幂等令牌写入到缓存，并将令牌作为唯一索引写入到DB中

标题：RocketMQ高级特性
作者：llp
地址：https://llinp.cn/articles/2022/09/12/1662945577593.html