Kafka的生产者（三）_producer clientid=producer-1 bootstrap broker disc

本文为博主自学笔记整理，内容来源于互联网，如有侵权，请联系删除。

个人笔记：https://github.com/dbses/TechNotes

01 | 生产消息的分区机制

在使用 Apache Kafka 生产和消费消息的时候，如何将大的数据量均匀地分配到 Kafka 的各个 Broker 上？

今天就来说说 Kafka 生产者如何实现这个需求。

Kafka 的消息组织方式实际上是三级结构：主题 - 分区 - 消息。主题下的每条消息只会保存在某一个分区中，而不会在多个分区中被保存多份。官网上的这张图非常清晰地展示了 Kafka 的三级结构，如下所示：

为什么要分区？

Kafka 为什么使用分区的概念而不是直接使用多个主题呢？

• 提供负载均衡的能力

• 实现系统的高伸缩性（Scalability）

  不同的分区能够被放置到不同节点的机器上，而数据的读写操作也都是针对分区这个粒度而进行的，这样每个节点的机器都能独立地执行各自分区的读写请求处理。

• 增加整体系统的吞吐量

  我们还可以通过添加新的节点机器来增加整体系统的吞吐量。

除了提供负载均衡这种最核心的功能之外，利用分区也可以实现其他一些业务级别的需求，比如实现业务级别的消息顺序的问题。

分区策略都有哪些？

• 自定义分区策略

  在编写生产者程序时，编写一个具体的类实现org.apache.kafka.clients.producer.Partitioner接口。这个接口定义了两个方法：partition() 和 close()。

  int partition(String topic, 
                Object key, 
                byte[] keyBytes, 
                Object value, 
                byte[] valueBytes, 
                Cluster cluster);
  1
  2
  3
  4
  5
  6

• 轮询策略

  也称 Round-robin 策略，即顺序分配。
  
  轮询策略是 Kafka Java 生产者 API 默认提供的分区策略。

• 随机策略

  也称 Randomness 策略。所谓随机就是我们随意地将消息放置到任意一个分区上，如下面这张图所示。
  
  实现随机策略版的 partition 方法如下：

  int partition(String topic, Object key, byte[] keyBytes, 
                Object value, byte[] valueBytes, Cluster cluster) {
      // 计算出该主题总的分区数
      List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
      return ThreadLocalRandom.current().nextInt(partitions.size());
  }
  1
  2
  3
  4
  5
  6

  随机策略是老版本生产者使用的分区策略。

• 按消息键保序策略

  也称 Key-ordering 策略。Kafka 允许为每条消息定义消息键，简称为 Key。一旦消息被定义了 Key，那么就可以保证同一个 Key 的所有消息都进入到相同的分区里面，由于每个分区下的消息处理都是有顺序的，故这个策略被称为按消息键保序策略，如下图所示。
  
  实现这个策略的 partition 方法如下：

  int partition(String topic, Object key, byte[] keyBytes, 
                Object value, byte[] valueBytes, Cluster cluster) {
      // 计算出该主题总的分区数
      List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
      return Math.abs(key.hashCode()) % partitions.size();
  }
  1
  2
  3
  4
  5
  6

• 其他分区策略

  针对大规模的 Kafka 集群，可以采用按地理位置的分区策略。假如这个集群中必然有一部分机器在北京，另外一部分机器在广州。

  如果你需要把南北方注册用户的注册消息正确地发送到位于南北方的不同机房中，实现如下：

  int partition(String topic, Object key, byte[] keyBytes, 
                Object value, byte[] valueBytes, Cluster cluster) {
      // 计算出该主题总的分区数
      List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
      return partitions.stream()
        .filter(p -> isSouth(p.leader().host()))
        .map(PartitionInfo::partition)
        .findAny()
        .get();
  }
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10

  我们可以从所有分区中找出那些 Leader 副本在南方的所有分区，然后随机挑选一个进行消息发送。

02 | 生产者压缩算法

压缩（compression），具体来说就是用 CPU 时间去换磁盘空间或网络 I/O 传输量。

怎么压缩？

Kafka 的消息层次分为两层：消息集合（message set）以及消息（message）。一个消息集合中包含若干条日志项（record item），而日志项才是真正封装消息的地方。Kafka 底层的消息日志由一系列消息集合日志项组成。Kafka 通常不会直接操作具体的一条条消息，它总是在消息集合这个层面上进行写入操作。

目前 Kafka 共有两大类消息模式，分别为 V1 版本和 V2 版本。V2 版本是 Kafka 0.11.0.0 中正式引入的，把消息的公共部分抽取出来放到外层消息集合里面，这样就不用每条消息都保存这些信息了。

