使用Java 生產(chǎn)和消費Apache Kafka 消息的關鍵在於正確配置Producer 和Consumer API 並理解其基本流程。 1. 首先添加Kafka 客戶端依賴，確保版本與集群兼容；2. 編寫生產(chǎn)者時配置bootstrap.servers、key.serializer 和value.serializer，並創(chuàng)建KafkaProducer 實例發(fā)送消息，注意關閉資源和可選回調處理；3. 編寫消費者時配置group.id、反序列化器等，使用KafkaConsumer 訂閱主題並循環(huán)拉取消息，注意提交offset 的策略；4. 常見問題包括group.id 衝突、offset 提交不當、資源未正確關閉等，建議先在本地單機環(huán)境測試驗證流程。

要用Java 生產(chǎn)和消費Apache Kafka 消息，其實並不難。核心在於理解Kafka 的基本概念，並掌握Producer 和Consumer API 的使用方法。只要配置得當，代碼結構清晰，就能輕鬆實現(xiàn)消息的發(fā)送與接收。

添加Kafka 依賴

在開始寫代碼之前，首先需要把Kafka 的客戶端庫引入項目中。如果你用的是Maven，可以在pom.xml中添加如下依賴：

 <dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.6.0</version>
</dependency>

這個版本比較新，也相對穩(wěn)定。當然你也可以根據(jù)你的Kafka 集群版本選擇對應的客戶端版本。

編寫Kafka 生產(chǎn)者（Producer）

生產(chǎn)者的任務是往Kafka 主題裡發(fā)消息。主要步驟包括配置屬性、創(chuàng)建Producer 實例、構建消息記錄並發(fā)送。

關鍵配置項有：

• bootstrap.servers ：Kafka broker 地址
• key.serializer和value.serializer ：指定key 和value 的序列化方式，一般用StringSerializer

示例代碼如下：

 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");

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

producer.send(record);
producer.close();

幾點需要注意：

• 發(fā)送後最好調用close() ，避免資源洩漏
• 如果想確認是否發(fā)送成功，可以加上回調函數(shù).send(record, callback)
• 如果只是測試，可以省略key

編寫Kafka 消費者（Consumer）

消費者負責從Kafka 主題中讀取消息。相比生產(chǎn)者，消費者的邏輯稍複雜一點，因為它需要主動拉取消息、處理偏移量等。

關鍵配置包括：

• bootstrap.servers ：同樣要指定broker 地址
• group.id ：消費者組ID，必須設置，否則會報錯
• key.deserializer和value.deserializer ：反序列化器，通常用StringDeserializer

一個簡單的消費者流程如下：

 Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("my-topic"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        System.out.println("Received: " record.value());
    }
}

注意點：

• poll()方法會拉取一批數(shù)據(jù)，循環(huán)處理即可
• 消費完不要忘了提交offset，可以選擇自動或手動提交
• 如果你想只消費一次然後退出，可以在處理完後break 循環(huán)

常見問題與註意事項

有時候你會發(fā)現(xiàn)消息沒收到或者重複消費了，這通常是下面幾個原因造成的：

• 消費者group.id 設置錯誤或衝突
• 自動提交offset 頻率太高或太低
• 沒有正確關閉producer 或consumer 導致狀態(tài)??不一致
• Kafka broker 沒啟動或者網(wǎng)絡不通

建議開發(fā)時先跑通本地單機環(huán)境，再上集群。

基本上就這些。 Java 接入Kafka 的過程不算複雜，但有些細節(jié)容易忽略，比如序列化、group.id、提交策略這些，只要注意配置和流程，很快就能跑起來。

以上是如何使用Java從Apache Kafka中產(chǎn)生和消費消息？的詳細內容。更多資訊請關注PHP中文網(wǎng)其他相關文章！