1 初始化配置

  Kafka 通过 KafkaProducer 构造器初始化生产者客户端的配置。
  常用的重要配置，详见官网。

• bootstrap.servers：Kafka 集群地址（host1:post,host2:post），Kafka 客户端初始化时会自动发现地址，所以可以不填写所有地址。
• key.serializer：实现了 Kafka 序列化接口的类，用来序列化 key。
• value.serializer：实现了 Kafka 序列化接口的类，用来序列化 value。
• acks：leader 接收到的 follower 确认的数量需要满足 acks 的配置。
   0：生产者把消息发送出去就认为发送完成了。
   1：leader 接收到消息后，不用等 follower 的确认，就表示发送完成了。
   all/-1：leader 接收到消息后，需要所有在 ISR 集合的 follower 确认后，才表示完成了。
• retries：消息发送失败后的重试次数。如果允许重试，而 max.in.flight.requests.per.connection>1，则可能导致消息乱序，因为如果把两批消息发送到同一个分区，第一批失败并重试，而第二批成功了，则第二批消息可能先生成了。
• retry.backoff.ms：消息重试发送的间隔。
• client.id：标识客户端的 id。
• compression.type：压缩类型。可选：none、gzip、snappy、lz4。
• buffer.memory：记录累加器可以使用的最大内存缓冲池大小。
• batch.size：内存缓冲池的缓冲列表大小。当 batch 的大小超过 batch.size 或者时间达到 linger.ms 就会发送 batch。
• transactional.id：事务 ID。

// 基础配置
Map<String, Object> configs = new HashMap<>();
// Kafka broker 集群
configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092,127.0.0.1:9093,127.0.0.1:9094");
// key 序列化
configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
// value 序列化
configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);KafkaProducer<String, String> producer = new KafkaProducer<>(configs);

2 构造消息

  Kafka 提供了6种构造器来构造消息。

• topic：消息主题，必填；
• partition：分区号，非必填。如果为空，会计算 key 的 hash 值，再和该主题的分区总数取余得到分区号；如果 key 也为空，客户端会生成递增的随机整数，再和该主题的分区总数区域得到分区号。
• timestamp：时间戳，非必填。如果为空，默认为 KafkaProducer 构造器初始化的时间。
• key：消息 key，非必填。关系到分区分配，broker 会对带 key 的消息进行日志压缩。
• value：消息内容，必填。
• headers：消息头，非必填。

public ProducerRecord(String topic, Integer partition, Long timestamp, K key, V value, Iterable<Header> headers);
public ProducerRecord(String topic, Integer partition, Long timestamp, K key, V value);
public ProducerRecord(String topic, Integer partition, K key, V value, Iterable<Header> headers);
public ProducerRecord(String topic, Integer partition, K key, V value);
public ProducerRecord(String topic, K key, V value);
public ProducerRecord(String topic, V value);

3 发送消息

  支持同步发送和异步发送消息。

  同步发送

producer.send(record).get();

  异步发送

producer.send(record, new Callback() {@Overridepublic void onCompletion(RecordMetadata metadata, Exception exception) {// 回调处理流程}
});