消息队列 kafka

消息队列 kafka

基本概念

Apache 基金会开源分布式消息系统，基于发布/订阅消息
Kafka 已被多家不同类型的公司作为多种类型的数据管道和消息系统使用。行为流数据是几乎所有站点在对其网站使用情况做报表时都要用到的数据中最常规的部分。

包括页面访问量 PV、页面曝光 Expose、页面点击 Click 等行为事件；
实时计算中的 Kafka Source，Dataflow Pipeline；
业务的消息系统，通过发布订阅消息解耦多组微服务，消除峰值；

Kafka 是由 LinkedIn 开发并开源的分布式消息系统，因其分布式及高吞吐率而被广泛使用，现已与 Cloudera Hadoop，Apache Storm，Apache Spark，Flink 集成。

Kafka 是一种分布式的，基于发布/订阅的消息系统。主要设计目标如下：

以时间复杂度为 O(1) 的方式提供消息持久化能力，即使对 TB 级以上数据也能保证常数时间复杂度的访问性能；
高吞吐率。即使在非常廉价的商用机器上也能做到单机支持每秒 100K 条以上消息的传输；
支持 Kafka Server 间的消息分区，及分布式消费，同时保证每个 Partition 内的消息顺序传输；
同时支持离线数据处理和实时数据处理；
Scale out：支持在线水平扩展；

为何使用消息系统

解耦
消息系统在处理过程中间插入了一个隐含的、基于数据的接口层，两边的处理过程都要实现这一接口。这允许你独立的扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束。
而基于消息发布订阅的机制，可以联动多个业务下游子系统，能够不侵入的情况下分步编排和开发，来保证数据一致性。
冗余
有些情况下，处理数据的过程会失败。除非数据被持久化，否则将造成丢失。消息队列把数据进行持久化直到它们已经被完全处理，通过这一方式规避了数据丢失风险。许多消息队列所采用的”插入-获取-删除”范式中，在把一个消息从队列中删除之前，需要你的处理系统明确的指出该消息已经被处理完毕，从而确保你的数据被安全的保存直到你使用完毕。
扩展性
因为消息队列解耦了你的处理过程，所以增大消息入队和处理的频率是很容易的，只要另外增加处理过程即可。不需要改变代码、不需要调节参数。扩展就像调大电力按钮一样简单。
灵活性和峰值处理
在访问量剧增的情况下，应用仍然需要继续发挥作用，但是这样的突发流量并不常见；如果为以能处理这类峰值访问为标准来投入资源随时待命无疑是巨大的浪费。使用消息队列能够使关键组件顶住突发的访问压力，而不会因为突发的超负荷的请求而完全崩溃。
可恢复性
系统的一部分组件失效时，不会影响到整个系统。消息队列降低了进程间的耦合度，所以即使一个处理消息的进程挂掉，加入队列中的消息仍然可以在系统恢复后被处理。
顺序保证
在大多使用场景下，数据处理的顺序都很重要。大部分消息队列本来就是排序的，并且能保证数据会按照特定的顺序来处理。Kafka 保证一个 Partition 内的消息的有序性。
缓冲
在任何重要的系统中，都会有需要不同的处理时间的元素。消息队列通过一个缓冲层来帮助任务最高效率的执行———写入队列的处理会尽可能的快速。该缓冲有助于控制和优化数据流经过系统的速度。
异步通讯
很多时候，用户不想也不需要立即处理消息。消息队列提供了异步处理机制，允许用户把一个消息放入队列，但并不立即处理它。想向队列中放入多少消息就放多少，然后在需要的时候再去处理它们。

Topic && Partition

Topic 可以理解为将消息分类，将一类的消息入到一个 queue 里面，顺序消息
Topic 在逻辑上可以被认为是一个 queue，每条消费都必须指定它的 Topic，可以简单理解为必须指明把这条消息放进哪个 queue 里。我们把一类消息按照主题来分类，有点类似于数据库中的表。
为了使得 Kafka 的吞吐率可以线性提高，物理上把 Topic 分成一个或多个 Partition。对应到系统上就是一个或若干个目录。

