图解系列 图解Kafka之Producer

开局一张图，其他全靠吹

发送消息流程如下：

1.初始化流程

• 指定bootstrap.servers，地址的格式为 host:port。它会连接bootstrap.servers参数指定的所有Broker，Producer启动时会发起与这些Broker的连接。因此，如果你为这个参数指定了1000个Broker连接信息，那么很遗憾，你的Producer启动时会首先创建与这1000个Broker的TCP连接。

  • 在实际使用过程中，我并不建议把集群中所有的Broker信息都配置到bootstrap.servers中，通常你指定3～4台就足以了。因为Producer一旦连接到集群中的任一台Broker，就能拿到整个集群的Broker信息，故没必要为bootstrap.servers指定所有的Broker。
  • props.put("bootstrap.servers", "localhost:9092");

• 指定Key和Value的序列化方式。

  •  props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
     props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    1
    2

• 指定acks配置，默认值是all(版本3.x)

  • props.put("acks", "all");
  • 设置为0，表示生产端发送消息后立即返回，不等待broker端的响应结果。通常此时生产端吞吐量最高，消息发送的可靠性最低。
  • 设置为1，表示leader副本成功写入PageCache就会响应Producer，而无需等待ISR（同步副本）集合中的其他副本写入成功。这种方案提供了适当的持久性，保证了一定的吞吐量。
  • 设置成all或-1，表示不仅要等leader副本成功写入，还要求ISR中的其他副本成功写入，才会响应Producer。这种方案提供了最高的持久性，但也提供了最差的吞吐量。

• producer = new KafkaProducer<>(props);

  • 从配置中获取必要的参数，如transactionalId和clientId。
  • 根据clientId创建日志记录上下文（LogContext），用于日志记录。
  • 配置度量（Metrics）相关信息，包括度量标签、度量配置、度量报告器等。
  • 创建度量上下文（MetricsContext）和度量实例（Metrics）。
  • 初始化分区器（Partitioner）。
  • 配置并初始化键（key）和值（value）的序列化器（Serializer）。
  • 配置并初始化拦截器（Interceptors）。
  • 配置集群资源监听器（ClusterResourceListeners）。
  • 设置最大请求大小（maxRequestSize）、内存大小（totalMemorySize）和压缩类型（compressionType）等参数。
  • 配置最大阻塞时间（maxBlockTimeMs）和交付超时时间（deliveryTimeoutMs）。
  • 初始化API版本（apiVersions）和事务管理器（transactionManager）。
  • 创建记录累加器（RecordAccumulator），用于累积记录以进行批量发送。
  • 解析并验证引导服务器地址（addresses）。
  • 如果提供了元数据（metadata），则使用提供的元数据，否则创建新的元数据实例，并通过引导服务器地址进行引导。
  • 初始化错误度量传感器（errors）。
  • 创建并启动IO线程（ioThread）来处理消息发送。
  • 注册应用程序信息，用于JMX度量和监控。
  • 如果在初始化过程中发生任何错误，将调用关闭方法以避免资源泄漏，并向上抛出Kafka异常。

2.发送消息流程

在消息发送的过程中，涉及到了两个线程——main 线程和 Sender 线程。在 main 线程中创建了一个双端队列 RecordAccumulator。main 线程将消息发送给 RecordAccumulator，Sender 线程不断从 RecordAccumulator 中拉取消息发送到 Kafka Broker。

• 构造消息记录ProducerRecord 对象，对象包含了四个属性:Topic,partition,key,value;topic 和 value 是必须的，key 和 partition 是可选的。

• 同步获取Kafka集群信息（Cluster）。

  • 如果缓存有，并且分区没有超过指定分区范围则返回缓存
  • 否则触发更新，等待从broker获取新的元数据信息
  • 默认强制拉取时间是metadata.max.age.ms: 5分钟

• 使用键序列化器（keySerializer）将消息的键序列化为字节数组，使用值序列化器（valueSerializer）将消息的值序列化为字节数组。

• 计算数据发送到那个分区，如果指定了 key，那么相同 key 的消息会发往同一个分区，如果实现了自定义分区器，那么就会走自定义分区器进行分区路由。

  • 如果有Key值，则使用Key值的Hash值来分配分区 murmurhash(key) % 主题分区总数。
  • 老版本：如果没有key值，则以Round-Robin的方式分配分区。
  • 新版本：如果没有key值，则以粘性分区的方式分配分区

• 创建一个TopicPartition对象，表示要发送消息的主题和分区。

• 判断消息的大小是否超过了我们设置的阈值。

• 异步发送时，给每一条消息都绑定他的回调函数

