生产者

生产者消息写入过程

1. 序列化 ProducerRecord

   每个消息是一个 ProducerRecord 对象，必须指定消息所属的 Topic 和消息值 Value，此外还可以指定消息所属的 分区 以及消息的 Key。

2. 推送到分区

   推送到分区kafka有三种策略：随机策略、轮询策略、key的hash策略。默认为轮询策略。

   当然如果该消息指定了 key，则 partitioner（分区器）会根据 key 的哈希值来选择目标分区，将具有相同 key 的所有消息都路由到相同的分区中；若该消息未指定 key，则 partitioner 使用轮询的方式确认目标分区。

3. 找到分区副本 leader

   producer 先从 zookeeper 的 “/brokers/…/state”节点找到该 partition 的 leader，将消息发送给该 leader。

   leader将消息追加到本地log, followers 副本从 leader 拉取消息，追加到各自的本地log,后向 leader 发送 ACK

   leader 收到所有 ISR 中的 replication 的 ACK 后，增加 HW（high watermark，最后 commit 的 offset）并向 producer 发送 ACK。

生产者消息发送方式

1. 批量发送

   消息写入到目标分区的一块内存缓冲池中。而 producer 的另一个工作线程(I/O 发送线程，也称 Sender 线程)则负责实时地从该缓冲区中提取准备就绪的消息封装进一个批次(batch),统一发送给对应的 broker。

2. 同步异步发送

   Kafka producer 发送消息的主方法是 send 方法，producer 在底层完全实现了异步化发送，并且通过 Java 提供的 Future 同时实现了同步发送和异步发送+回调(Callback)(默认异步)两种发送方式。最后 producer 程序结束时需要关闭 producer。

asks 参数

• acks 设置为 0，表示 producer 不会等待 broker 的响应；所以 producer 无法知道消息是否发送成功，这样有可能会导致数据丢失，但同时，acks 值为 0 会得到最大的系统吞吐量。
• 若 acks 设置为 1，表示 producer 会在 leader partition 收到消息时得到 broker 的一个确认，这样会有更好的可靠性，因为客户端会等待直到 broker 确认收到消息。
• 若设置为-1，producer 会在所有备份的 partition 收到消息时得到 broker 的确认，这个设置可以得到最高的可靠性保证。

消费者

消费消息过程

1. Consumer 端使用 zookeeper 用来注册 consumer 信息,其中包括 consumer 消费的 partition 列表等,同时也用来发现 broker 列表,并和 partition leader 建立 socket 连接,并获取消息。

2. 订阅 topic 后，当 consumer 调用 pull(拉取模型)时，会自动加入相应的 Consumer Group；只要 consumer 持续 pull，consumer 将持续的从分配给他的 topic partitions 接收消息；

3. 提交偏移量

   消费者通过往一个叫作 ＿consumer_offset 的特殊主题发送消息，消息里包含每个分区的偏移量。 如果消费者一直处于运行状态，那么偏移量就没有什么用处。不过，如果有消费者退出或者新分区加入，此时就会触发再均衡。完成再均衡之后，每个消费者可能分配到新的分区，而不是之前处理的那个。为了能够继续之前的工作，消费者需要读取每个分区最后一次提交的偏移量，然后从偏移量指定的地方继续处理。 因为这个原因，所以如果不能正确提交偏移量，就可能会导致数据丢失或者重复出现消费

4. consumer 会在后台持续向服务发送心跳，如果 consumer 进程崩溃或者在 session.timeout.ms 期间没有发送心跳，这个 consumer 将会被认为已经死掉了，他的分区将会被重新分配。

参数配置

1. enable.auto.commit

   将以配置项 auto.commit.interval.ms 指定的频率自动提交 offset

2. auto.commit.interval.ms

   提交 offset 的频率

总结

本文简单介绍了生成与消费过程，不过我们在过程中可能会出现消息丢失与重复消费的问题。