Kafka源码P1-消息ProducerRecord

欢迎来到啾啾的博客🐱。
记录学习点滴。分享工作思考和实用技巧，偶尔也分享一些杂谈💬。
有很多很多不足的地方，欢迎评论交流，感谢您的阅读和评论😄。

1 引言

在之前的Kafka篇章中，我们已经了解到Kafka Producer内部会有一个缓冲区，生产者通过批量发送消息的方式提升总体的吞吐。

批量发送在很多场景中都很常见，Kafka是如何实现的？

首先，我们需要理解消息。

2 消息ProducerRecord

Kafka消息的封装如下：

2.1 分区器

Kafka Producer可以指定消息发送到哪个Topic，也可以指定到哪个Partition。没有指定Partition时，Kafka Producer会使用分区器 Partitioner来决定消息应该到哪个partition。

可以看一下分区器 Partitioner 的方法

• 参数如下：
  ProducerRecord<K, V> record：要发送的 Kafka 消息记录，包含主题、键、值、分区等信息。
  byte[] serializedKey：消息键的序列化字节数组。
  byte[] serializedValue：消息值的序列化字节数组。
  Cluster cluster：Kafka 集群的元数据信息，包含主题、分区等信息。

其中serializedKey和serializedValue是分别用了同一个序列化类的不同的序列化对象来做的：

有意思的是序列化方法传入的topic与headers信息在Kafka的默认序列化中是没有被使用的。

很显然，这两个参数topic和headers是为了序列化的扩展性预留的设计。
使用topic参数，可以为不同的topic准备不同的序列化策略，比如加密。

import org.apache.kafka.common.serialization.Serializer;
import java.nio.charset.StandardCharsets;
import java.util.Map;

public class TopicAwareSerializer implements Serializer<String> {
    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        // 配置逻辑
    }

    @Override
    public byte[] serialize(String topic, String data) {
        if ("sensitive-topic".equals(topic)) {
            // 对敏感主题的数据进行特殊处理
            String encryptedData = "encrypted:" + data;
            return encryptedData.getBytes(StandardCharsets.UTF_8);
        }
        return data.getBytes(StandardCharsets.UTF_8);
    }

    @Override
    public void close() {
        // 关闭逻辑
    }
}

使用headers参数，比如编码格式、版本号等，也可以做一些定制化操作。

import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.serialization.Serializer;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.Map;

public class HeaderAwareSerializer implements Serializer<String> {
    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        // 配置逻辑
    }

    @Override
    public byte[] serialize(String topic, Headers headers, String data) {
        Charset charset = StandardCharsets.UTF_8;
        if (headers.lastHeader("charset") != null) {
            String charsetName = new String(headers.lastHeader("charset").value(), StandardCharsets.UTF_8);
            charset = Charset.forName(charsetName);
        }
        return data.getBytes(charset);
    }

    @Override
    public byte[] serialize(String topic, String data) {
        return serialize(topic, null, data);
    }

    @Override
    public void close() {
        // 关闭逻辑
    }
}

需要注意的是，key|value.serializer都必须被设置为实现了org.apache.kafka.common.serialization.Serializer接口的类。

没有指定分区partition时，分区器partitioner.partition方法有三种实现。Kafka 在 Producer 中默认（不配置分区器）使用的是 DefaultPartitioner。你可以在创建 KafkaProducer 时通过配置 partitioner.class 属性来指定使用的分区器，若不指定，就会使用默认的 DefaultPartitioner。

• DefaultPartitioner
  用途：这是 Kafka 默认的分区器。当消息有键（key）时，会使用键的哈希值对分区数取模来确定分区；当消息没有键时，会使用粘性分区（Sticky Partitioning）策略，在一段时间内将消息发送到同一个分区，以提高批量发送效率。
  使用场景：大多数常规业务场景，无需特殊分区策略时使用。
  

• RoundRobinPartitioner
  用途：采用轮询的方式依次将消息发送到各个分区，确保消息均匀分布在所有分区上。
  使用场景：需要消息均匀分布，且对消息顺序没有严格要求的场景。
  

• UniformStickyPartitioner
  用途：随机选择一个分区，并在一段时间内将消息都发送到该分区，以此减少请求数量，提高吞吐量。
  使用场景：对吞吐量要求较高，且对消息顺序没有严格要求的场景。

所以要指定好Partition。