• 把消息放入记录累加器（accumulator）(32M的一个内存)，然后有accumulator把消息封装成为一个批次一个批次的去发送。

• 如果批次满了或者新创建出来一个批次， 唤醒sender线程,他才是真正发送数据的线程，发送的时候并不是来一个消息就发送一个消息，这样的话吞吐量比较低，并且频繁的进行网络请求。消息是按照批次来发送的或者等待时间来发的的.

• Leader Broker接收到消息写入到PageCache，当Producer的acks设置为"all"时，这意味着Producer会等待所有ISR（In-Sync Replicas，即同步副本）都成功确认消息之后才会认为消息发送成功。

• Leader Partition接收消息：每个Partition都有一个Leader Broker，该Leader Broker负责接收所有消息并处理副本同步。Leader Partition会接收Producer发送的消息。

• 消息复制到Follower副本：Leader Partition会将消息复制到ISR中的Follower副本（In-Sync Replicas），这些副本是与Leader保持同步的副本。Kafka允许配置多个ISR，以提高可用性。

• 等待Follower副本确认：Leader Partition会等待ISR中的Follower副本确认已成功复制消息。这些确认信息包括副本在哪个Offset处复制消息。

• Leader确认消息：一旦ISR中的所有Follower副本都确认了消息，Leader Partition会向Producer发送确认消息。这表示消息已经成功写入Leader Partition并且已在ISR中的所有Follower副本中成功复制。

• Producer接收确认：Producer收到来自Kafka的确认消息，这时候Producer认为消息发送成功。

• 定期提交消息到磁盘：Leader Partition会定期将已接收的消息写入磁盘以确保持久性，以防Broker故障。

• Follower副本的同步：Follower副本会定期从Leader Partition拉取消息，确保与Leader保持同步。如果Follower副本无法赶上Leader，它可能会被认为是“失去同步”，不再被视为ISR的一部分。

示例

Properties props = new Properties();
// 配置broker列表，一般配置3-5个即可，如果你有100个broker，不需要全部配置，能连上一个就行
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.65.60:9092,192.168.65.60:9093,192.168.65.60:9094");
// 持久化机制参数
props.put(ProducerConfig.ACKS_CONFIG, "-1");
// 发送失败会重试发送失败会重试，默认重试间隔100ms，重试能保证消息发送的可靠性，但是也可能造成消息重复发送，比如网络抖动，所以需要在 接收者那边做好消息接收的幂等性处理 
// 注意：消息发送失败会自动重试，不需要我们在回调函数中手动重试
props.put(ProducerConfig.RETRIES_CONFIG, 3);
// 设置重试间隔
props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 300);
// 设置发送消息的本地缓冲区大小 32MB
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
// kafka本地线程会从缓冲区取数据，批量发送到broker，
// 设置批量发送消息的大小，默认值是16384，即16kb，就是说一个batch满了16kb就发送出去
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
// 默认值是0，意思就是消息必须立即被发送，但这样会影响性能
// 一般设置100毫秒左右，就是说这个消息发送完后会进入本地的一个batch，如果100毫秒内，这个batch满了16kb就会随batch一起被发送出去
// 如果100毫秒内，batch没满，那么也必须把消息发送出去，不能让消息的发送延迟时间太长
props.put(ProducerConfig.LINGER_MS_CONFIG, 100);
// 把发送的key从字符串序列化为字节数组
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
//把发送消息value从字符串序列化为字节数组
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
Producer<String, String> producer = new KafkaProducer<String, String>(props);

for (int i = 1; i <= msgNum; i++) {
     Order order = new Order(i, 100 + i, 1, 1000.00);
     // 1.指定发送分区
     ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(TOPIC_NAME
                    , 0, order.getOrderId().toString(), JSON.toJSONString(order));
     // 2.未指定发送分区，指定Key 具体发送的分区计算公式：hash(key) % partitionNum
     ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(TOPIC_NAME
                    , order.getOrderId().toString(), JSON.toJSONString(order));
	 // 3.既没有指定分区，也没有指定Key Kafka采用Sticky Partition（黏性分区器），会随机选择一个分区，并尽可能一直使用该分区，
     //   待该分区的batch已满或者或者linger.ms设置的时间到，Kafka再随机一个分区进行使用（和上一次的分区不同）
     // 4.自定义分区器
     // 同步发送：等待消息发送成功的同步阻塞方法
     RecordMetadata metadata = producer.send(producerRecord).get();
     System.out.println("同步方式发送消息结果：" + "topic-" + metadata.topic() + "|partition-"
                    + metadata.partition() + "|offset-" + metadata.offset());*/

     // 异步回调方式发送
     producer.send(producerRecord, new Callback() {
          // 回调函数会在 producer 收到 ack 时调用，为异步调用，该方法有两个参数，分别是元数据信息（RecordMetadata）和异常信息（Exception），
          // 如果 Exception 为 null，说明消息发送成功，如果 Exception 不为 null，说明消息发送失败
public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (exception != null) {
                        System.err.println("发送消息失败：" + exception.getStackTrace());
                    }
                    if (metadata != null) {
                        System.out.println("异步方式发送消息结果：" + "topic-" + metadata.topic() + "|partition-"
                                + metadata.partition() + "|offset-" + metadata.offset());
                    }
                }
            });
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

