【Kafka】01 生产者

0. 目录

1. 生产消息步骤

• 配置生产者客户端参数，创建生产者实例
• 构建待发送的消息
• 发送消息
• 关闭生产者

2. 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;
}

key用来计算分区号，确定发送到指定分区

3. 必要的参数

• bootstrap.servers：指定kafka集群的broker的地址，默认为空，虽然可以从给定的broker找到其他broker，但是为防止某一节点宕机导致消息发送失败，建议填写至少2个broker地址。
• key.serializer、value.serializer
• client.id：客户端ID，若不设置，会自动生成非空字符串。
• 设置技巧：
  属性名不好记，可以使用ProducerConfig类中的常量。

4. 发送消息

• 发后即忘（fire-and-foget）
  性能最高，可靠性最差

try {
    producer.send(record);
} catch (Exception e) {
    e.printStackTrace();
}

• 同步（sync）
  同步发送的性能差，需要阻塞等待一条消息发送完成之后，再去发送下一条。
  send方法本身是异步的，通过使用get方法返回的Future对象实现同步。

try {
    producer.send(record).get();
} catch (Exception e | InterruptedException e) {
    e.printStackTrace();
}

如果需要Future.get()的返回值，可以采用以下方式。返回值中包含消息的主题、分区号、偏移量、时间戳等元数据信息。

try {
    Future<RecordMetadata> future = producer.send(record);
    RecordMetadata metadata = future.get();
    System.out.println(metadata.topic() + "-" + 
        metadata.partition() + "-" + metadata.offset());
} catch (Exception e | InterruptedException e) {
    e.printStackTrace();
}

由于send()方法返回的是Future对象，可以使用java并发的相关方法丰富实现。
同步发送可以配置客户端参数的重试次数，如果超过重试次数，则必须处理异常。

• 异步（async）
  Future send(ProducerRecord<K, V> record, Callback callback);

producer.send(record, new Callback() {
    @Override
    public void onCompletion(RecordMetadata metadata, Exception exception) {
        if (exception != null) {
            exception.printStackTrace();
        } else {
            System.out.println(metadata.topic() + "-" + 
                metadata.partition() + "-" + metadata.offset());
        }
    }
})

kafka消息发送保证分区有序的情况下，回调函数也能保证有序。

producer.close();
public void close(long timeout, TimeUnit timeUnit); //等待超时时间之后，强行退出。

close()方法会阻塞等待所有的消息都发送完成再关闭。

5. 分区器

• 消息体重指定partition字段，根据该字段发送到对应分区
• 没有指定分区字段，根据消息体中key字段，由分区器分配分区
  kafa使用的分区器即客户端的DefaultPartitioner类

DefaultPartitioner implements Partitioner {
    public int partition(...);  //计算分区号
    public void close();    //关闭分区器时释放相应资源，空方法
}

• 默认分区器：
  （1）如果key不为null，对key进行哈希得到分区号（所有分区中的任意一个，不论是否可用）。
  （2）key为null，轮询发送到各个可用分区，如果没有可用分区，则轮询到各个分区。
• 如何自定义分区器？
  （1）实现Partitioner接口，重写partition()方法
  （2）配置客户端的配置参数

props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, DemoPartitinoer.class.getName());

6. 拦截器

• 作用：
  在发送之前做准备工作（eg：过滤消息，修改部分内容）
  在发送的回调逻辑之前，做一些定制化的需求（eg：统计工作）

• 自定义拦截器
  （1）实现ProducerInterceptor<String, String>接口，重写onsend()方法，重写onAcknowledgement()方法
  （2）配置客户端的配置参数：ProducerConfig.INTECERPTOR_CLASSES_CONFIG
  可以指定多个，类名中间以“,”连接，多个拦截器时，某一个失败，会从上一个成功的开始接着执行下一个拦截器

  7. 整体架构

  主线程：KafkaProducer-->拦截器（非必须，一般不）-->序列号器（必须）-->分区器（根据key可选）-->消息累加器-->Sender线程
  Sender线程：Sender-->创建Request-->提交给Selector发送-->未收到响应的请求根据节点分别放在节点对应的InFlightRequests中。
  Sender线程，先将分区信息转换为节点信息，做应用逻辑层到网络IO层的转换，然后将消息转换成满足kafka协议的Request。
  

可以通过比较InFlightRequests的大小的配置参数max.in.flight.requests.per.connection与在InFlightRequests中的消息多少，来判断各个节点的负载情况。

元数据（主题分区的数量、leader副本所在节点的地址、端口等信息）的更新由Sender线程，选择负载最小的节点进行发送MetaDataRequest获取。默认每5分钟更新一次。
主线程也需要使用Sender更新过的元数据。

8. 几个重要生产者参数

• acks --> 什么情况下算发送成功
  默认值：1，只要leader写入成功即成功。是可靠性和吞吐量之间的折中方案。
  0，发送即成功，不管服务端是否返回相应。吞吐量最大。
  -1或all，所有ISR中的副本都成功。可靠性最高，但不保证一定可靠，因为ISR中可能只有leader，此时的效果与1相同。
• max.request.size
  生产者能发送的消息的最大值，默认1MB

9. 生产者是线程安全的

end ：）

【Kafka】01 生产者

原文：https://www.cnblogs.com/suyeSean/p/11241900.html