生产者发送消息的流程
- 发送原理
- 生产者重要参数列表
使用 Java 程序操作 Kafka
异步发送 & 同步发送
- 带回调函数的异步发送
- 同步发送
生产者分区
=======生产经验=======
生产者如何提高吞吐量
数据可靠性
数据去重
数据有序 & 数据乱序

生产者发送消息的流程

发送原理

在消息发送的过程中，涉及到了两个线程：main 线程和 Sender 线程。
在 main 线程中创建了一个双端队列 RecordAccumulator。
main 线程将消息发送给双端队列，Sender 线程不断从双端队列中拉取消息发送到 Kafka Broker。

batch.size：数据积累到 batch.size 之后，sender 才会发送数据。batch.size 的值默认为：16k
linger.ms：如果数据迟迟未达到 batch.size，sender 等待 linger.ms 设置的时间到了之后就会发送数据。
linger.ms 的单位是：ms，默认值是：0 ms，表示没有延迟。

sender 线程发送数据的时机：要么数据积累到了 batch.size 值的大小，要么等待时间达到了 linger.ms 值。

Kafka 的应答级别：

0：生产者发送过来的数据，不需要等数据落盘就应答。
1：生产者发送过来的数据，Leader 收到数据后就应答。
-1：生产者发送过来的数据，Leader 和 ISR 队列里面的所有节点都收到数据后应答。（-1 和 all 等价）
生产者重要参数列表
| 参数名称 | 描述 | | —- | —- | | bootstrap.servers | 生产者连接集群所需的 broker 地址清单。
例如：hadoop102:9092,hadoop103:9092,hadoop104:9092，
可以设置 1 个或多个，中间用逗号隔开。
注意这里并非需要所有的 broker 地址，
因为生产者从给定的 broker 里查找到其他 broker 信息。 | | key.serializer 和 value.serializer | 指定发送消息的 key 和 value 的序列化类型。
一定要写全类名。 | | buffer.memory | 双端队列 RecordAccumulator 缓冲区总大小，默认值为：32m。 | | batch.size | 缓冲区一批数据的最大值，默认值为：16k。
适当增加该值，可以提高吞吐量，
但是如果该值设置太大，会导致数据传输延迟增加。 | | linger.ms | 如果数据未达到 batch.size，sender 等待 linger.time 后就会发送数据。
单位 ms，默认值是 0ms，表示没有延迟。
生产环境建议该值大小为 5-100ms 之间。 | | acks | 0：生产者发送过来的数据，不需要等数据落盘应答。
1：生产者发送过来的数据，Leader 收到数据后应答。
-1：生产者发送过来的数据，Leader 和 ISR 队列里面的所有节点都收到数据后应答。该参数的默认值是：-1，-1 和 all 是等价的。 | | max.in.flight.requests.per.connection | 允许最多没有返回 ack 的次数，默认值为：5，
开启幂等性要保证该值是 1-5 的数字。 | | retries | 当消息发送出现错误的时候，系统会重发消息。
retries 表示重试次数。默认值为：int 的最大值，2147483647。
如果设置了重试，还想保证消息的有序性，需要设置
MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION=1
否则在重试此失败消息的时候，其他的消息可能发送成功了。 | | retry.backoff.ms | 两次重试之间的时间间隔，默认值为：100ms。 | | enable.idempotence | 是否开启幂等性，默认值为：true，开启幂等性。 | | compression.type | 生产者发送的所有数据的压缩方式。默认值为：none，也就是不压缩。
支持的压缩类型：none、gzip、snappy、lz4 和 zstd。 |

使用 Java 程序操作 Kafka

对云服务器的操作：开放：9092 端口
配置参考博客
对 Kafka 需要修改 kafka/config/server.properties 配置文件中的 advertised.listeners 才能使得 Kafka 集群能被远程客户端访问。advertised.listeners=PLAINTEXT://<公网IP>:9092（写的时候不加 <> 尖括号）
此配置选项的作用就是：为客户端提供监听服务，如果不配置会默认设置为 localhost，但是云服务器的 IP 地址分为内外和外网，localhost 是内网地址。因此无法通过 Java 客户端访问到 Kafka 集群，因此将此配置项配置为此台云服务器的外网地址即解决问题。

添加依赖项

编写业务代码

<dependency>
 <groupId>org.apache.kafka</groupId>
 <artifactId>kafka-clients</artifactId>
 <version>3.0.0</version>
</dependency>
<dependency>
 <groupId>org.slf4j</groupId>
 <artifactId>slf4j-log4j12</artifactId>
 <version>1.7.30</version>
</dependency>

