Kafka 生产者客户端开发
-
- 1、Kafka 客户端依赖
- 2、初始化 Producer
-
- 2.1、参数配置
- 3、使用同步的方式发送消息
- 4、异步方式发送一条消息
- 5、小结
上一小节中，主要介绍了如何安装和运行 Kafka Broker，也简单介绍了如何使用 Kafka 自带的命令行客户端进行消息的生产发送和消费。
这一小节主要内容是如何编写 Java 版的 “Hello World” 生产者客户端，实践如何通过 Java 客户端向 Kafka 发送消息。
前面提到过现在的 Kafka 源码是由 Scala 编写的 Broker 端和由 Java 编写的客户端两大部分组成。但在历史上，Kafka 客户端一开始也是由 Scala 写的，从 Kafka 0.8.x 之后版本开始改为 Java，并改进 Scala 版本客户端的许多不足，因此这里不推荐使用老版的客户端，也不会介绍老版的客户端如何进行开发。
客户端支持的客户端语言列表，链接如下：但这些客户端不是 Kafka 官方维护的，有些可能已经不再支持和维护：
https://cwiki.apache.org/confluence/display/KAFKA/Clients
Kafka 笔记系列的源码会同步到 GitHub 和 Gitee：
https://github.com/zhshuixian/kafka-action-notes
https://gitee.com/ylooq/kafka-action-notes
在开始之前，需要先使用命令行创建一个具有三个分区的 topic
# 命令的含义跟上一节的一样
kafka-topics.sh --create --bootstrap-server 192.168.128.11:9092 \
--replication-factor 1 --partitions 3 \
-topic hello-kafka
1、Kafka 客户端依赖
Kafka 生产者和消费者依赖于 **“kafka-clients”**的 jar 包，通过访问下面的连接可以快速找到对应版本的 Maven、Gradle 等构建工具的依赖配置的信息：
https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients
新建一个 Gradle 或者 Maven 的项目，直接将上面链接的依赖配置拷贝到项目配置中即可，示例源代码中使用了 Gradle 作为构建工具，如果你使用 Maven 可以这么配置，依赖配置好之后，刷新一下项目依赖。
<!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients -->
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.7.0</version>
</dependency>
2、初始化 Producer
Kafka 的 Producer 客户端的只需要指定一些必须配置后，直接初始化即可，代码如下：
Properties props = new Properties();
// Kafka 集群的地址，多个用 英文逗号 隔开 192.168.128.11:9092,192.168.128.12:9092
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.128.11:9092");
// Key 和 Value 的系列化方式
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
// 客户端 id
props.put(ProducerConfig.CLIENT_ID_CONFIG, "HelloKafka");
// Producer 的 acks 机制配置
props.put(ProducerConfig.ACKS_CONFIG, "-1");
// 初始化 Kafka Producer 的客户端， key 是 Integer， value 是 String 字符串
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
KafkaProducer<K, V> 需要传入两个泛型，分别表示消息的 Key、Value 的类型。
如下两种写法的等价的，不过在实际开发中，推荐使用方法 1，因为很可能一不小心就拼写错误了。配置的实际名称在 IDEA 点进去可以看到。这也是值得学习的一个点，使用 Config 类的静态常量或者 properties 文件进行一些配置，避免在编写代码的时候因为编写错误而发生错误，而且改动起来也比较方便。
// 写法 1
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.128.11:9092");
// 写法 2
props.put("bootstrap.servers", "192.168.128.11:9092");
2.1、参数配置
Kafka Producer 客户端的参数配置中，有三个是必须的：
bootstrap.servers ：用来指定 Kafka 集群的 Broker 的 IP 和端口号。可以指定多个，只要客户端成功连接到其中一个即可更新整个 Broker 集群的元数据信息。
key.serializer 和 value.serializer ： 指定 Key 和 Value 的序列化方式，在消费者端要指定他们相对应的反系列化方式。Kafka 自带实现了像 String、Integer、Long、Double、Float、Byte、ByteBuffer 等的系列化和反系列化的类，具体在如下的包下可以找到。
org.apache.kafka.common.serialization
// Kafka 的系列化接口
Serializer
// Kafka 的反系列化接口
Deserializer
以上三个配置项是必须的，下面的是一些比较常用也比较实用的配置项：
client.id ：客户端 ID，如果没有设置则是 producer-1。
retries ： 重试次数，默认 0 次，意思是消息发送失败后的重试次数。
retry.backoff.ms ：重试时间间隔，消息失败后等待多少毫秒以后才进行重试。
acks : ACKS 是生产者客户端一个非常重要的配置项，其有以下三种选项：
- acks = 0，客户端发送出去以后就认为发送成功了，retries 配置了也不会重试。因此存在消息丢失的可能，但不会重复发送，也就是“At most once”。
- acks = 1，生产者客户端是先发到 Leader 分区，再由 Flower 分区去 Leader 分区拉取同步的，当 acks = 1 的时候，客户端只要收到 Leader 分区返回的成功写入的响应就认为是发送成功了。同样也是存在消息丢失的可能，是因为 Broker 默认是将消息写入内存，刷写进磁盘是通过操作系统机制来实现的。因此还在内存中但没有刷写进磁盘、还没有同步到 Flower 分区那部分的消息会在 Broker 宕机后丢失。
- acks = -1，也就是 acks = all， 跟 acks = 1 的区别是，Leader 需要收到所有 ISR 的 FLower 分区也同步拉取了这条消息以后，才会给客户端返回写入成功的响应。
3、使用同步的方式发送消息
Kafka 生产者客户端原本就是异步方式发送消息的，但可以通过 producer.send() 方法返回的 Future<RecordMetadata> 实现一种“同步”发送消息的效果。代码如下：
// 同步的方式发生 100 条 Hello Kafka 的消息
for (int i = 0; i <= NUM; i++) {
try {
String messageKey = "sync-key-" + i;
String messageValue = "sync-value-hello-kafka-" + i;
long startTime = System.currentTimeMillis();
// send() 其实也是异步的方式发送的，返回 Future，使用 get 会阻塞到结果返回
// metadata 是一条消息发送成功后返回的元数据信息
RecordMetadata metadata = producer.send(new ProducerRecord<>(TOPIC, messageKey, messageValue)).get();
log.info("同步方式发送一条消息耗时{}ms, 第{}条消息, key={}, message={}，partition={}, offset={}",
System.currentTimeMillis() - startTime, i, messageKey, messageValue, metadata.partition(), metadata.offset());
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
}
上述代码比较简单，只需要注意
new ProducerRecord<>(TOPIC, messageKey, messageValue)
即可，生产者发送一条消息，只需要指定 TOPIC 和 消息 Value 的值即可，消息 Key 都是可选的。
4、异步方式发送一条消息
上面代码，不使用 Future.get() 方法阻塞就变成了生产者异步发送消息了，但如果我们要在消息发送后进行一些其他操作，除了使用 Future.get() 之外， Kafka 还提供了“回调方法”来实现消息的发送后的需要进行的操作。
下面演示了一个 很简单的“回调方法”，统计消息发送成功和失败的条数，全部代码在 SimpleCallback 中：
/**
* 异步发送的方式:实现一个 Callback,这个方法是消息发送成功或者失败后的回调
*/
@Override
public void onCompletion(RecordMetadata metadata, Exception exception) {
long elapsedTime = System.currentTimeMillis() - startTime;
if(metadata != null) {
HelloProducer.asyncSendSuccess.incrementAndGet();
log.info("异步发送一条消息耗时{}ms ,key={}, message={}，partition={}, offset={}"
, elapsedTime, key, message, metadata.partition(), metadata.offset());
} else {
HelloProducer.asyncSendFailed.incrementAndGet();
exception.printStackTrace();
}
HelloProducer.LATCH.countDown();
}
使用“回调方法”的异步发送方式，发生成功或者失败后客户端会自动调用 onCompletion()：
// 异步方式发送
for (int i = 0; i < NUM; i++) {
String messageKey = "async-key-" + i;
String messageValue = "async-value-hello-kafka-" + i;
producer.send(new ProducerRecord<>(TOPIC, messageKey, messageValue),
new SimpleCallback(messageKey, messageValue, System.currentTimeMillis()));
}
5、小结
为什么第一条数据发送的耗时远远大于后面发送的耗时？
因为 Kafka Producer 初始化后不会主动去拉取集群的元数据信息，而是等到需要发送消息的时候，才会主动去拉取集群的 Topic、 Partition 以及这些Topic、Partition 的所在的 Broker 等信息，然后跟对应的 Broker 建立连接后才能发送消息。
为什么同步发送一条消息的耗时看起来比异步发送一条消息的时间要少呢？
通过观察刚才的 HelloProducer 的运行时候的输出，可以发现同步方式发送一条消息的时间好像要比异步要少。这是因为 Kafka 的 Producer 端并不是来一条消息就发一条消息的，而是先将消息暂存一下然后一批一批的发送。这些以后会在 Kafka 笔记后面的 Kafka 源码阅读部分继续深入。
这一节主要是介绍了如何在使用 Kafka 的客户端 jar 包开发一个 Java 版的“Hello World”生产者。在下一小节中，会继续介绍如何编写消费者客户端，用实践的方式看看什么是消费者组。Kafka 笔记系列前面几章的部分都以实践为主，介绍完如何编写消费者客户端后，会进入源码阅读的部分。
![Java String.format方法的简单使用[图]](data:image/gif;base64,R0lGODlhAQABAIAAAP///wAAACwAAAAAAQABAAACAkQBADs=)