Kafka 生产者

生产者就是负责向Kafka发送消息的应用程序。

1 生产者入门

maven 依赖：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.9.0</version>
</dependency>

生产者的使用步骤：

1. 配置生产者客户端参数。
2. 创建生产者实例。
3. 构建待发送的消息。
4. 发送消息。
5. 关闭生产者实例。

public class ProducerClient {

    private static Properties initProp() {
        Properties props = new Properties();
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); // key 的系列化类的全名
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName()); // value 的系列化类的全名
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, CommonAttribute.URL_HOST); // kafka 服务器地址
        return props;
    }

    public static void main(String[] args) {
        Producer<String,String> producer = new KafkaProducer<>(initProp()); // 创建生产者实例
        ProducerRecord<String,String> record = new ProducerRecord<>(CommonAttribute.TOPIC,"hello kafka"); // 构建待发送的消息，第一个参数为主题名称
        producer.send(record);
        producer.close();
    }
}

1.1 发送消息的三种模式

图 发送消息的三种模式

public class SendMode {

    public static void main(String[] args) throws InterruptedException {
        Producer<String,String> producer = new KafkaProducer<>(CommonAttribute.getProducerProps());
        new Thread(() -> {
            ProducerRecord<String,String> record = new ProducerRecord<>(CommonAttribute.TOPIC,"发送即忘");
            producer.send(record);
            System.out.println("消息已发送");
        }).start();

        new Thread(() -> {
            ProducerRecord<String,String> record = new ProducerRecord<>(CommonAttribute.TOPIC,"同步发送");
            Future<RecordMetadata> future = producer.send(record);
            try {
                future.get();// 同步发送
                System.out.println("同步发送完成");
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }).start();

        new Thread(() -> {
            ProducerRecord<String,String> record = new ProducerRecord<>(CommonAttribute.TOPIC,"异步发送");
            producer.send(record, (metadata, exception) -> { // 当发送成功时 exception为空
                System.out.println("发送完成:" + metadata);
            });
            producer.send(record);
        }).start();

        Thread.sleep(2000);
        producer.close();
    }
}

1.1.1 同步发送原理

生产者Producer的send方法将返回一个Future实例。KafkaProducer类返回的是CompletableFuture实例。下面是该类的get方法。

public T get() throws InterruptedException, ExecutionException {
    Object r;
    return reportGet((r = result) == null ? waitingGet(true) : r);
}

当结果为空的时候，调用waitingGet方法。

图 CompletableFuture类的waitingGet方法部分代码。

该方法有个while循环，其退出条件是响应结果不为空。

1.2 消息记录的处理

图 Kafka生产者处理消息的流程

1.2.1 生产者拦截器

可以用于在消息发送前修改消息内容及在发送回调逻辑之前进行一些定制化的需求。

在配置生产者参数(interceptor.classes)时指定拦截器的全名，可以指定多个，用“,”隔开。

图 ProducerInterceptor接口的UML

configure方法用于获取配置信息及初始化数据。

close方法用于在关闭拦截器时执行一些资源的清理工作。

onSend方法会在将消息序列化之前调用，返回值是经过处理的消息（返回值不能为空，否则会报错）。

onAcknowledgement方法会在消息被应答之前或消息发送失败时被调用。优先于用户设定的Callback之前执行。该方法运行在Producer的I/O线程中，如果定制的方法逻辑复杂，可能会影响消息的发送速度。

出了configure方法外，其他方法抛出的异常都会被捕获并记录到日志中，但不会再向上传递。

public class User1ProducerInterceptor implements ProducerInterceptor<Integer, User> {
    @Override
    public ProducerRecord<Integer, User> onSend(ProducerRecord<Integer, User> record) {
        System.out.println("interceptor1:onSend" + record.value());
        if (record.value() != null && record.value().getName() != null) {
            record.value().setName("interceptor1a," + record.value().getName());
        }
        return record;
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        System.out.println("interceptor1:onAcknowledgement：" + metadata.serializedValueSize());
    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> configs) {

    }
}

1.2.2 序列化器

生产者需要用序列化器把对象转成字节数组才能发送给Kafka。同理，消费者也需要用反序列化器把这些字节数组转换成相应的对象。

在配置生产者参数时指定（key.serializer与value.serializer）。

图 Serializer的UML

close方法在关闭序列化器时被调用。

configure方法用于获取配置信息及初始化数据。

serialize 将目标对象转换成字节数组。

public class StringSerializer implements Serializer<String> {
    private Charset encoding = StandardCharsets.UTF_8;

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        String propertyName = isKey ? "key.serializer.encoding" : "value.serializer.encoding";
        Object encodingValue = configs.get(propertyName);
        if (encodingValue == null)
            encodingValue = configs.get("serializer.encoding");
        if (encodingValue instanceof String) {
            String encodingName = (String) encodingValue;
            try {
                encoding = Charset.forName(encodingName);
            } catch (UnsupportedCharsetException | IllegalCharsetNameException e) {
                throw new SerializationException("Unsupported encoding " + encodingName, e);
            }
        }
    }

