多图详解kafka生产者消息发送过程[通俗易懂]

持续创作，加速成长！这是我参与「掘金日新计划 · 10 月更文挑战」的第15天，点击查看活动详情

生产者客户端代码

public class SzzTestSend {

    public static final String bootStrap = "xxxxxx:9090";
    public static final String topic = "t_3_1";

    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,bootStrap);
        // 序列化协议 下面两种写法都可以
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        //过滤器 可配置多个用逗号隔开
        properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,"org.apache.kafka.clients.producer.SzzProducerInterceptorsTest");
        //构造 KafkaProducer
        KafkaProducer producer = new KafkaProducer(properties);
        // 发送消息, 并设置 回调(回调函数也可以不要)
        ProducerRecord<String,String> record = new ProducerRecord(topic,"Hello World!");
        try {
            producer.send(record,new SzzTestCallBack(record.topic(), record.key(), record.value()));
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    /** * 发送成功回调类 */
    public static class SzzTestCallBack implements Callback{
        private static final Logger log = LoggerFactory.getLogger(SzzTestCallBack.class);
        private String topic;
        private String key;
        private String value;

        public SzzTestCallBack(String topic, String key, String value) {
            this.topic = topic;
            this.key = key;
            this.value = value;

        }
        public void onCompletion(RecordMetadata metadata, Exception e) {
            if (e != null) {
                log.error("Error when sending message to topic {} with key: {}, value: {} with error:",
                        topic, key,value, e);
            }else {
                log.info("send message to topic {} with key: {} value:{} success, partiton:{} offset:{}",
                        topic, key,value,metadata.partition(),metadata.offset());
            }
        }
    }
}

IT知识分享网

构造KafkaProducer

KafkaProducer通过解析producer.propeties文件里面的属性来构造自己。 例如 ：分区器、Key和Value序列化器、拦截器、RecordAccumulator消息累加器 、元信息更新器、启动发送请求的后台线程

IT知识分享网        //构造 KafkaProducer
        KafkaProducer producer = new KafkaProducer(properties);

生产者元信息更新器

我们之前有讲过. 客户端都会保存集群的元信息,例如生产者的元信息是 ProducerMetadata. 消费组的是ConsumerMetadata 。

元信息都会有自己的自动更新逻辑, 详细请看Kafka的客户端发起元信息更新请求

相关的Producer配置有:

虽然Producer元信息会自动更新, 但是有可能在生产者发送消息的时候,发现某个TopicPartition不存在,这个时候可能就需要立刻发起一个元信息更新了。

集群资源变更监听器

org.apache.kafka.common.ClusterResourceListener

在构造KafkaConsumer的时候, 还会构造一个 集群资源变更监听器ClusterResourceListener

当用户希望收到有关集群元数据更改的通知时，可以实现回调接口。

需要在拦截器、指标采样器、序列化器和反序列化器 中访问集群元数据的用户可以实现此接口。

public interface ClusterResourceListener {
    /** * 用户可以实现以获取 ClusterResource 更新的回调方法。 * @param clusterResource cluster metadata */
    void onUpdate(ClusterResource clusterResource);
}

下面描述了每种类型的方法调用顺序。

Clients

在每个元数据响应之后都会调用一次 onUpdate(ClusterResource)

当在org.apache.kafka.clients.producer.ProducerInterceptor实现的ClusterResourceListener的时候

调用顺序为: ProducerInterceptor.onSend() -> onUpdate(ClusterResource) -> ProducerInterceptor.onAcknowledgement()

当在org.apache.kafka.clients.consumer.ConsumerInterceptor实现的ClusterResourceListener的时候

调用顺序为：onUpdate() – > ConsumerInterceptor.onConsume()

当在org.apache.kafka.common.serialization.Serializer实现的ClusterResourceListener的时候

调用顺序为：onUpdate() – > Serializer.serialize(String, Object)

当在org.apache.kafka.common.serialization.Deserializer 实现的ClusterResourceListener的时候

调用顺序为：onUpdate() – > .Deserializer.deserialize(String, byte[])

生产者拦截器

生产者拦截器在消息发送之前可以做一些准备工作, 比如 按照某个规则过滤某条消息, 又或者对 消息体做一些改造, 还可以用来在发送回调逻辑之前做一些定制化的需求,例如统计类的工作! 拦截器的执行时机在最前面,在消息序列化和分区计算之前

相关的Producer配置有:

生产者分区器

用来设置发送的消息具体要发送到哪个分区上

相关的Producer配置有:

Sender线程启动

Sender是专门负责将消息发送到Broker的I/O线程。

相关的Producer配置有:

发送请求

IT知识分享网
            producer.send(record,new SzzTestCallBack(record.topic(), record.key(), record.value()));
            

生产者拦截器

发送消息的第一步就是执行拦截器

