Kafka 的生产者如何发送消息？

推荐答案

Kafka 生产者发送消息的步骤如下：

1. 创建生产者实例：首先需要创建一个 KafkaProducer 实例，配置必要的参数，如 bootstrap.servers、key.serializer、value.serializer 等。

2. 构建消息：使用 ProducerRecord 类创建一个消息对象，指定目标主题（Topic）、分区（Partition，可选）、键（Key，可选）和值（Value）。

3. 发送消息：通过 KafkaProducer 的 send() 方法发送消息。send() 方法支持同步和异步两种方式：

   • 同步发送：调用 send() 方法后，使用 get() 方法等待消息发送完成并获取结果。
   • 异步发送：调用 send() 方法时传入一个回调函数（Callback），Kafka 会在消息发送完成后调用该回调函数。

4. 关闭生产者：发送完消息后，调用 KafkaProducer 的 close() 方法关闭生产者，释放资源。

本题详细解读

1. 创建生产者实例

在 Kafka 中，生产者是通过 KafkaProducer 类来实现的。创建生产者实例时，需要配置一些必要的参数：

• bootstrap.servers：指定 Kafka 集群的地址列表。
• key.serializer：指定键的序列化器，通常使用 StringSerializer 或 ByteArraySerializer。
• value.serializer：指定值的序列化器，通常使用 StringSerializer 或 ByteArraySerializer。

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

2. 构建消息

消息是通过 ProducerRecord 类来表示的。ProducerRecord 的构造函数可以指定目标主题、分区、键和值。

ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "key", "value");

3. 发送消息

发送消息时，可以选择同步或异步方式。

同步发送

同步发送会阻塞当前线程，直到消息发送完成并返回结果。

Future<RecordMetadata> future = producer.send(record);
RecordMetadata metadata = future.get();  // 阻塞等待结果
System.out.println("Message sent to partition " + metadata.partition() + " with offset " + metadata.offset());

异步发送

异步发送不会阻塞当前线程，Kafka 会在消息发送完成后调用回调函数。

// www.javascriptcn.com code example
producer.send(record, new Callback() {
    @Override
    public void onCompletion(RecordMetadata metadata, Exception exception) {
        if (exception == null) {
            System.out.println("Message sent to partition " + metadata.partition() + " with offset " + metadata.offset());
        } else {
            exception.printStackTrace();
        }
    }
});

4. 关闭生产者

发送完消息后，应该关闭生产者以释放资源。

producer.close();

通过以上步骤，Kafka 生产者可以成功发送消息到指定的主题。