Kafka java客户端数据生产流程解析
ProducerRecord
ProducerRecord 含义: 发送给Kafka Broker的key/value 值对
//ProducerRecord的成员变量
public class ProducerRecord<K, V> {
private final String topic;//主题
private final Integer partition;//分区号
private final Headers headers;//消息头
private final K key;//键
private final V value;//值
private final Long timestamp;//消息时间戳
headers：可以设定一些与应用相关的信息
KafkaProducer
KafkaProducer是线程安全的，可以在多个线程中共享单个KafkaProducer实例，也可以将它实例进行池化来统一管理。
KafkaProducer的参数众多，bootstrap.servers，retries，key.serializer等，开发人员很难记住所有的配置，也很容易写错，所以可以用org.apache.kafka.clients.producer.ProducerConfig类预防：
Properties properties = new Properties();
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
properties.put(ProducerConfig.RETRIES_CONFIG, "10");
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.1.51:9092");
KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
消息的发送
发后即忘
producer.send(record);
上述这种方式就是发后即忘，它只往kafka中发送消息并不关心消息是否正确到达，可能会造成消息丢失，发送性能最高，但是可靠性最差。
同步
@Override
public Future<RecordMetadata> send(ProducerRecord<K, V> record) {
return send(record, null);
}
因为直接send返回值是Future，我们知道Future.get()会阻塞线程直至线程运行结果返回，通过此方法即可达到同步目的。
try {
producer.send(record).get();
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
Future中可以获取一个RecordMetadata对象，包含了消息大的一些元数据信息，比如主题、分区号、分区中的偏移量(offset)、时间戳等。
此外Future.get(long timeout, TimeUnit unit)可以实现可超时的阻塞，
异步
producer.send(record, (metadata, exception) -> {
System.out.println(metadata.topic());
});
当kafka有响应时，会触发回调方法
序列化器
生产者需要用序列化器把对象转换成字节数组才能通过网络发送给kafka。而在消费者需要用反序列化器把字节数组转成相应对象。
序列化器需要实现Serializer接口。客户端实现了多种序列化器供我们开发使用。
public interface Serializer<T> extends Closeable {
/**
* Configure this class.
* @param configs configs in key/value pairs
* @param isKey whether is for key or value
*/
//用来配置当前类
void configure(Map<String, ?> configs, boolean isKey);
/**
* Convert {@code data} into a byte array.
*
* @param topic topic associated with data
* @param data typed data
* @return serialized bytes
*/
//序列化操作
byte[] serialize(String topic, T data);
/**
* Close this serializer.
*
* This method must be idempotent as it may be called multiple times.
*/
@Override
//关闭当前序列化器
void close();
}
生产者和消费者使用的序列化器是需要一一对应的。例如生产者shencghan端使用了StringSerializer，那么消费者端需要使用StringDeserializer。
分区器
消息在通过send()方法发往broker过程中，有可能经过拦截器、序列化器、分区器的一系列作用后才能真正发送到broker。拦截器一般不是必需的，而序列化器时必需的。消息经过序列化后就需要确定它发往的分区。
如果ProducerRecord 中指定了partition字段，那么就不需要分区器的作用。如果没有指定，那么就需要依赖分区器，根据key这个字段来计算partition的值。分区器的作用就是为消息分配分区。
kafka默认分区器是DefaultPartitioner，实现了Partitioner接口
public interface Partitioner extends Configurable, Closeable {
/**
* Compute the partition for the given record.
*
* @param topic The topic name 主题名
* @param key The key to partition on (or null if no key) 键
* @param keyBytes The serialized key to partition on( or null if no key) 序列化后的键
* @param value The value to partition on or null 值
* @param valueBytes The serialized value to partition on or null 序列化后的值
* @param cluster The current cluster metadata 集群元数据信息
*/
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster);
/**
* This is called when partitioner is closed.
*/
public void close();
}
Partitioner的父接口Configurable
public interface Configurable {
/**
* Configure this class with the given key-value pairs
*/
//该方法主要用来获取配置信息以及配置初始化数据
void configure(Map<String, ?> configs);
}
在默认DefaultPartitioner中的分区计算方法：
我们也可以自定义分区器，然后配置在properties:
properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"....");
生产者拦截器
实现生产者拦截器只需要实现ProducerInterceptor接口
public interface ProducerInterceptor<K, V> extends Configurable {
public ProducerRecord<K, V> onSend(ProducerRecord<K, V> record);
public void onAcknowledgement(RecordMetadata metadata, Exception exception);
public void close();
}
onSend方法调用时机：KafkaProducer在将消息序列化和计算分区之前会调用拦截器的onSend()方法来对消息进行相应的定制化操作，一般来说不要修改ProducerRecord 的topic、key、partition信息。因为可能会影响分区的计算，同样会影响broker端日志压缩。
onAcknowledgement方法调用时机：该方法会在消息被应答之前或消息发送失败时调用，优先于用户设定的Callback之前执行。
close：主要用于在关闭拦截器时执行一些资源的清理工作
自定义生产者拦截器：
public class MyProducerInterceptor implements ProducerInterceptor {
@Override
public ProducerRecord onSend(ProducerRecord record) {
System.out.println("onSend");
return record;
}
@Override
public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
System.out.println("onAcknowledgement");
}
@Override
public void close() {
System.out.println("close");
}
@Override
public void configure(Map<String, ?> configs) {
}
}
//设置properties参数
properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, MyProducerInterceptor.class.getName());
此外可以指定多个拦截器，形成拦截器链，拦截器链会按照配置的拦截器顺序来一一执行（各个拦截器之间用逗号隔开）
properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, MyProducerInterceptor.class.getName()+","+MyProducerInterceptor2.class.getName());
原理分析
整个生产者客户端由两个线程协调运行，分别为主线程和Sender线程(发送线程)
RecordAccumulator：消息累加器，主要用来缓存消息以便Sender线程可以批量发送。RecordAccumulator缓存大小可以通过生产者客户端参数buffer.memory配置，默认32MB.如果生产者发送消息速度超过发送到服务器速度，会导致生产者空间不足，这时候，send方法会要么被阻塞要么抛出异常，取决于max.block.ms配置，默认60000，即60s。
主线程发送过来的消息会被追加到RecordAccumulator的某个双端队列中Deque，即Deque<ProducerBatch>。ProducerBatch中包含一个至多个ProducerRecord。将较小的ProducerRecord拼凑成较大的ProducerBatch，可以减少网络请求的次数以提升整体的吞吐量。ProducerBatch的大小跟batch.size有关。
Sender从RecordAccumulator中获取到缓存的消息之后，会将原本<分区,Deque<ProducerBatch>>的保存形式转换为<Node,List<ProducerBatch>>，Node表示Kafka的broker节点。
转换为<Node,List<ProducerBatch>>之后，会进一步<Node，Request>的形式，这样就可以将Request发往各个Node了。Request表示Kafka的各种协议请求。
请求在发往kafka之前还会保存到InFlightRequests中，它的具体形式为Map<NodeId,Deque<Request>>，它的作用是缓存了已经发出去但还没有收到相应的请求。配置参数：max.in.flight.requests.per.connection，默认值是5，即每个连接最多只能缓存5个未响应的请求，超过该数目就不能向这个连接发送更多请求了。
重要的生产者参数
以下仅介绍部分参数，还有一些高级功能（事务、幂等）这里不做介绍。
acks
这个参数用来指定分区中必须要有多少个副本收到这条消息，之后生产者者才会认为这条消息是成功写入的。
acks的值都是字符串类型的。
acks=1，默认值，生产者发送消息之后，只要分区的leader副本成功写入消息，那么它就会收到来自服务端的成功响应。acks设置为1，是消息可靠性和吞吐量之间的这种方案。
acks=0，生产者发送消息之后不需要等待任何服务端的响应。
acks=-1或acks=all，生产者在消息发送之后，需要等待ISR中的所有副本都成功写入消息之后才能够来自服务端的成功响应。acks=-1可以达到最强的可靠性。
max.request.size
这个参数用来限制生产者客户端能发送的消息的最大值，默认1MB。这个参数在对kafka整体脉络没有把控的时候，不建议修改，因为这个参数还涉及到其他参数的修改。比如broker端的message.max.bytes参数。
retries和retry.backoff.ms
retries参数用来配置生产者重试的次数，默认值是0
retry.backoff.ms用来设置两次重试之间的时间间隔，避免无效的频繁重试，默认值是100ms。
connections.max.ide.ms
这个参数用来指定在多久之后关闭闲置的连接，默认是9分钟
linger.mx
指定生产者发送ProducerBatch之前等待更多消息加入ProducerBatch的时间，默认为0.
request.timeout.ms
用来配置Producer等待请求响应的最长时间，默认时间3000ms。请求超时后可以选择重试。
receive.buffer.bytes
设置socket接收消息缓冲区（SO_RECBUF）的大小，默认值32768B，即32KB，如果设置为-1，则为操作系统默认值。
send.buffer.bytes
设置socket发送消息缓冲区（SO_RECBUF）的大小，默认值131072B，即128KB，如果设置为-1，则为操作系统默认值。