Broker: kafka 集群包含一个或多个服务器，每个服务器节点名称称为一个 broker
Broker 存储 Topic 的数据。如果某 Topic 有 N 个 Partition，集群有 N 个 Broker，那么每个 Broker 存储该 Topic 的一个 Partition。
从 scale out 的性能角度思考，通过 Broker Kafka server 的更多节点，带更多的存储，建立更多的 Partition 把 IO 负载到更多的物理节点，提高总吞吐 IOPS。
从 scale up 的角度思考，一个 Node 拥有越多的 Physical Disk，也可以负载更多的 Partition，提升总吞吐 IOPS。

如果某 Topic 有 N 个 Partition，集群有（N+M）个 Broker，那么其中有 N 个 Broker 存储该 Topic 的一个 Partition，剩下的 M 个 Broker 不存储该 Topic 的 Partition 数据。
如果某 Topic 有 N 个 Partition，集群中 Broker 数目少于 N 个，那么一个 Broker 存储该 Topic 的一个或多个 Partition。
Topic 只是一个逻辑概念，真正在 Broker 间分布式的 Partition。
每一条消息被发送到 Broker 中，会根据 Partition 规则选择被存储到哪一个 Partition。如果 Partition 规则设置的合理，所有消息可以均匀分布到不同的 Partition 中。

实验条件：3 个 Broker，1 个 Topic，无 Replication，异步模式，3 个 Producer，消息 Payload 为 100 字节：
当 Partition 数量小于 Broker 个数时，Partition 数量越大，吞吐率越高，且呈线性提升。
Kafka 会将所有 Partition 均匀分布到所有 Broker 上，所以当只有 2 个 Partition 时，会有 2 个 Broker 为该 Topic 服务。3 个 Partition 时同理会有 3 个 Broker 为该 Topic 服务。
当 Partition 数量多于 Broker 个数时，总吞吐量并未有所提升，甚至还有所下降。可能的原因是，当 Partition 数量为 4 和 5 时，不同 Broker 上的 Partition 数量不同，而 Producer 会将数据均匀发送到各 Partition 上，这就造成各 Broker 的负载不同，不能最大化集群吞吐量。

存储原理

Kafka 的消息是存在于文件系统之上的。Kafka 高度依赖文件系统来存储和缓存消息，一般的人认为 “磁盘是缓慢的”。
操作系统还会将主内存剩余的所有空闲内存空间都用作磁盘缓存，所有的磁盘读写操作都会经过统一的磁盘缓存（除了直接 I/O 会绕过磁盘缓存）。
Kafka 正是利用顺序 IO，以及 Page Cache 达成的超高吞吐。
任何发布到 Partition 的消息都会被追加到 Partition 数据文件的尾部，这样的顺序写磁盘操作让 Kafka 的效率非常高。

kafka 对消息的存储是根据消息过期策略来清理的，不管这个消息有没有被消息过。比如设置保留策略是两天内，kafka 的性能跟存储的数量的大小无关
Offset：偏移量，每条消息都有一个当前 partition 的偏移量，即第几条消息
消费者可以指定消费的消息位置，下次恢复后指定这个位置继续消费
存储文件解读：每个 partition 都是一个文件目录，每个文件目录又被平均分配成多个大小相等的 Segment File。每个 Segment file 又有.log 文件（数据文件）和.index(索引文件)他们总是成对出现，文件名有文件偏移决定。这样顺序命令索引文件，就可以查找的时候用于二分查找，提高查询效率
新版本增加一个.timeindex 文件，用于索引时间信息

存储原理

假设我们现在 Kafka 集群只有一个 Broker，我们创建 2 个 Topic 名称分别为：「Topic1」和「Topic2」，Partition 数量分别为 1、2。
那么我们的根目录下就会创建如下三个文件夹

topic1-0
topic2-0
topic2-1

在 Kafka 的文件存储中，同一个 Topic 下有多个不同的 Partition，每个 Partition 都为一个目录。
而每一个目录又被平均分配成多个大小相等的 Segment File 中，Segment File 又由 index file 和 data file 组成，他们总是成对出现，后缀 “.index” 和 “.log” 分表表示 Segment 索引文件和数据文件。

