Kafka源码解读——生产者_KafkaProducer

痛定思痛。 2022-01-27 15:13 305阅读 0赞

### Kafka源码解读——生产者\_KafkaProducer（间断持续更......） ###

KafkaProducer类的注释里有以下几段：

*  producer是线程安全的，因此多线程使用一个单例模式的对象比多个对象更快些。
 *  producer有一个缓冲池用来保存还没有发送到服务器的消息数据，以及一个后台I/O线程负责将这些消息数据转换成一个request请求发送到Kafka集群上，注意发送完成之后一定要close掉producer，否则会造成数据丢失。
 *  发送消息数据（record）的request失败之后，producer会自动重试，将**<retries>**设置为0之后就不会自动重试。注意：启用之后有几率重复发送。
 *  producer为每个分区维护未发送记录的缓冲区，这个缓冲区的大小由**<batch.size>**定义，值越大批处理单次传输的数据越大，但是受主机内存的限制（每一个活动中的partition都有一个这样的缓冲区）
 *  默认情况下，即使缓冲区内有额外的未使用的空间也会立即send buffer将缓冲区的数据发出去。如果你要减少request的数量可以使用**<linger.ms>**配置要求值大于等于0，producer将在request之前等待该毫秒值，希望更多记录到达相同批次，这类似于TCP中的Nagle算法。举个例子，在下面的代码中，假如有100个record在一个请求中发送，我们将<linger.ms>设置为1，在缓冲区没有填满的时候，会等待1ms等待更多record到达缓冲区；注意：配置为0，在时间上紧密到达的record通常也会一起批处理，所以无论延迟配置如何，最终都会发生批处理。将**<linger.ms>**设置为大于0的值可以在不受最大负载影响的情况下以较少的延迟为代价，从而导致更少，更高效的请求。
 *  **<buffer.memory>**配置缓存区的大小，如果records的发送速度大于发送到服务器的速度，缓冲空间会被用完，额外的call将会被阻塞，阻塞的阈值时长由**<max.block.ms>**配置，超出之后会抛TimeoutException

// 提供一个简单的producer发送消息的样例
    Properties props = new Properties();
    props.put("bootstrap.servers", "localhost:9092");
    // 控制消息完整性
    props.put("acks", "all");
    props.put("retries", 0);
    props.put("batch.size", 16384);
    props.put("linger.ms", 1);
    props.put("buffer.memory", 33554432);
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    
    Producer<String, String> producer = new KafkaProducer<>(props);
    for(int i = 0; i < 100; i++)
        producer.send(new ProducerRecord<String, String>("my-topic", Integer.toString(i), Integer.toString(i)));
    
    producer.close();

### 参数 ###

private static final Logger log = LoggerFactory.getLogger(KafkaProducer.class);
        private static final AtomicInteger PRODUCER_CLIENT_ID_SEQUENCE = new AtomicInteger(1);
        private static final String JMX_PREFIX = "kafka.producer";
        // client.id
        private String clientId;
        private final Partitioner partitioner;
        private final int maxRequestSize;
        private final long totalMemorySize;
        // 存储Kafka元数据
        private final Metadata metadata;
        private final RecordAccumulator accumulator;
        private final Sender sender;
        private final Metrics metrics;
        // i/o线程，用来异步同步数据使用
        private final Thread ioThread;
        private final CompressionType compressionType;
        private final Sensor errors;
        // 系统时间
        private final Time time;
        private final Serializer<K> keySerializer;
        private final Serializer<V> valueSerializer;
        // 生产者配置信息
        private final ProducerConfig producerConfig;
        private final long maxBlockTimeMs;
        private final int requestTimeoutMs;
        private final ProducerInterceptors<K, V> interceptors;

### 方法 ###

KafkaProducer(ProducerConfig, Serializer<K>, Serializer<V>)

*  构造方法主要作为初始化producer配置信息，K和V的序列化类。暂时不做解读，有兴趣的请看kafka源码，地址看文末。
 *  注意：producer使用完成之后需要调用close()，避免资源数据泄露。

send(ProducerRecord，Callback)

*  send方法是异步的，调用该方法会将记录数据（record）发到还没发送的记录数据（record）缓冲区，并更新record元数据，确认发送时调用提供回调方法（Callback），优点是：producer可以异步批量的发送单例的消息（record）以提高效率。
 *  返回结果RecordMetadata记录了records发送到的指定分区、偏移量、时间戳。如果Topic使用CreateTime，则时间戳将是提供给用户的时间戳，如果用户没有为记录指定时间戳，则为记录发送时间。如果对Topic使用LogAppendTime，则在添加消息时，时间戳将是Kafka broker主机本地时间。
 *  由于send调用是异步的，所以它返回将分配给该记录的RecordMetadata的未来。在此将来调用get()将阻塞，直到相关请求完成，然后返回记录的元数据，或者抛出发送记录时发生的任何异常。
 *  模拟阻塞代码

byte[] key = "key".getBytes();
     byte[] value = "value".getBytes();
     ProducerRecord<byte[],byte[]> record = new ProducerRecord<byte[],byte[]>("my-topic", key, value)
     producer.send(record).get();

*  完全非阻塞使用可以使用回调参数来提供一个回调，该回调将在请求完成时调用。

ProducerRecord<byte[],byte[]> record = new ProducerRecord<byte[],byte[]>("the-topic", key, value);
     producer.send(myRecord,
                   new Callback() {
                       public void onCompletion(RecordMetadata metadata, Exception e) {
                           if(e != null) {
                              e.printStackTrace();
                           } else {
                              System.out.println("The offset of the record we just sent is: " + metadata.offset());
                           }
                       }
                   });

*  对于发送到相同分区的记录，保证按顺序执行回调。也就是说，在下面的例子中，callback1保证在callback2之前执行。

producer.send(new ProducerRecord<byte[],byte[]>(topic, partition, key1, value1), callback1);
    producer.send(new ProducerRecord<byte[],byte[]>(topic, partition, key2, value2), callback2);

*  当作为事务的一部分使用时，不需要定义回调函数或检查将来的结果来检测发送的错误。如果任何发送调用失败，并且出现不可恢复的错误，则最后一个commitTransaction()调用将失败，并从上次失败的发送中抛出异常。发生这种情况时，应用程序应该调用abortTransaction()重置状态并继续发送数据。

参考： [KafkaProducer文档][KafkaProducer]

[KafkaProducer]: http://kafka.apache.org/22/javadoc/index.html?org/apache/kafka/clients/producer/KafkaProducer.html