生产者

生产者将消息写入 Kafka，之后可由消费者读取。本节课将详细讲解消息如何写入 Kafka。

写入流程

具体步骤如下：

1. 实例化一个 ProducerRecord 类的对象，该对象必须包含消息的目标主题和消息内容（即值）。消息的键和分区可选。

2. 接下来，将要通过网络发送的键和值进行序列化。如果序列化失败，可能会抛出序列化异常。需要设置两个属性：key.serializer 和 value.serializer，分别指定能够将键和值对象转换为字节的类。

3. 数据随后被发送到分区器。如果在 ProducerRecord 中已指定分区，则分区器不执行任何操作，直接返回已指定的分区。若未指定分区，分区器将根据消息键选择一个分区。

4. 分区确定后，生产者将消息添加到等待发送至同一主题和分区的消息批次中。另一个线程负责将这些批次发送到正确的 Kafka broker。Kafka 集群的位置通过 bootstrap.servers 属性获取，该属性是一个 host:port 地址列表，生产者会尝试连接这些地址。无需提供所有 broker 的完整列表，因为初始连接后生产者会获取更多集群信息。

5. 当 broker 成功接收消息批次并将其写入 Kafka 后，会返回一个 RecordMetadata 对象，其中包含主题、分区以及消息在分区中的偏移量信息。

6. 如果 broker 无法写入消息，会向生产者返回错误。生产者会在放弃前尝试若干次重试。

示例

下方代码框展示了一个生产者示例，它将一条消息写入名为 datajek-topic 的主题。请结合代码注释，将上述流程与代码对应起来。之后将提供一个练习，用于在 Kafka 服务器上运行该生产者。

1    public static void main(String[] args) {
2        // 设置必需的属性
3        Properties kafkaProps = new Properties();
4        kafkaProps.put("bootstrap.servers", "localhost:9092");
5        kafkaProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
6        kafkaProps.put("value.serializer",
7                "org.apache.kafka.common.serialization.StringSerializer");
8
9        // 创建 KafkaProducer 对象，并传入配置属性
10        KafkaProducer producer = new KafkaProducer<String, String>(kafkaProps);
11
12        // 创建要发送到 Kafka 的记录/消息。注意，此示例中我们传入了主题、键和值。回忆一下，消息键是可选的。
13        ProducerRecord<String, String> record = new ProducerRecord<>("datajek-topic", "my-key",
14                "my-value");
15        try {
16            // 将消息发送到 Kafka。send(...) 方法返回一个 RecordMetadata 的 Future 对象。
17            Future<RecordMetadata> future = producer.send(record);
18
19            // 对 Future 对象调用 get()，使发送操作变为同步。
20            RecordMetadata recordMetadata = future.get();
21
22            // RecordMetadata 对象包含消息写入的分区和偏移量。
23            System.out.println(String.format("消息已写入分区 %s，偏移量为 %s", recordMetadata.partition(),
24                    recordMetadata.offset()));
25
26        } catch (Exception e) {
27            System.out.println("发送消息时发生异常：" + e.getMessage());
28        } finally {
29            // 发送完成后，可以调用 flush() 确保所有消息都已写入 Kafka，然后调用 close() 释放生产者资源。
30            producer.flush();
31            producer.close();
32        }
33    }

如果消息成功发送到 Kafka，future.get() 会返回 RecordMetadata 对象；否则，get 方法将抛出异常。该元数据对象包含消息在分区中写入的偏移量。请按照以下说明执行上方代码框中创建的生产者。

1# cd DataJek
2java -cp ".:./*" io.datajek.kafka.producer.ProducerExample

/dev/null 2>&1 & bin/kafka-server-start.sh config/server.properties > /dev/null 2>&1 & bin/kafka-topics.sh --create --topic datajek-topic --bootstrap-server localhost:9092 cd /DataJek " mode="edit">