生产者发送消息三大模式

• 发后即忘模式 fire-and-forget 模式

producer.send(producerRecord)

• 异步发送发送

producer.send(producerRecord, new Callback() )

• 同步发送模式

producer.send(producerRecord, new Callback() ).get()

生产者核心参数

• bootstrap.servers: Kafka集群的地址列表，Producer用于发现Broker。这个参数是必需的。一般配置3-5个即可，如果你有100个broker，不需要全部配置，能连上一个就行

• acks: 默认值是all，控制生产者要求Broker确认消息写入的级别。要求消息可靠性设置-1，要求吞吐量设置为1。

  • 设置为0，表示生产端发送消息后立即返回，不等待broker端的响应结果。通常此时生产端吞吐量最高，消息发送的可靠性最低。
  • 设置为1，表示leader副本成功写入PageCache就会响应Producer，而无需等待ISR（同步副本）集合中的其他副本写入成功。这种方案提供了适当的持久性，保证了一定的吞吐量。
  • 设置成all或-1，表示不仅要等leader副本成功写入，还要求ISR中的其他副本成功写入，才会响应Producer。这种方案提供了最高的持久性，但也提供了最差的吞吐量。

• retries: 默认值Integer.MAX_VALUE，生产者在发送消息时的重试次数。如果消息发送失败，Producer会尝试重新发送。设置为大于0的值可以增加消息的可靠性，一般建议设置为3。

• retry.backoff.ms: 两次重试之间的时间间隔，默认是 100ms。

• max.in.flight.requests.per.connection: 默认值5，建议 1-5，控制每个连接上允许的未确认请求的最大数量。较高的值可以提高吞吐量，但也可能导致更大的内存使用。

• buffer.memory: 设置发送消息的缓冲区，默认是32M

• compression.type: 默认是none，不压缩，但是也可以使用lz4压缩，效率还是不错的，但是会提高cpu的开销

• batch.size: 设置batch的大小，如果batch太小，会导致频繁的网络请求，吞吐量下降，如果batch太大，会导致一条消息需要等待很久才能被发送出去,而且会让内存缓冲区有很大的压力过多的数据缓存在内存里，默认值是16KB，也就是一个batch满了的时候会把这个batch发送出去，一般在生产环境会适当的增大这个值，如果消息大小大于这个，那么就会使用消息大小（源码中是这样的），一般不改动默认值

• linger.ms: 这个值默认是0，意思就是消息必须立即被发送，但是这样是不对的，一般设置一个100ms，这样的话就是如果100ms内这个batch满了16KB就会发送出去，如果是0 那么 batch.size 就失效，建议要求实时性设置0，要求吞吐量设置50-100

• key.serializer和value.serializer: 指定消息的键和值的序列化器，将Java对象转换为字节数组以便发送到Kafka。

• enable.idempotence ：默认true，用于确保生产者产生的消息具有幂等性。幂等性是指无论发送多少次相同的消息，最终的结果都是相同的，不会产生副作用，仅能保证单分区，仅该生成者产生的消息幂等，重启下就废了。

• max.block.ms: 默认60s，控制在生产者缓冲区已满时调用send方法的行为。如果生产者缓冲区已满，send方法可以选择阻塞等待空间可用，直到指定的超时时间过去。

• block.on.buffer.full: 默认true，TRUE表示当我们内存用尽时，停止接收新消息记录或者抛出错误。
  默认情况下，这个设置为TRUE。然而某些阻塞可能不值得期待，因此立即抛出错误更好。如果设置为false，则producer抛出一个异常错误：BufferExhaustedException

• request.timeout.ms: 默认30s，设置一个请求最大等待时间（单位为ms），超过这个时间则会抛Timeout异常。
  超时时间如果设置大一些，如127000（127秒），高并发的场景中，能减少发送失败的情况。

以上Kafka版本为3.X

实际上，当batch.size和linger.ms二者都配置的时候，只要满足其中一个要求，就会发送请求到broker上。

参考

• https://www.clairvoyant.ai/blog/unleash-kafka-producers-architecture-and-internal-workings
• 尚硅谷 Kafka