V2 版本还有一个和压缩息息相关的改进，就是保存压缩消息的方法发生了变化。之前 V1 版本中保存压缩消息的方法是把多条消息进行压缩然后保存到外层消息的消息体字段中；而 V2 版本的做法是对整个消息集合进行压缩。

何时压缩？

压缩可能发生在两个地方：生产者端和 Broker 端。

生产者程序中配置 compression.type 参数即表示启用指定类型的压缩算法。比如下面这段程序代码展示了如何构建一个开启 GZIP 的 Producer 对象：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("acks", "all");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
// 开启 GZIP 压缩
props.put("compression.type", "gzip");

Producer<String, String> producer = new KafkaProducer<>(props);
1
2
3
4
5
6
7
8
9

大部分情况下 Broker 从 Producer 端接收到消息后仅仅是原封不动地保存而不会对其进行任何修改。但有两种例外情况就可能让 Broker 重新压缩消息。

情况一：Broker 端指定了和 Producer 端不同的压缩算法。

如果 Broker 端设置了与 Producer 端不同的 compression.type 值，就会发生压缩/解压缩操作。这个参数的默认值是 producer，表示使用 Producer 端的压缩算法。

情况二：Broker 端发生了消息格式转换。

为了兼容老版本的格式，Broker 端会对新版本消息执行向老版本格式的转换。这个过程中会涉及消息的解压缩和重新压缩。

何时解压缩？

通常来说解压缩发生在消费者程序中。那 Consumer 怎么知道这些消息是用何种压缩算法压缩的呢？

Kafka 会将启用了哪种压缩算法封装进消息集合中，这样当 Consumer 读取到消息集合时，它自然就知道了这些消息使用的是哪种压缩算法。

除了在 Consumer 端解压缩，Broker 端也会进行解压缩，目的就是为了对消息执行各种验证。

各种压缩算法对比

在 Kafka 2.1.0 版本之前，Kafka 支持 3 种压缩算法：GZIP、Snappy 和 LZ4。

从 2.1.0 开始，Kafka 正式支持 Zstandard 算法（简写为 zstd）。它是 Facebook 开源的一个压缩算法，能够提供超高的压缩比（compression ratio）。

衡量压缩算法有两个重要的指标：一个指标是压缩比；另一个指标就是压缩 / 解压缩吞吐量。

压缩比：原先占 100 份空间的东西经压缩之后变成了占 20 份空间，那么压缩比就是 5，压缩比越高越好；

吞吐量：比如每秒能压缩或解压缩多少 MB 的数据。吞吐量也是越高越好；

下面这张表是 Facebook Zstandard 官网提供的一份压缩算法 benchmark 比较结果：

吞吐量方面：LZ4 > Snappy > zstd 和 GZIP；

压缩比方面，zstd > LZ4 > GZIP > Snappy。

（综合来看，使用 LZ4 算法效果会更好）

最佳实践

何时启用压缩是比较合适的时机呢？

启用压缩的一个条件就是 Producer 程序运行机器上的 CPU 资源要很充足。如果 Producer 运行机器本身 CPU 已经消耗殆尽了，那么启用消息压缩无疑是雪上加霜，只会适得其反。

如果 CPU 资源充足，且你的环境中带宽资源有限，那么我建议你开启压缩。

一旦启用压缩，解压缩是不可避免的事情。但要尽量规避掉那些意料之外的解压缩，比如兼容老版本消息格式而引起的解压缩。

03 | Java生产者何时创建/关闭TCP连接？

Apache Kafka 的所有通信都是基于 TCP 的，而不是基于 HTTP 或其他协议。无论是生产者、消费者，还是 Broker 之间的通信都是如此。

为何采用 TCP?

从社区的角度来看，在开发客户端时，人们能够利用 TCP 本身提供的一些高级功能，比如多路复用请求以及同时轮询多个连接的能力。

除了 TCP 提供的这些高级功能有可能被 Kafka 客户端的开发人员使用之外，社区还发现，目前已知的 HTTP 库在很多编程语言中都略显简陋。（这个理由有些牵强）

基于这两个原因，Kafka 社区决定采用 TCP 协议作为所有请求通信的底层协议。

Kafka 生产消息的过程

通常我们开发一个生产者的步骤有 4 步。

// 第 1 步：构造生产者对象所需的参数对象。
Properties props = new Properties ();
props.put("参数 1", "参数 1 的值");
props.put("参数 2", "参数 2 的值");
// ……
// 第 2 步：创建 KafkaProducer 对象实例。
try (Producer<String, String> producer = new KafkaProducer<>(props)) {
    // 第 3 步：发送消息。
    producer.send(new ProducerRecord<String, String>(……), callback);
    // ……
}
// 第 4 步：关闭生产者并释放各种系统资源。
1
2
3
4
5
6
7
8
9
10
11
12

