Kafka

architecture

• 同一个topic的partition不会跨cluster
• 一个broker里面可能含有不同topic的partition
• partition在存储上可以被看作为一个可追加的日志文件，每条消息加进去的时候带有一个offset
• fault-tolerant: 多个节点故障时，集群仍能持续提供服务且数据不丢失。
  • 同一个partition会有多个replica,每个replica要么是leader（只有一个）, 要么是follower
  • 同一个broker里面不会有同一个partition的多个replica
  • 同一个broker里面会有同一个topic的不同的partition
• 一群broker中有一个是controller,它不负责消息的传递，只负责选举partition的新leader
• Zookeeper: 管理broker的状态

partiton

• 高并发：多个producer可以向多个partition同时写入
• Kafka 只保证在单个 Partition 内消息的顺序性
• 高可用：多个replica

leader

• 负责读写
• Leader 负责管理 In-Sync Replicas（同步副本集）。它要确保 Follower 能跟上自己的进度，将落后的 Follower 踢出 ISR。
  • ISR：如果一个 Follower 在replica.lag.time.max.ms时间内没有发送pull request,它就会被踢出ISR
  • ISR存放在zookeeper中，controller broker会查看ISR中的信息，以方便知道每个partition的follwer是否跟上了它们的leader

follower

定期向leader发送pull请求

flush

持久化的时候，是先写到broker的page cache里面，然后定期刷盘，可以保证磁盘的顺序写入

producer

先序列化消息，在根据路由规则，把消息准备发送到对应的partition。但消息会被攒在内存缓冲区，属于同一个partition的消息被合并成一个batch, 由独立的 Sender 线程通过 Selector 异步发送到对应的 Broker

怎么把消息发送给对应的partition?

• if you don’t set a key, kafka will do round-robin
• if you set a key, kafka will do hashing
• if you want to keep messages in order, you can set the messages with the same key

集群启动的时候，会给每个producer分配一个unique PID,当producer发送消息给cluster的时候，每条消息其实带上了PID, partition, sequence number。 sequence number是producer对于每个partition维护的。

• ack=0: Producer 发出消息后立即认为成功，完全不等待 Broker 的任何确认
• ack=1: Producer 等待 Partition Leader 将消息写入其本地日志后，就返回成功。
• ack=-1: Producer 等待 Leader 收到所有 ISR（In-Sync Replicas）副本的同步确认 后，才返回成功

batch.size：当发往同一个 Partition 的消息累计到这个字节数，就send一次

linger.m：避免消息在缓冲区等待太久。只要消息在缓冲区中等待的时间超过了这个值，也会立即触发发送

retry

当producer发送消息给cluster失败时，它会自动retry,可以通过设置retries参数来决定最大的retry次数

如果broker已经收到了消息，但是因为网络原因，producer以为消息丢失，producer会retry，这就有可能造成消息的重复。

通过设置enable.idempotence=true可以保证幂等性

因为每个broker在收到消息的时候，一个消息会带有[PID, partition, sequence number]这样的标识。同时每个partition会维护一个最新的sequence number。

• 如果收到的新消息是最新sequence number+1，则直接更新最新sequence number+1
• 如果收到了同样的sequence number,说明消息重复， 直接丢弃
• 如果收到了跳过的sequence number, 触发producer retry

故障转移

controller是一个特殊的broker.当 Leader 故障时，Controller 会介入处理。首先，它会从当前 ISR 列表中选出一个副本作为新 Leader。然后，Controller 更新集群元数据并广播给所有 Broker，客户端随之切换到新的 Leader 进行读写

consumer group

在同一个 Consumer Group 内，任意一个 Partition 在同一时刻，只能被分配给组内的 一个特定的 Consumer 进行消费。只能消费leader partition。

• 场景一：横向拓展。消费并行度上限 = Partition 数。
• 场景二：同一份数据被多个独立业务处理。为每个业务创建一个独立的 Consumer Group。每个 Group 都会收到 Topic 的全量消息。Group 之间互不影响。

pull模式

• consumer自己决定什么时候拉、一次拉多少。（fetch.max.bytes）
• 批量拉取，减少网络往返
• 解决空轮询： Consumer 在发起拉取请求时，可以告诉 Broker：“如果积压的数据没有达到 fetch.min.bytes（比如 1KB），就别立即响应我，先等着，直到凑够这么多数据，或者等到超时时间 fetch.max.wait.ms 再回复我。

offset

kafka内部有一个topic叫做__consumer_offsets。Offset 是消费者在每个 Partition 上的消费位置指针。消费者每进行一些消费，就会把新的offset持久化存储到__consumer_offsets上。 如果配置了enable.auto.commit=true,那么默认每隔5秒，consumer把它最新读到的消息commit到那个主题上

transaction

比如一个流式处理，从Topic A读消息，写到Topic B和Topic C, 发送成功之后，再更新Topic A的offset。那么这三件事情可以放在transaction中。

可以防止丢消息/重复消费。

questions

Kafka 和传统消息队列（RocketMQ）相比，核心优势是什么？

• Kafka is natively designed witn distributed systems in mind from the start, while rocketMQ may require more complex config to be distributed.
• 吞吐量大。顺序I/O：消息顺序追加到磁盘文件，速度极快（磁盘顺序读写比内存随机读写还快）。
• 所有消息按持久化日志存储，可设置保留时间（如7天）。消费后消息仍在，支持多个消费者组重复消费、回溯数据。
• 页缓存（Page Cache）：操作系统会将频繁读写的磁盘文件缓存到空闲内存中。后续的读写很可能直接在内存中完成，性能接近内存。
• 零拷贝：消费者读取时，数据可以直接从操作系统的页缓存发送到网卡(NIC Card)，跳过用户空间的复制

如何保证 Kafka 消息的不丢失？

• producer:
  • linger配置， 如果消息在缓冲区超过linger时间还未发送，会直接触发发送
  • retries>0+幂等性设置
  • 设置ack=-1
• broker: 每个partition有多个replica
• consumer：关闭auto commit，防止有些消息还没有收到就被自动commit了。

如何保证 Kafka 消息的不重复？

• producer:
  • 配置enable.idempotence=true
  • 开启事务 Producer。比如消费A->写入B->提交A的offset，这一个过程必须是原子性的
• consumer（业务层）：
  • 数据库唯一约束
  • Redis SETNX

如何保证 Kafka 消息的顺序性？

• producer: 同一个key一定进入一个partition, 因为Kafka只能保证partition内部消息有序。key可以是userID.
• consumer: 一个 partition 在任意时刻，只能分配给一个 consumer
• 分区策略很重要，可以按照消息的属性来分区，这样相关的消息总会被分到同一个partition处理

分区策略

• 轮询策略: round robin
• 按键哈希策略: 如果消息带有键，就会根据键的哈希值来决定消息发送到哪个分区
• 自定义分区策略: 实现 Partitioner 接口