    @Override
    public byte[] serialize(String topic, String data) {
        if (data == null)
            return null;
        else
            return data.getBytes(encoding);
    }
}

1.2.3 分区器

用于确定消息发往的分区。如果消息ProducerRecord中指定了partition字段，则不需要分区器的作用。

在配置生产者参数(partitioner.class)时指定。

图 Partitioner 的UML

partition方法返回值为计算的分区号。

注意：如果key不为null,那么计算得到的分区号是所有分区中的任意一个；如果为null，则得到的分区号仅为可用分区中的任意一个。

2 消息发送过程

图 生产者客户端的整体架构

整个生产者客户端由主线程及Sender线程协调运行。

1. 创建消息并通过可能的拦截器、序列化器和分区器的作用之后，缓存到消息累加器中。
2. Sender线程从累加器获取缓存消息之后，将原本<分区,Deque<ProducerBatch>>的保存形式转变成<Node,List<ProducerBatch>>的形式（Node表示Kafka的broker节点信息），再进一步分装成<Node,Request>的形式。
3. 在将消息发往Kafka之前，会将请求保存到InFlightRequests中，用于缓存已经发送出去但还没收到响应的请求。
4. 将请求提交到Selector。
5. Selector将请求发送给Kafka集群。
6. 集群将响应返回给Selector。
7. Selector根据响应来更新InFlightRequests
8. 清理消息累加器。

2.1 消息累加器

RecordAccumulator 用于缓存消息以便Sender线程可以批量发送，进而减少网络运输的资源消耗。

主线程中发送过来的消息都会被追加到某个分区的双端队列中。

图 消息流入并添加到消息累加器的过程

在RecordAccumulator 的内部，为每个分区都维护了一个双端队列Deque<ProducerBatch>。

2.1.1 内存的复用

在消息发送之前，需要创建一块内存区域来保存对应的消息。如果频繁的创建和释放是比较耗费资源的。

RecordAccumulator 内部的BufferPool主要用来实现ByteBuffer的复用。

图 创建消息批次到内存释放过程

注意：创建ProducerBatch时，匹配ByteBuffer时，如果消息大小小于batch.size，那么创建的ByteBuffer 就能被复用。否则就以该消息大小作为ByteBuffer的大小来创建，且不会复用它。

buffer.memory 配置消息累加器的缓存大小，默认值33554432B,即32MB。如果生产消息的速度过快于发送到Kafka服务器的速度，则会导致其缓存大小不足。此时send方法要么被阻塞，要么抛出异常。

而BufferPool的缓存大小由batch.size 配置，默认值为16384B,即16KB。

2.1.2 内存管理策略

增加buffer.memory 大小来提高缓存容量。增加batch.size的大小可以减少ByteBuffer的创建和销毁次数，提高内存复用效率。

但是过大的缓存容量及Batch大小可能会导致内存浪费和延迟增加。

2.2 leastLoadedNode

所有Node中，负载最小的。即每个Node在InFlightRequest中还未确定的请求数量最少的。

选择leastLoadedNode发送请求可以使它能够尽快发出，避免因网络堵塞等异常而影响整体的进度。

2.2.1 元数据的更新

元数据是指Kafka集群的元数据，包括主题信息、分区信息等。这些信息是通过动态获取的。

图 原始更新的流程

2.3 重要的生产者参数

图 重要的生产者参数