当我们开发一个 Producer 应用时，生产者会向 Kafka 集群中指定的主题（Topic）发送消息，这必然涉及与 Kafka Broker 创建 TCP 连接。那么，Kafka 的 Producer 客户端是如何管理这些 TCP 连接的呢？

要回答这个问题，我们首先要弄明白生产者代码是什么时候创建 TCP 连接的。

何时创建 TCP 连接？

就上面的代码而言，可能创建 TCP 连接的地方有 2 和 3 处。

实际上在创建 KafkaProducer 实例时，Producer 应用就开始创建与 Broker 的 TCP 连接了。调用消息发送方法后，Producer 向某一台 Broker 发送 METADATA 请求。

为了说明这一点，测试环境中我为 bootstrap.servers 配置了 localhost:9092、localhost:9093 来模拟不同的 Broker。下面是测试环境的日志：

// 创建与 Broker 的连接
[2018-12-09 09:35:45,620] DEBUG [Producer clientId=producer-1] Initialize connection to node localhost:9093 (id: -2 rack: null) for sending metadata request (org.apache.kafka.clients.NetworkClient:1084)
[2018-12-09 09:35:45,622] DEBUG [Producer clientId=producer-1] Initiating connection to node localhost:9093 (id: -2 rack: null) using address localhost/127.0.0.1 (org.apache.kafka.clients.NetworkClient:914)
1
2
3

// 创建与 Broker 的连接
[2018-12-09 09:35:45,814] DEBUG [Producer clientId=producer-1] Initialize connection to node localhost:9092 (id: -1 rack: null) for sending metadata request (org.apache.kafka.clients.NetworkClient:1084)
[2018-12-09 09:35:45,815] DEBUG [Producer clientId=producer-1] Initiating connection to node localhost:9092 (id: -1 rack: null) using address localhost/127.0.0.1 (org.apache.kafka.clients.NetworkClient:914)
1
2
3

// 尝试获取集群的元数据信息
[2018-12-09 09:35:45,828] DEBUG [Producer clientId=producer-1] Sending metadata request (type=MetadataRequest, topics=) to node localhost:9093 (id: -2 rack: null) (org.apache.kafka.clients.NetworkClient:1068)
1
2

Producer 会连接 bootstrap.servers 参数指定的所有 Broker。

bootstrap.servers 参数：指定了这个 Producer 启动时要连接的 Broker 地址。如果你为这个参数指定了 1000 个 Broker 连接信息，你的 Producer 启动时会首先创建与这 1000 个 Broker 的 TCP 连接。

在实际使用过程中，通常你指定 3～4 台就足以了。因为 Producer 一旦连接到集群中的任一台 Broker，就能拿到整个集群的 Broker 信息，故没必要为 bootstrap.servers 指定所有的 Broker。

目前我们的结论是这样的：TCP 连接是在创建 KafkaProducer 实例时建立的。

TCP 连接还可能在两个地方被创建：一个是在更新元数据后，另一个是在消息发送时。当 Producer 更新了集群的元数据信息之后，如果发现与某些 Broker 当前没有连接，那么它就会创建一个 TCP 连接。同样地，当要发送消息时，Producer 发现尚不存在与目标 Broker 的连接，也会创建一个。

为什么说是可能？因为这两个地方并非总是创建 TCP 连接。

（总结：创建 TCP 连接有 3 个时机）

接下来，我们来看看 Producer 更新集群元数据信息的两个场景。

场景一：当 Producer 尝试给一个不存在的主题发送消息时，Broker 会告诉 Producer 说这个主题不存在。此时 Producer 会发送 METADATA 请求给 Kafka 集群，去尝试获取最新的元数据信息。

场景二：Producer 通过 metadata.max.age.ms 参数定期地去更新元数据信息。该参数的默认值是 300000 (ms)，即 5 分钟，也就是说不管集群那边是否有变化，Producer 每 5 分钟都会强制刷新一次元数据以保证它是最及时的数据。

何时关闭 TCP 连接？

Producer 端关闭 TCP 连接的方式有两种：一种是用户主动关闭；一种是 Kafka 自动关闭。

先说第一种。这里的主动关闭实际上是广义的主动关闭，甚至包括用户调用 kill -9 主动“杀掉”Producer 应用。当然最推荐的方式还是调用 producer.close() 方法来关闭。

