---
date: 2021-01-24
---

<!-- TOC -->

- [消息中间件的应用场景](#消息中间件的应用场景)
- [主流 MQ 框架及对比](#主流-mq-框架及对比)
  - [说明](#说明)
  - [Kafka 优点](#kafka-优点)
  - [Kafka 缺点](#kafka-缺点)
  - [RocketMQ](#rocketmq)
  - [Pulsar](#pulsar)
  - [发展趋势](#发展趋势)
- [各公司发展](#各公司发展)
- [Kafka](#kafka)
  - [Kafka 是什么？](#kafka-是什么)
  - [Kafka 术语](#kafka-术语)
  - [Kafka 如何持久化？](#kafka-如何持久化)
  - [Kafka 文件存储机制](#kafka-文件存储机制)
  - [分区](#分区)
    - [为什么分区？](#为什么分区)
    - [分区策略？](#分区策略)
  - [Kafka 是否会消息丢失？](#kafka-是否会消息丢失)
  - [控制器](#控制器)
    - [控制器如何选举？](#控制器如何选举)
    - [控制器有什么用？](#控制器有什么用)
    - [控制器故障转移](#控制器故障转移)
  - [Kafka 的 ZooKeeper 存储结构](#kafka-的-zookeeper-存储结构)
- [分布式事务的应用场景](#分布式事务的应用场景)
  - [两阶段最终一致](#两阶段最终一致)
  - [如何保证最终一致？](#如何保证最终一致)
  - [消息发送的一致性如何保证？](#消息发送的一致性如何保证)
  - [发送异常会如何？](#发送异常会如何)

<!-- /TOC -->

![](http://yano.oss-cn-beijing.aliyuncs.com/2021-01-24-092401.png)

# 消息中间件的应用场景

- 异步解耦
- 削峰填谷
- 顺序收发
- 分布式事务一致性

腾讯应用案例：

![](http://yano.oss-cn-beijing.aliyuncs.com/2021-01-24-093404.png)

# 主流 MQ 框架及对比

![](http://yano.oss-cn-beijing.aliyuncs.com/2021-01-24-092609.png)

## 说明
* Kafka：整个行业应用广泛
* RocketMQ：阿里，从 apache 孵化
* Pulsar：雅虎开源，符合云原生架构的消息队列，社区活跃
* RabbitMQ 架构比较老，AMQP并没有在主流的 MQ 得到支持
* NSQ：内存型，不是最优选择
* ActiveMQ、ZeroMQ 可忽略


## Kafka 优点
* 非常成熟，生态丰富，与 Hadoop 连接紧密
* 吞吐非常高，可用性高
    * sharding
    * 提升 replication 速度
* 主要功能：pub-sub，压缩支持良好
* 可按照 at least once, at most once 进行配置使用，exactly once 需要 Consumer 配合
* 集群部署简单，但 controller 逻辑很复杂，实现partition 的多副本、数据一致性
* controller 依赖 ZooKeeper
* 异步刷磁盘（除了钱的业务，很少有同步 flush 的需求）

## Kafka 缺点
* 写入延时稳定性问题，partition 很多时
    * Kafka 通常用机械盘，随机写造成吞吐下降和延时上升
    * 100ms ~ 500ms
* 运维的复杂性
    * 单机故障后补充副本
    * 数据迁移
    * 快手的优化：迁移 partition 时旧数据不动，新数据写入新 partition 一定时间后直接切换

## RocketMQ
* 阿里根据 Kafka 改造适应电商等在线业务场景
* 以牺牲性能为代价增强功能
    * 按 key 对消息查询，维护 hash 表，影响 io
    * 为了在多 shard 场景下保证写入延迟稳定，在 broker 级别将所有 shard 当前写入的数据放入一个文件，形成 commitlog list，放若干个 index 文件维护逻辑 topic 信息，造成更多的随机读
* 没有中心管理节点，现在看起来并没有什么用，元数据并不多
* 高精度的延迟消息（快手已支持秒级精度的延迟消息）

## Pulsar
* 存储、计算分离，方便扩容
    * 存储：bookkeeper
    * MQ逻辑：无状态的 broker 处理


## 发展趋势
* 云原生
* 批流一体：跑任务时，需要先把 Kafka 数据→HDFS，资源消耗大。如果本来就存在 HDFS，能节省很大资源
* Serverless

# 各公司发展
* 快手：Kafka
    * 所有场景均在使用
    * 特殊形态的读写分离
        * 数据实时消费到 HDFS
        * 在有明显 lag 的 consumer 读取时，broker 把请求从本地磁盘转发的 HDFS
        * 不会因为有 lag 的 consumer 对日常读写造成明显的磁盘随机读写
    * 由于自己改造，社区新功能引入困难
* 阿里巴巴：开源 RocketMQ
* 字节跳动
    * 在线场景：NSQ→RocketMQ
    * 离线场景：Kafka→自研的存储计算分类的 BMQ（协议层直接兼容Kafka，用户可以不换 client）
* 百度：自研的 BigPipe，不怎么样
* 美团：Kafka 架构基础上用 Java 进行重构，内部叫 Mafka
* 腾讯：部分使用了自研的 PhxQueue，底层是 KV 系统
* 滴滴：DDMQ
    * 对 RocketMQ 和 Kafka 进行封装
    * 多机房数据一致性可能有问题
* 小米：自研 Talos
    * 架构类似 pulsar，存储是 HDFS，读场景有优化

# Kafka

- Kafka官网：https://kafka.apache.org/documentation/#uses
- 最新版本：2.7

## Kafka 是什么？
* 开源的消息引擎系统（消息队列/消息中间件）
* 分布式流处理平台
* 发布/订阅模型
* 削峰填谷

## Kafka 术语
* Topic：发布订阅的主题
* Producer：向Topic发布消息的客户端
* Consumer：消费者
* Consumer Group：消费者组，多个消费者共同组成一个组
* Broker：Kafka的服务进程
* Replication：备份，相同数据拷贝到多台机器
    * Leader Replica
    * Follower Replica，不与外界交互
* Partition：分区，解决伸缩性问题，多个Partition组成一个Topic
* Segment：partition 由多个 segment 组成

## Kafka 如何持久化？
* 消息日志（Log）保存数据，磁盘追加写（Append-only）
    * 避免缓慢的随机I/O操作
    * 高吞吐
* 定期删除消息（日志段）

![](http://yano.oss-cn-beijing.aliyuncs.com/2021-01-24-093304.png)

## Kafka 文件存储机制
https://www.open-open.com/lib/view/open1421150566328.html

* 每个 partition 相当于一个巨型文件→多个大小相等 segment 数据文件中
* 每个 partition 只需要顺序读写就行了，segment 文件生命周期由配置决定
* segment file 组成：
    * index file：索引文件
    * data file：数据文件
* segment file 文件命名规则：
    * 全局第一个 segment 是 0
    * 后序每个加上全局 partition 的最大 offset

![](http://yano.oss-cn-beijing.aliyuncs.com/2021-01-24-092946.png)

![](http://yano.oss-cn-beijing.aliyuncs.com/2021-01-24-093010.png)

一对 segment file

![](http://yano.oss-cn-beijing.aliyuncs.com/2021-01-24-093027.png)

message 物理结构

![](http://yano.oss-cn-beijing.aliyuncs.com/2021-01-24-093043.png)

## 分区

### 为什么分区？
* Kafka的消息组织方式：主题-分区-消息
* 一条消息，仅存在某一个分区中
* 提高伸缩性，不同分区可以放到不同机器，读写操作也是以分区粒度

### 分区策略？
* 轮询
* 随机
* 按 key 保序，单分区有序

![](http://yano.oss-cn-beijing.aliyuncs.com/2021-01-24-093103.png)

## Kafka 是否会消息丢失？
* 只对“已提交”的消息做有限度的持久化保证
    * 已提交的消息：消息写入日志文件
    * 有限度的持久化保证：N个 broker 至少一个存活
* 生产者丢失数据
    * producer.send(msg) 异步发送消息，不保证数据到达Kafka
    * producer.send(msg, callback) 判断回调
* 消费者程序丢失数据
    * 应该「先消费消息，后更新位移的顺序」
    * 新问题：消息的重复处理
    * 多线程异步处理消息，Consumer不要开启自动提交位移，应用程序手动提交位移

## 控制器
* 在 ZooKeeper帮助下管理和协调整个 Kafka 集群
* 运行过程中，只能有一个 Broker 成为控制器

### 控制器如何选举？
在 ZooKeeper 创建 /controller 节点，第一个创建成功的 Broker 被指定为控制器。

### 控制器有什么用？
* 主题管理（创建、删除、增加分区）
* 分区重分配
* 领导者选举
* 集群成员管理（新增 Broker、Broker 主动关闭、Broker 宕机）（ZooKeeper 临时节点）
* 数据服务：最全的集群元数据信息

### 控制器故障转移
* 只有一个 Broker 当控制器，单点失效，立即启用备用控制器

![](http://yano.oss-cn-beijing.aliyuncs.com/2021-01-24-093221.png)

## Kafka 的 ZooKeeper 存储结构

![](http://yano.oss-cn-beijing.aliyuncs.com/2021-01-24-093954.png)

# 分布式事务的应用场景
* 团队内部，某些操作要同时更新多个数据源
* 业务团队 A 完成某个操作后，B 业务的某个操作也必须完成，A 业务并不能直接访问 B 的数据库
* 公司之间，用户付款后，支付系统（支付宝/微信）必须通知商家的系统更新订单状态

## 两阶段最终一致
* 先完成数据源 A 的事务（一阶段）
* 成功后通过某种机制，保证数据源 B 的事务（二阶段）也一定最终完成
    * 不成功，会不断重试直到成功为止
    * 或达到一定重试次数后停止（配合对账、人工处理）

## 如何保证最终一致？
为了保证最终一致，消息系统和业务程序需要保证：
* 消息发送的一致性：消息发送时，一阶段事务和消息发送必须同时成功或失败
* 消息存储不丢失：消息发送成功后，到消息被成功消费前，消息服务器（broker）必须存储好消息，保证发生故障时，消息不丢失
* 消费者不丢失消息：处理失败不丢弃，重试直到成功为止

## 消息发送的一致性如何保证？

![](http://yano.oss-cn-beijing.aliyuncs.com/2021-01-24-093814.png)

`目标`：本地事务、消息发送必须同时成功/失败

`问题`
* 先执行本地事务，再发送消息，消息可能发送失败
* 可把失败的消息放入内存，稍后重试，但成功率也无法达到 100%

解决方案`* 先发送半消息（Half Msg，类似 Prepare 操作），不会投递给消费者
* 半消息发送成功，再执行 DB 操作
* DB 操作执行成功后，提交半消息

## 发送异常会如何？
* 1 异常，半消息发送失败，本地 DB 没有执行，整个操作失败，DB/消息的状态一致（都没有提交）
* 2 异常/超时
    * 生产者以为失败了，不执行 DB
    * broker 存储半消息成功，等不到后序操作，会询问生产者是提交还是回滚（第6步）
* 3 DB操作失败：生产者在第 4 步告知 broker 回滚半消息
* 4 提交/回滚半消息失败：broker 等不到这个操作，触发回查（第 6 步）
* 5、6、7回查失败：RocketMQ 最多回查 15 次