kafka消息發送如何進行消息確認

Kafka 消息發送的消息確認機制是通過生產者客戶端來實現的。生產者客戶端在發送消息到 Kafka 集群時，可以配置不同的確認策略來確保消息的可靠傳輸。以下是 Kafka 生產者提供的兩種主要確認方式：

1. 同步確認（Synchronous Acknowledgment）：

   • 在這種模式下，生產者在成功發送消息后，會等待 Kafka 集群的響應，確認消息已經被成功寫入。
   • 生產者會阻塞，直到收到 Kafka 的確認消息。
   • 這種方式可以保證消息的可靠傳輸，但可能會降低生產者的吞吐量。

2. 異步確認（Asynchronous Acknowledgment）：

   • 在這種模式下，生產者發送消息后不會等待 Kafka 的響應，而是立即返回成功。
   • 生產者可以通過輪詢或者回調的方式來檢查消息是否已經被成功寫入。
   • 這種方式可以提高生產者的吞吐量，但可能會增加消息丟失的風險。

配置和使用示例

同步確認

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");
props.put("acks", "all"); // 設置為 all 表示需要所有副本都確認

Producer<String, String> producer = new KafkaProducer<>(props);
producer.send(new ProducerRecord<>("my-topic", "key", "value"), new Callback() {
    public void onCompletion(RecordMetadata metadata, Exception exception) {
        if (exception != null) {
            exception.printStackTrace();
        } else {
            System.out.println("Message sent to topic: " + metadata.topic() + " partition: " + metadata.partition() + " offset: " + metadata.offset());
        }
    }
});
producer.close();

異步確認

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");
props.put("acks", "1"); // 設置為 1 表示只需要 leader 確認

Producer<String, String> producer = new KafkaProducer<>(props);
producer.send(new ProducerRecord<>("my-topic", "key", "value"), new Callback() {
    public void onCompletion(RecordMetadata metadata, Exception exception) {
        if (exception != null) {
            exception.printStackTrace();
        } else {
            System.out.println("Message sent to topic: " + metadata.topic() + " partition: " + metadata.partition() + " offset: " + metadata.offset());
        }
    }
});
producer.close();

總結

• 同步確認：適用于對消息可靠性要求較高的場景，但可能會降低吞吐量。
• 異步確認：適用于對吞吐量要求較高的場景，但可能會增加消息丟失的風險。

根據具體的應用場景和需求，可以選擇合適的確認策略來確保消息的可靠傳輸。