ThinPanda

消息中间件 RocketMQ
消息中间件功能应用解耦原来需要同步的两个应用/模块,使用消息队列后可以异步执行流量削峰流量达到高峰时,通常使用限流...
扫描右侧二维码阅读全文
04
2019/12

消息中间件 RocketMQ

消息中间件功能

应用解耦
  • 原来需要同步的两个应用/模块,使用消息队列后可以异步执行
流量削峰
  • 流量达到高峰时,通常使用限流算法来控制流量涌入系统,避免系统被击瘫,但是这种方式损失了一部分请求。
  • 使用消息队列可以缓冲大量请求,匀速消费。当消息堆积过多时,可以动态上线增加消费端,保证重要请求不会丢失
大数据处理
  • 消息中间件可以将日志、用户行为等数据文件作为消息收集到主题中,数据使用方可以订阅自己需要的数据,互不影响
异构系统
  • 跨语言

RocketMQ角色

nameserver

  • 类似于注册中心,提供了路由管理、服务注册、服务发现的功能,集群中其他角色需要定时向nameserver上报自己的状态,以便互相发现彼此,超时不上报会被nameserver从列表中剔除
  • nameserver为了避免单点故障可以部署多个,保证高可用,但集群之间互不通信,没有主备的概念

    • nameserver是无状态的,使用内存存储broke、topic等信息,默认不会持久化

broker

  • 向每个nameserver注册自己的信息,负责存储和转发消息
  • broker集群

    • 高可用,master/slave机构,master可读可写,slave只能读,master将数据同步给slave
    • 指定相同的BrockerName,brokerId为0表示master,非0表示slave
  • topic

    • 对消息进行分类的一个逻辑概念
  • queue

    • 实际存放消息的物理单位,一个topic有多个queue
  • tag

    • 相比于topic一种更细粒度的划分
  • groupname

    • 标识一个生产组或一个消费组
  • offset

    • 消息实际被消费到的位置,queue数组的下标

producer

  • 消息生产者
  • 与其中一个nameserver节点建立长连接,获得topic路由,包括topic下面有几个queue,这些queue分布在哪些broker上
  • 与提供topic服务对应的master建立长连接,且定时发送心跳包

consumer

  • 通过nameserver获得topic路由信息,连接到对应的broker上消费
  • 与提供topic服务的master和slave都建立连接

RocketMQ消费模式

集群消费

  • 每个消息只被处理一次,只发个一个group中的一个consume
  • 每个订阅的group都会消费到消息
  • 消费状态由broker维护
  • 在消息重投时不能保证被路由到同一台机器上

广播消费

  • 每个订阅的group中的consumer都会消费一次
  • 消费进度由consumer维护
  • 消费失败不会重投

RocketMQ的分布式事务支持

  • 使用2PC方式实现
  • Half Message预处理消息,进行预提交
  • TransactionListener最终提交

    • executeLocalTransaction半消息发送成功用该方法执行本地事务
    • checkLocalTransaction检查本地事务状态

      • LocalTransactionState.COMMIT_MESSAGE 确认提交
      • LocalTransactionState.ROLLBACK_MESSAGE 回滚
      • LocalTransactionState.UNKNOWN 等待broker回查

RocketMQ消息获得方式

pull拉

推荐,由客户端决定何时消费消息

push推

有可能造成consumer端消息堆积过多又不能被其他consumer消费

RocketMQ消息消费问题

消息被消费后会被立即删除吗?

  • 不会,每条消息都被持久化到commitLog中
  • 每个consumer维持消费进度,消费消息后只是当前的consumer消费进度(commitLog中的offset)更新了
  • 48小时候会删除不再使用的commitLog文件

消息丢失

  • producer在发送消息后会收到sendResult,表示消息发送成功
  • 当消息发送失败的时候,重新发送,避免丢失(重试16次之后进死信队列)
  • 整个流程使用单机事务保证可靠性

消息重复消费

原因
  • 网络原因

    • consumer首次启动
    • broker没有收到ack导致消息重投
    • 消息的幂等性
  • 集群模式下消息消费以group为单位,每个group都会消费
解决方案(保证消息幂等性)
  • 数据库本地事务表,使用transactionID作为主键插入,重复消费时会插入失败,单机事务保证可靠
  • 单机时用ConcurrentHashMap的putIfAbsent
  • Redis setnx主键

如何保证消息顺序消费

使用单线程处理
  • 同一个topic、同一个queue发消息的时候一个线程去发送消息、消费的时候一个线程消费一个queue里的消息
  • 如果多个queue要求有序,设置最大消费线程数和最小消费线程数都为1

如何保证消息不丢失

检测消息丢失

利用消息队列的有序性,在producer端给每个发出的消息附加一个连续递增的序号,在consumer端检查序号的连续性

确保消息可靠传递
  • 生产阶段 | 捕获异常进行重发

    • 同步消费捕获异常进行重发
    • 异步消费在回调方法检查消费状态
  • 存储阶段 | 配置刷盘为同步刷盘(而不是异步)

    • 配置broker参数flushDiskType刷盘方式为SYNC_FLUSH同步刷盘
  • 消费阶段 | 处理完所有消费逻辑后再确认

    • 在执行完所有的消费逻辑之后再发送消费确认

消息堆积如何处理

消息堆积时间过长会超时吗?

RocketMQ中的消息只会在commitLog被删除的时候才会消失,不会超时

解决办法
  • 控制生产者的速率
  • 简单来说上线更多consumer临时解决消息堆积问题
  • 如果堆积了海量数据

    • 注意:一定不能动生产方
    • 准备一个临时的topic,并设置queue的数量为原先好几倍,分部到多个broker中
    • 上线一个consumer将原来topic中的消息挪到新的topic里,不做业务处理
    • 上线多台consumer同时消费临时topic中的数据
    • 查找原先代码的bug
    • 消费完积压数据后,恢复原来的架构部署,使用原来的consumer消费

零拷贝技术

使用NIO的MappedByteBuffer

MQ选型

RabbitMQ

erlang开发,低延迟

RocketMQ

Java开发,面向互联网集群化功能丰富

Kafka

Scala开发,面向日志

ActiveMQ

Java开发,简单稳定,社区减少维护

Last modification:September 3rd, 2020 at 11:00 am
如果觉得我的文章对你有用,请随意赞赏

Leave a Comment