其中以索引文件中元数据 <3, 497> 为例，依次在数据文件中表示第 3 个 Message（在全局 Partition 表示第 368769 + 3 = 368772 个 message）以及该消息的物理偏移地址为 497。
注意该 Index 文件并不是从 0 开始，也不是每次递增 1 的，这是因为 Kafka 采取稀疏索引存储的方式，每隔一定字节的数据建立一条索引。
它减少了索引文件大小，使得能够把 Index 映射到内存，降低了查询时的磁盘 IO 开销，同时也并没有给查询带来太多的时间消耗。
因为其文件名为上一个 Segment 最后一条消息的 Offset ，所以当需要查找一个指定 Offset 的 Message 时，通过在所有 Segment 的文件名中进行二分查找就能找到它归属的 Segment。
再在其 Index 文件中找到其对应到文件上的物理位置，就能拿出该 Message。

Kafka 是如何准确的知道 Message 的偏移的呢？
这是因为在 Kafka 定义了标准的数据存储结构，在 Partition 中的每一条 Message 都包含了以下三个属性：
Offset：表示 Message 在当前 Partition 中的偏移量，是一个逻辑上的值，唯一确定了 Partition 中的一条 Message，可以简单的认为是一个 ID。
MessageSize：表示 Message 内容 Data 的大小。
Data：Message 的具体内容。

Producer

发布器发布消息到 Broker 会根据 partition 设置规则，落盘到具体到 partition 内。如果 partition 分区规则设置合理，消息就能均匀的分配到不同的 partition 内，这样就实现了负载均衡

指明 Partition 的情况下，直接将给定的 Value 作为 Partition 的值。
没有设置 partition 值，但有 key 的情况下，将 key 的 hash 值于分区数取余得到 partition 值。即根据 key 做 hash 打散。
既没有 partition 也没有 key。第一次调用随机生成一个整数，后面每次在这个整数上递增，然后用这个整数对 partition 数量取余得到 partition 值。即 Round-Robin 算法

为保证 Producer 发送的数据，能可靠地发送到指定的 Topic，Topic 的每个 Partition 收到 Producer 发送的数据后，都需要向 Producer 发送 ACK。如果 Producer 收到 ACK，就会进行下一轮的发送，否则重新发送数据。

选择完分区后，生产者知道了消息所属的主题和分区，它将这条记录添加到相同主题和分区的批量消息中，另一个线程负责发送这些批量消息到对应的 Kafka Broker。
当 Broker 接收到消息后，如果成功写入则返回一个包含消息的主题、分区及位移的 RecordMetadata 对象，否则返回异常。
生产者接收到结果后，对于异常可能会进行重试。

Producer Exactly Once

0.11 版本的 Kafka，引入了幂等性：Producer 不论向 Server 发送多少重复数据，Server 端都只会持久化一条。
这样的机制，可能导致消息重复发，需要引入幂等性，可以在发送消息到一个 Sequence Number 序列号，用于事务号

Consumer

从 kafka 消息系统消费消息。
消费组，多个消费者形成一个消费组来消息主题消息，提高消息消费能力
消费者的数量不应该比分区数多，多出的分区数将接收不到消息，没有任何意义
消费组，每个不同的消费组可以得到全量的消息，并不会因为某个消费组已经消费该条消息，收不到消息
当一个消费者加入或离开消费组后，会重新计算平衡，导致消息短暂接收不到。消费者应该保持心跳到消费组协调器
消息消费等级提交
At Last Once 至少一次
At More Once 至多一次
At Only Once 恰好一次

总而言之，我们可以通过增加消费组的消费者来进行水平扩展提升消费能力。
这也是为什么建议创建主题时使用比较多的分区数，这样可以在消费负载高的情况下增加消费者来提升性能。
另外，消费者的数量不应该比分区数多，因为多出来的消费者是空闲的，没有任何帮助。
如果我们的 C1 处理消息仍然还有瓶颈，我们如何优化和处理？
把 C1 内部的消息进行二次 sharding，开启多个 goroutine worker 进行消费，为了保障 offset 提交的正确性，需要使用 watermark 机制，保障最小的 offset 保存，才能往 Broker 提交。

Consumer Group

Kafka 一个很重要的特性就是，只需写入一次消息，可以支持任意多的应用读取这个消息。
换句话说，每个应用都可以读到全量的消息。为了使得每个应用都能读到全量消息，应用需要有不同的消费组。
对于上面的例子，假如我们新增了一个新的消费组 G2，而这个消费组有两个消费者如图。
在这个场景中，消费组 G1 和消费组 G2 都能收到 T1 主题的全量消息，在逻辑意义上来说它们属于不同的应用。
最后，总结起来就是：如果应用需要读取全量消息，那么请为该应用设置一个消费组；如果该应用消费能力不足，那么可以考虑在这个消费组里增加消费者。