# 该配置文件必须要写，不然会报 WARN：Please initialize the log4j system properly.
log4j.rootLogger=DEBUG,Console
log4j.appender.Console=org.apache.log4j.ConsoleAppender
log4j.appender.Console.layout=org.apache.log4j.PatternLayout
log4j.appender.Console.layout.ConversionPattern=%d [%t] %-5p [%c] - %m%n

如果不添加 slf4j 的依赖，则会报如下错误，并无法显示 Kafka 程序执行过程中的日志信息

public class CustomProducer {
 public static void main(String[] args) {
     // Kafka 的配置信息
     Properties kafkaProperties = new Properties();
     // 连接的 Kafka 服务器信息、key value 的序列号方式（必要信息）
     kafkaProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "47.96.188.26:9092");
     kafkaProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
     kafkaProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
     // 创建 Kafka 生产者对象
     KafkaProducer<String, String> kafkaProducer = new KafkaProducer(kafkaProperties);
     // 调用 send() 发送消息
     String topic = "myFirstTopic";
     String value = "hello,world";
     Future<RecordMetadata> future = kafkaProducer.send(new ProducerRecord<>(topic, value));
     // 关闭资源
     kafkaProducer.close();
 }
}

异步发送 & 同步发送

以下的同步和异步是针对 main 线程将消息发送给双端队列 RecordAccumulator 来讲的。
而 sender 线程将双端队列 RecordAccumulator 中的数据发送到 Broker 总是异步的。

同步发送：发送一次数据后，必须等待 Leader 的 ACK 到达后才能发送下一个数据。
异步发送：发送一次数据后，不需要等待 Leader 的 ACK，就可以继续发送数据。
对应到 Java 程序中，同步就是：响应之后，程序才会继续执行下面的指令
异步就是：无需等待响应，程序就可以继续执行下面的指令

带回调函数的异步发送

Future<RecordMetadata> future = kafkaProducer.send(new ProducerRecord<>(topic, value));就是普通异步发送。

回调函数 onCompletion() 会在生产者收到 Leader 发的 ACK 时调用，
该 onCompletion() 方法有两个参数，分别是：元数据信息（RecordMetadata）和异常信息（Exception），

如果 Exception 为 null，说明消息发送成功
如果Exception 不为 null，说明消息发送失败。

消息发送失败会自动重试，不需要我们在回调函数中手动重试。

public class CustomProducer {
    public static void main(String[] args) {
        // Kafka 的配置信息
        Properties kafkaProperties = new Properties();
        // 连接的 Kafka 服务器信息、key value 的序列号方式（必要信息）
        kafkaProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "47.96.188.26:9092");
        kafkaProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        kafkaProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 创建 Kafka 生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer(kafkaProperties);
        // 调用 send() 发送消息
        String topic = "myFirstTopic";
        String value = "hello,world";
        Future<RecordMetadata> future = kafkaProducer.send(new ProducerRecord<>(topic, value), new Callback() {
            // 该方法在 生产者 收到 ACK 时调用，为异步调用
            // 如果 Exception 为 null，说明消息发送成功
            // 如果 Exception 不为 null，说明消息发送失败。
            @Override
            public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                if (e == null) {
                    System.out.println("消息发送成功");
                    System.out.println("topic:" + recordMetadata.topic());
                    System.out.println("分区:" + recordMetadata.partition());
                } else {
                    System.out.println("消息发送失败");
                    e.printStackTrace();
                }
            }
        });
        // 关闭资源
        kafkaProducer.close();
    }
}

同步发送

如果需要使用同步发送，可以在每次发送之后使用 get() 方法。future.get();
因为 producer.send() 返回一个 Future 类型对象，使用该对象的 get() 可以获得任务执行的结果。
Future 的 get() 会阻塞，直到该线程的任务得到返回值，也就是 Broker 返回发送结果。

生产者分区

分区的好处

便于合理使用存储资源，每个 Partition 在一个 Broker 上存储，
可以把海量的数据按照分区切割成一块一块的数据存储在多台 Broker 上。
（一个 Topic 的不同 Partition 可以分布在同一个 Broker，也可以分布在不同的 Broker 上）

合理控制分区的任务，可以实现负载均衡的效果。

提高并行度，生产者可以以分区为单位发送数据；消费者可以以分区为单位进行消费数据。

生产者发送消息的默认分区策略

默认的分区器类：DefaultPartitioner public class DefaultPartitioner implements Partitioner {}
如果我们想使用自定义的分区器，那么就可以实现 Partitioner 接口，并进行相关配置。

ProducerRecord 类用于对发送的消息和相关参数进行封装。
ProducerRecord 类有好多个构造器，默认分区策略对不同的构造器采用不同的策略。

