kafka producer batch 发送消息

本文转载自 allenwas3 查看原文 2019/01/06 4 kafka

1. 使用 KafkaProducer 发送消息，是按 batch 发送的，producer 首先把消息放入 ProducerBatch 中：

org.apache.kafka.clients.producer.internals.ProducerBatch


2. KafkaProduer 类中有一个 Thread 属性，负责 IO，发送和接收数据：

            this.sender = new Sender(logContext,
                    client,
                    this.metadata,
                    this.accumulator,
                    maxInflightRequests == 1,
                    config.getInt(ProducerConfig.MAX_REQUEST_SIZE_CONFIG),
                    acks,
                    retries,
                    metricsRegistry.senderMetrics,
                    Time.SYSTEM,
                    this.requestTimeoutMs,
                    config.getLong(ProducerConfig.RETRY_BACKOFF_MS_CONFIG),
                    this.transactionManager,
                    apiVersions);
            String ioThreadName = NETWORK_THREAD_PREFIX + " | " + clientId;
            this.ioThread = new KafkaThread(ioThreadName, this.sender, true);
            this.ioThread.start();

Sender 类实现了 Runnable 接口，封装了具体的逻辑，发送消息和接收响应都在这个类中。

// 发送消息
long pollTimeout = sendProducerData(now);
// 接收响应
client.poll(pollTimeout, now);

3. 执行回调

org.apache.kafka.clients.producer.internals.Sender#completeBatch()

注意！

本站转载的文章为个人学习借鉴使用，本站对版权不负任何法律责任。如果侵犯了您的隐私权益，请联系我们删除。

猜您在找

kafka producer 发送消息简介 Kafka Producer机制优化-提高发送消息可靠性 producer怎样发送消息到指定的partitions kafka producer生产消息时报错 kafka消息服务的producer、broker、consumer的配置