可以看到，当新的消费者加入消费组，它会消费一个或多个分区，而这些分区之前是由其他消费者负责的。
另外，当消费者离开消费组（比如重启、宕机等）时，它所消费的分区会分配给其他分区。这种现象称为重平衡（Rebalance）。
重平衡是 Kafka 一个很重要的性质，这个性质保证了高可用和水平扩展。不过也需要注意到，在重平衡期间，所有消费者都不能消费消息，因此会造成整个消费组短暂的不可用。
而且，将分区进行重平衡也会导致原来的消费者状态过期，从而导致消费者需要重新更新状态，这段期间也会降低消费性能。
消费者通过定期发送心跳（Hearbeat）到一个作为组协调者（Group Coordinator）的 Broker 来保持在消费组内存活。这个 Broker 不是固定的，每个消费组都可能不同。
当消费者拉取消息或者提交时，便会发送心跳。如果消费者超过一定时间没有发送心跳，那么它的会话（Session）就会过期，组协调者会认为该消费者已经宕机，然后触发重平衡。

可以看到，从消费者宕机到会话过期是有一定时间的，这段时间内该消费者的分区都不能进行消息消费。
通常情况下，我们可以进行优雅关闭，这样消费者会发送离开的消息到组协调者，这样组协调者可以立即进行重平衡而不需要等待会话过期。
在 0.10.1 版本，Kafka 对心跳机制进行了修改，将发送心跳与拉取消息进行分离，这样使得发送心跳的频率不受拉取的频率影响。
另外更高版本的 Kafka 支持配置一个消费者多长时间不拉取消息但仍然保持存活，这个配置可以避免活锁（livelock）。活锁，是指应用没有故障但是由于某些原因不能进一步消费。
但是活锁也很容易导致连锁故障，当消费端下游的组件性能退化，那么消息消费会变的很慢，会很容易出发 livelock 的重新均衡机制，反而影响力吞吐。

leader

早期没有备份机制，一个 broker 挂掉之后，该 broker 负责的分区，数据就丢失了。所以引入了备份机制，每个分区都开始有了备份。这就需要引入一个 leader 来协调这些备份。生产器和消费器只与这个 leader 来交互。其它的备份作为 Follower 从 leader 来获取数据

由于 kafka 集群依赖 zookeeper 集群，最简单的方案所有节点都在 zookeeper 上设置一个 watch，当 leader 挂掉之后，都开始尝试枚举 leader 节点，成功的当选。这种方案容易产生脑裂。急需引入新的算法解决这个问题
前面的方案有以下缺点：

split-brain (脑裂): 这是由 ZooKeeper 的特性引起的，虽然 ZooKeeper 能保证所有 Watch 按顺序触发，但并不能保证同一时刻所有 Replica “看”到的状态是一样的，这就可能造成不同 Replica 的响应不一致。
herd effect (羊群效应): 如果宕机的那个 Broker 上的 Partition 比较多，会造成多个 Watch 被触发，造成集群内大量的调整。
ZooKeeper 负载过重 : 每个 Replica 都要为此在 ZooKeeper 上注册一个 Watch，当集群规模增加到几千个 Partition 时 ZooKeeper 负载会过重。

备份算法：
raft 算法：基于多数成功就认为成功，kafka 即将使用
pacific 基于所有节点成功认为成功，kafka 正在使用
bookkeeper:基于固定节点数成功认为成功，

数据可靠性

0： producer 不等待 Broker 的 Ack，牺牲可靠性保证时效性
1： producer 等待 broker 的 Ack，等待 leader 落盘，在返回，如果 leader 故障未来得及备份，则丢失数据
-1: 所有 broker 的 Ack，但是 broker 发送 Ack 时，leader 发生故障，则数据重复

性能优化

架构层面：
分区并行： Broker，disk，consumer
ISR

IO 层面：

batch IO 读写
磁盘顺序 IO
page cache
zero copy
压缩

References

golang

12.消息队列kafka

消息队列 kafka

基本概念

为何使用消息系统

Topic && Partition

存储原理

存储原理

Producer

Producer Exactly Once

Consumer

Consumer Group

leader

数据可靠性

性能优化

References