默认分区策略对不同的构造器分为三种情况

有 partition 值的情况
没有 partition 值，但有 key 值的情况
既没有 partition 值，又没有 key 值的情况

有 partition 值的情况下，直接使用 partition 值对应的分区。
例如：partition = 0，所有数据写入分区 0

没有 partition 值，但有 key 值的情况下，将 key 的 hash 值与该 topic 的分区数进行取余得到 partition 值；
例如：key1 的 hash 值为：5， key2 的 hash 值为：6 ，topic 的分区数为：2，
那么 key1 对应的 value1 写入 1 号分区，key2 对应的 value2 写入 0 号分区。

既没有 partition 值，又没有 key 值的情况下，Kafka 采用 Sticky Partition（黏性分区器），会随机选择一个分区，并尽可能一直使用该分区，等到该分区的 batch 已满或者已完成，Kafka 再随机一个分区进行使用（保证和上一次的分区不同）。
例如：第一次随机选择 0 号分区，等到 0 号分区当前批次满了（默认为：16k）或者 linger.ms 设置的等待时间到了， Kafka 再随机一个分区进行使用（如果还是 0 会继续随机，即保证和上一次的分区不同）。

自定义分区器

自定义分区器的实现步骤

定义一个类，该类实现 Partitioner 接口
重写 partition() 方法

发送消息时，使用自定义分区器的方法：在生产者的配置中添加分区器参数。
properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, MyPartitioner.class.getName());

public class MyPartitioner implements Partitioner {
    /**
     * 返回信息对应的分区
     *
     * @param topic      主题
     * @param key        消息的 key
     * @param keyBytes   消息的 key 序列化后的字节数组
     * @param value      消息的 value
     * @param valueBytes 消息的 value 序列化后的字节数组
     * @param cluster    集群元数据可以查看分区信息
     * @return
     */
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        // 实现自己的分区策略
        // 返回数据写入的分区号
        return 0;
    }
    // 关闭资源
    @Override
    public void close() {
    }
    // 配置方法
    @Override
    public void configure(Map<String, ?> configs) {
    }
}

=======生产经验=======

以下都为生产经验。

生产者如何提高吞吐量

修改相关的参数：
batch.size：批次大小，默认为：16k
linger.ms：等待时间，修改为：5-100ms
compression.type：压缩方式，修改为：snappy，可配置值：gzip、snappy、lz4 和 zstd
RecordAccumulator：缓冲区大小（双端队列），修改为：64m

// batch.size： 批次大小，默认为：16K
kafkaProperties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
// linger.ms：等待时间，默认为：0 ms
kafkaProperties.put(ProducerConfig.LINGER_MS_CONFIG, 1);
// RecordAccumulator：缓冲区大小，默认 32M：buffer.memory
kafkaProperties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
// compression.type：压缩，默认为：none，无压缩
kafkaProperties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");

数据可靠性

可以为 Kafka 集群设置不同的应答级别

0：生产者发送过来的数据，不需要等数据落盘就应答。
1：生产者发送过来的数据，Leader 收到数据后就应答。
-1：生产者发送过来的数据，Leader 和 ISR 队列里面的所有节点都收到数据后应答。（-1 和 all 等价）

当应答级别为 0 时，如果应答完成后，数据还没落盘，Broker 就故障了，则消息丢失，不可靠
当应答级别为 1 时，如果应答完成后，还没开始同步副本，Leader 就挂了，切换 Leader 后，消息丢失，不可靠

当应答级别为 -1 时，思考：Leader 收到数据，所有 Follower 开始同步数据，但是有一个 Follower，因为某种故障，迟迟不能与 Leader 进行同步，那这个问题该怎么解决呢？

Leader 维护了一个动态的 in-sync replica set（ISR），
意为和 Leader 保持同步的 Follower + Leader 集合 (leader：0，isr：0, 1, 2)。

如果 Follower 长时间未向 Leader 发送通信请求或同步数据，则该 Follower 将被踢出 ISR。
该时间阈值由 replica.lag.time.max.ms 参数设定，默认为：30s。例如 2 超时，(leader：0, isr：0, 1)。
这样就不用等长期联系不上或者已经故障的节点，直接应答 ACK 即可。

数据可靠性分析
如果分区副本设置为 1 个，或者 ISR 里应答的最小副本数量（ min.insync.replicas 默认为：1）设置为 1，和ack=1 的效果是一样的，仍然有丢数的风险（leader：0，isr：0）。

数据完全可靠的条件 = ACK级别设置为-1 + 分区副本数≥2 + ISR 里应答的最小副本数≥2

可靠性总结：

