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

要用 Java 生產(chǎn)和消費 Apache Kafka 消息，其實并不難。核心在于理解 Kafka 的基本概念，并掌握 Producer 和 Consumer API 的使用方法。只要配置得當(dāng)，代碼結(jié)構(gòu)清晰，就能輕松實現(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)定。當(dāng)然你也可以根據(jù)你的 Kafka 集群版本選擇對應(yīng)的客戶端版本。

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

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

關(guān)鍵配置項有：

• 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ā)送后最好調(diào)用 close()，避免資源泄漏
• 如果想確認(rèn)是否發(fā)送成功，可以加上回調(diào)函數(shù) .send(record, callback)
• 如果只是測試，可以省略 key

編寫 Kafka 消費者（Consumer）

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

關(guān)鍵配置包括：

• bootstrap.servers：同樣要指定 broker 地址
• group.id：消費者組 ID，必須設(shè)置，否則會報錯
• 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)消息沒收到或者重復(fù)消費了，這通常是下面幾個原因造成的：

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

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

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

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