一般情况下我们可能不需要拦截器, 但是我们需要用拦截器的时候按照下面操作执行:

1. 在配置文件中配置属性interceptor.classes=拦截器1,拦截器2,拦截器3
2. 实现接口org.apache.kafka.clients.producer.ProducerInterceptor<K, V>

这个interceptor.classes中的属性可以配置多个拦截器, 用逗号隔开,并且执行顺序就是按照配置的顺序执行的。

拦截器的执行时机在最前面,在消息序列化和分区计算之前

ProducerInterceptor

org.apache.kafka.clients.producer.ProducerInterceptor<K, V> 接口方法讲解：

  public ProducerRecord<K, V> onSend(ProducerRecord<K, V> record);
  
  public void onAcknowledgement(RecordMetadata metadata, Exception exception);
  
  public void close();

onSend(ProducerRecord<K, V> record)方法 :

当客户端将记录发送到 KafkaProducer 时，在键和值被序列化之前调用。 该方法调用ProducerInterceptor.onSend(ProducerRecord)方法。 从第一个拦截器的onSend()返回的 ProducerRecord传递给第二个拦截器 onSend()，在拦截器链中依此类推。 从最后一个拦截器返回的记录就是从这个方法返回的。 此方法不会抛出异常。 任何拦截器方法抛出的异常都会被捕获并忽略。 如果链中间的拦截器（通常会修改记录）抛出异常，则链中的下一个拦截器将使用前一个未抛出异常的拦截器返回的记录调用。

调用地方 ①. 拦截器执行时机在键值序列化之前 ②. 拦截器抛出异常会被捕获,并打印日志,那么也意味着这个拦截器所做的修改不会生效 ③.拦截器中修改的消息体会被传递到下一个拦截器

onAcknowledgement(RecordMetadata metadata, Exception exception)方法:

当发送到服务器的记录已被确认时，或者当发送记录在发送到服务器之前失败时，将调用此方法。 此方法通常在用户设置的Callback之前调用，此方法不会抛出异常。 任何拦截器方法抛出的异常都会被捕获并忽略。这个方法运行在Producer的I/O线程中,所以这个方法中的代码逻辑需要越简单越好。 否则，来自其他线程的消息发送可能会延迟。

参数： metadata – 已发送记录的元数据（即分区和偏移量）。 如果发生错误，元数据将只包含有效的主题和分区。 如果 ProducerRecord 中没有给出 partition 并且在分配 partition 之前发生错误，则 partition 将设置为 RecordMetadata.NO_PARTITION。 如果客户端将空记录传递给KafkaProducer.send(ProducerRecord)则元数据可能为空。 exception– 在处理此记录期间抛出的异常。 如果没有发生错误，则为空。

close()

主要用于在关闭拦截器时自行一些资源清理工作。

configure(Map<String, ?> configs)

ProducerInterceptor接口中集成了一Configurable接口,接口有个方法

    void configure(Map<String, ?> configs);

也就是说在拦截器中,我们可以拿到所有的配置属性了; 这个方法在这几个方法中最早执行

生产者拦截器示例

将发送的消息加上后缀 注意这里消息value的类型是String ，如果是byte则需要处理一下

    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        System.out.println("生产者拦截器 onSend() run ."+record);
        return new ProducerRecord<>(
                record.topic(), record.partition(), record.key(), record.value().concat("_后缀"));    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        System.out.println("生产者拦截器 onAcknowledgement run ."+metadata.toString() +" exception:"+exception);
    }
    @Override
    public void close() {
        System.out.println("生产者拦截器 close() run .");
    }

    @Override
    public void configure(Map<String, ?> configs) {
        this.configs = configs;
        System.out.println("生产者拦截器 configure run ."+configs);
    }

更新元信息waitOnMetadata

在发送消息之前,要先获取一下将要发送的TopicPartition的元信息。这个获取元信息的请求也是通过唤醒 Sender线程进行发送的。

1 . ProducerMetadata 元信息Map<String, Long> topics中保存Topic的有效期时间, metadata.max.idle.ms 控制,默认300000 2. ProducerMetadata 元信息Set<String> newTopics中保存所有Topic 3. 获取 Topic的元数据集群以及我们等待的时间（以毫秒为单位）, 这个获取元数据不是这里获取的,这里只是判断当前是否已经获取到了元数据,如果没有获取到,则一直等待,最大等待时间由max.block.ms控制,默认60000(1分钟),关于获取元数据在最上面已经分析过了, 是Sender线程获取并更新的。如果等待时间超过了max.block.ms,很有可能网络异常,那么会抛出超时异常。 4. 当你发送消息的时候指定了分区号, 但是这个分区号是不存在的, 这个时候就会一直发起Metadata请求(流程看最上面), 直到超时(max.block.ms)之后 抛出异常