acks = 0，生产者发送过来数据就不管了，可靠性差，效率高
acks = 1，生产者发送过来数据 Leader 应答，可靠性中等，效率中等；
acks = -1，生产者发送过来数据 Leader 和 ISR 队列里面所有 Follwer 应答，可靠性高，效率低；

在生产环境中，acks=0 很少使用；
acks=1，一般用于传输普通日志，允许丢个别数据；
acks=-1，一般用于传输和钱相关的数据，对可靠性要求比较高的场景。

数据去重

设置应答级别为 -1 时，生产者发送过来的数据，Leader 和 ISR 队列里面的所有节点都收到数据后才应答。

如果已经开始同步副本，并且已经有分区副本完成了同步，此时 Leader 挂了，切换 Leader 后，生产者会重发消息，这时就会有一些副本里存在两份相同的数据，导致数据重复。

如何解决数据重复的问题呢？

数据传递语义

至少一次（At Least Once）= ACK级别设置为-1 + 分区副本数≥2 + ISR里应答的最小副本数≥2
最多一次（At Most Once）= ACK级别设置为0
精确一次（Exactly Once）：对于一些非常重要的信息，比如和钱相关的数据，要求数据既不能重复也不丢失

对至少一次 & 最多一次的总结：

至少一次（At Least Once）可以保证数据不丢失，但是不能保证数据不重复
最多一次（At Most Once）可以保证数据不重复，但是不能保证数据不丢失

Kafka 0.11 版本以后，引入了重大特性：幂等性和事务。

幂等性

幂等性就是指：生产者不论向 Broker 发送多少次重复数据，Broker 端都只会持久化一次，保证数据不重复。
精确一次（Exactly Once） = 幂等性 + 至少一次

重复数据的判断标准：具有相同主键的消息提交时，Broker 只会持久化一次。

PID 是 Kafka 每次重启都会分配一个新的；（Producer Id，生产者ID）
Partition 表示分区号；
Sequence Number 是单调自增的。

所以幂等性只能保证：在单分区单会话内不重复。

开启幂等性，需要设置参数：enable.idempotence，默认为：true，即默认开启幂等性。

生产者事务

开启事务，必须开启幂等性。

生产者在使用事务功能前，必须先自定义一个唯一的 transactional.id。（事务 ID）
有了 transactional.id 即使客户端挂掉了，客户端重启后也能继续处理未完成的事务

单个生产者，使用事务保证消息的仅一次发送。步骤：

设置事务 ID，事务 ID 任意起名
初始化事务
开启事务

提交事务 / 终止事务

public class CustomProducer {
 public static void main(String[] args) {
     // Kafka 的配置信息
     Properties kafkaProperties = new Properties();
     // 连接的 Kafka 服务器信息、key value 的序列号方式（必要信息）
     kafkaProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "47.96.188.26:9092");
     kafkaProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
     kafkaProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
     // 设置事务 ID，事务 ID 任意起名
     kafkaProperties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transaction_id_0");
     // 创建 Kafka 生产者对象
     KafkaProducer<String, String> kafkaProducer = new KafkaProducer(kafkaProperties);
     // 初始化事务
     kafkaProducer.initTransactions();
     // 开启事务
     kafkaProducer.beginTransaction();
     try {
         // 调用 send() 发送消息
         String topic = "myFirstTopic";
         String value = "hello,world1";
         for (int i = 0; i < 5; i++) {
             kafkaProducer.send(new ProducerRecord<>(topic, value));
         }
         // 手动模拟异常
         int a = 10 / 0;
         // 提交事务
         kafkaProducer.commitTransaction();
     } catch (Exception e) {
         // 出现异常, 放弃事务（类似于回滚事务的操作）
         kafkaProducer.abortTransaction();
         e.printStackTrace();
     } finally {
         // 关闭资源
         kafkaProducer.close();
     }
 }
}

数据有序 & 数据乱序

单分区内有序（满足一定条件，才能保证单分区内有序）；
多分区、分区与分区间无序；

kafka 在 1.x 版本之前保证数据单分区内有序，需要满足的条件为：必须设置max.in.flight.requests.per.connection=1（不需要考虑是否开启幂等性）。

kafka 在 1.x 及以后版本保证数据单分区内有序，需要满足的条件如下：

如果未开启幂等性：必须设置max.in.flight.requests.per.connection=1
如果开启了幂等性：max.in.flight.requests.per.connection 需要设置 ≤ 5

需要设置 ≤ 5 的原因说明：因为在 kafka1.x 以后，开启幂等后，kafka 服务端会缓存生产者发送的最近 5 个
写入请求（request）的元数据，故无论如何，都可以保证最近的 5 个 request 的数据都是有序的。

Kafka 生产者