第二种是 Kafka 帮你关闭，这与 Producer 端参数 connections.max.idle.ms 的值有关。默认情况下该参数值是 9 分钟，即如果在 9 分钟内没有任何请求“流过”某个 TCP 连接，那么 Kafka 会主动帮你把该 TCP 连接关闭。用户可以在 Producer 端设置 connections.max.idle.ms=-1 禁掉这种机制。一旦被设置成 -1，TCP 连接将成为永久长连接。当然这只是软件层面的“长连接”机制，由于 Kafka 创建的这些 Socket 连接都开启了 keepalive，因此 keepalive 探活机制还是会遵守的。

值得注意的是，在第二种方式中，TCP 连接是在 Broker 端被关闭的，但其实这个 TCP 连接的发起方是客户端，因此在 TCP 看来，这属于被动关闭的场景，即 passive close。被动关闭的后果就是会产生大量的 CLOSE_WAIT 连接，因此 Producer 端或 Client 端没有机会显式地观测到此连接已被中断。

04 | 幂等生产者和事务生产者

消息交付可靠性保障有以下三种：

• 最多一次（at most once）：消息可能会丢失，但绝不会被重复发送。
• 至少一次（at least once）：消息不会丢失，但有可能被重复发送。
• 精确一次（exactly once）：消息不会丢失，也不会被重复发送。

目前，Kafka 默认提供的交付可靠性保障是第二种。

什么情况下 Kafka 发送至少一次消息？如果 Broker 的应答没有成功发送回 Producer 端，比如网络出现瞬时抖动，那么 Producer 就无法确定消息是否真的提交成功了。因此，它只能选择重试。

Kafka 也可以提供最多一次交付保障，只需要让 Producer 禁止重试即可。这样一来，消息要么写入成功，要么写入失败，但绝不会重复发送。

Kafka 是怎么做到精确一次的呢？简单来说，是通过两种机制：幂等性（Idempotence）和事务（Transaction）。

幂等性 Producer

幂等指的是某些操作或函数能够被执行多次，但每次得到的结果都是不变的。

在 Kafka 中，Producer 默认不是幂等性的。在 0.11.0.0 版本引入了幂等性功能。使用方法如下：

props.put("enable.idempotence", ture)
或 props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true)
1
2

具体的原理就是经典的用空间去换时间的优化思路，即在 Broker 端多保存一些字段。当 Producer 发送了具有相同字段值的消息后，Broker 能够自动知晓这些消息已经重复了，于是可以在后台默默地把它们“丢弃”掉。

当然，实际的实现原理并没有这么简单，但你大致可以这么理解。

实际上，幂等性 Producer 也是有作用范围的。

首先，它只能保证单分区上的幂等性，即一个幂等性 Producer 能够保证某个主题的一个分区上不出现重复消息，它无法实现多个分区的幂等性。

其次，它只能实现单会话上的幂等性，不能实现跨会话的幂等性。当你重启了 Producer 进程之后，这种幂等性保证就丧失了。

那如果我想实现多分区以及多会话上的消息无重复，应该怎么做呢？答案就是事务（transaction）型 Producer。

Kafka 自 0.11 版本开始也提供了对事务的支持，目前主要是在 read committed 隔离级别上做事情。它能保证多条消息原子性地写入到目标分区，同时也能保证 Consumer 只能看到事务成功提交的消息。

事务型 Producer

事务型 Producer 能够保证将消息原子性地写入到多个分区中。这批消息要么全部写入成功，要么全部失败。另外，事务型 Producer 也不惧进程的重启。Producer 重启回来后，Kafka 依然保证它们发送消息的精确一次处理。

使用事务型 Producer 的配置如下：

• 开启 enable.idempotence = true
• 设置 Producer 端参数 transctional.id。最好为其设置一个有意义的名字

此外，还需要在 Producer 代码中做一些调整，如这段代码所示：

producer.initTransactions();
try {
    producer.beginTransaction();
    producer.send(record1);
    producer.send(record2);
    producer.commitTransaction();
} catch (KafkaException e) {
    producer.abortTransaction();
}
1
2
3
4
5
6
7
8
9

这段代码能够保证 Record1 和 Record2 被当作一个事务统一提交到 Kafka，要么它们全部提交成功，要么全部写入失败。

实际上即使写入失败，Kafka 也会把它们写入到底层的日志中，也就是说 Consumer 还是会看到这些消息。因此在 Consumer 端，读取事务型 Producer 发送的消息也是需要一些变更的。修改起来也很简单，设置 isolation.level 参数的值即可。当前这个参数有两个取值：

• read_uncommitted：这是默认值，表明 Consumer 能够读取到 Kafka 写入的任何消息，不论事务型 Producer 提交事务还是终止事务，其写入的消息都可以读取。很显然，如果你用了事务型 Producer，那么对应的 Consumer 就不要使用这个值。
• read_committed：表明 Consumer 只会读取事务型 Producer 成功提交事务写入的消息。当然了，它也能看到非事务型 Producer 写入的所有消息。