org.apache.kafka.common.errors.TimeoutException: Topic t_3_1 not present in metadata after 60000 ms.

相关的Producer配置有:

KafkaProducer producer = new KafkaProducer(properties); 在构建KafkaProducer对象的时候, 有构建 producer I/O thread, 并且启动了, Runnable是 sender

最终调用NetworkClient.poll(long timeout, long now)里面maybeUpdate()方法 这个方法会获取 前Node中负载最少的节点发起网络请求, 如果所有Node都是满负载则请求不会被发起。

如何判断哪个节点负载最少？

通过每个节点的InFlightRequests(空中请求数量)里面的最小数量判断,这个表示当前正在发起的请求,但是还没有收到回复的请求数量; 保存形式是一个HashMap,key是Node的Id, value是所有当前还在请求中的节点; 当请求完成,请求就会在这个队列里面移除; 如果这个队列一直是满的,说明当前负载很高或者网络连接有问题。如果所有Node都是满负载则请求不会被发起，除非等到队列数量减少。

    private final Map<String, Deque<NetworkClient.InFlightRequest>> requests = new HashMap<>();

每个Node最大负载数 ？

每个客户端在发起请求还没有收到回复的时候都会被缓存到InFlightRequests(空中请求数量)里面,但是这个数量是有限制的,这个可以通过配置max.in.flight.requests.per.connection 进行设置, 默认是: 5; 也就是每个客户端对每个Node最多也就同时发起 5 个未完成的请求; 如果超时这个数量就会等待有请求完成并释放额度了才可以发起新的请求；

相关的Producer配置有:

KeyValue序列化

将key和Value先序列化。

自定义序列化器,需要实现org.apache.kafka.common.serialization.Serializer接口。 我们简单看下StringSerializer序列化器

public class StringSerializer implements Serializer<String> {
    private String encoding = "UTF8";

    @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)
            encoding = (String) encodingValue;
    }

    @Override
    public byte[] serialize(String topic, String data) {
        try {
            if (data == null)
                return null;
            else
                return data.getBytes(encoding);
        } catch (UnsupportedEncodingException e) {
            throw new SerializationException("Error when serializing string to byte[] due to unsupported encoding " + encoding);
        }
    }
}

configure(Map<String, ?> configs, boolean isKey) 这个方法是在构造 KafkaProduce实例的时候调用的。isKey 表示是 key还是value来进行序列化 这里 serialize(String topic, String data) 方法直接将字符串转换成byte[]类型。

Kafka客户端提供了很多种序列化器供我们选择,如果这些序列化器你都不满意,你也可以选择其他一些开源的序列化工具,或者自己进行实现。

计算分区号

将序列化后的key、 value 调用合适的分区器选择将要发送的分区号。

分区三种策略

将消息缓存进RecordAccumulator累加器中

图解Kafka Producer中的消息缓存模型

Sender发送消息

Sender线程在构造KafkaProducer的时候就已经启动了,它的职责就是从

以下忽略部分代码省略

    void runOnce() {
        long currentTimeMs = time.milliseconds();
        long pollTimeout = sendProducerData(currentTimeMs);
        client.poll(pollTimeout, currentTimeMs);
    }
    
    private long sendProducerData(long now) {
      
       // 获取哪些数据准备好了发送
        RecordAccumulator.ReadyCheckResult result = this.accumulator.ready(cluster, now);

    }

寻找准备好发送的消息Batch,获取对应Leader所在的ReadyNode

我们都知道生产者生产的消息是暂时缓存在消息累加器RecordAccumulator中的, Sender负责从RecordAccumulator里面获取准备好的数据进行发送

那么 ，哪些属于准备好的数据呢？

我们先回顾一下 RecordAccumulator的结构。

每个TopicPartition的消息都会被暂存在ProducerBatch Deque 阻塞队列中的其中一个ProducerBatch中,每个ProducerBatch都存放着一条或者多条消息。

具体请看 图解Kafka Producer 消息缓存模型

满足发送的条件的Batch

遍历每个TopicPartition里面的Deque, 获取队列中的第一个ProducerBatch 如果该TopicPartition不存在Leader,则忽略该Batch,如果有则进入判断流程

因为消息是要发Leader所在的Broker发送的, 所以必须要有Leader。

   public Map<Integer, List<ProducerBatch>> drain(Cluster cluster, Set<Node> nodes, int maxSize, long now) {
        if (nodes.isEmpty())
            return Collections.emptyMap();
		// 遍历ReadyNodes 每个Node下的队列都获取一遍
        Map<Integer, List<ProducerBatch>> batches = new HashMap<>();
        for (Node node : nodes) {
            List<ProducerBatch> ready = drainBatchesForOneNode(cluster, node, maxSize, now);
            batches.put(node.id(), ready);
        }
        return batches;
    }

 Map<Integer, List<ProducerBatch>> batches