java - 如何识别哪些消息成功发布到kafka主题以及哪些消息失败

Question

将消息列表发布到 apache kafka。任何人都可以提供使用 kafka api 的示例代码，显示如何识别哪些消息已成功发布到主题以及哪些消息在响应中失败？ (请注意，我在一个请求中批量发送消息列表。)

Answer 1

KafkaProducer.send() 方法采用单个 ProducerRecord(消息)。

有两种方法可以检查集群是否成功接收到该消息:

• 使用回调:send() 可以将回调作为第二个参数

  ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);
  producer.send(record, new Callback() {
      @Override
      public void onCompletion(RecordMetadata metadata, Exception exception) {
          // If Exception is null, the record was sent successfully
      }
  });
  

• 使用 Future:send() 返回一个 Future

   ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);
   Future<RecordMetadata> future = producer.send(record);
   try {
       RecordMetadata rm = future.get();
       // The record was sent successfully
   } catch (ExecutionException e) {
       // The record failed
   }