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

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

添加Kafka 依賴

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

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

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

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

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

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

• 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();

幾點(diǎn)需要注意：

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

編寫Kafka 消費(fèi)者（Consumer）

消費(fèi)者負(fù)責(zé)從Kafka 主題中讀取消息。相比生產(chǎn)者，消費(fèi)者的邏輯稍複雜一點(diǎn)，因?yàn)樗枰鲃?dòng)拉取消息、處理偏移量等。

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

• bootstrap.servers ：同樣要指定broker 地址
• group.id ：消費(fèi)者組ID，必須設(shè)置，否則會(huì)報(bào)錯(cuò)
• key.deserializer和value.deserializer ：反序列化器，通常用StringDeserializer

一個(gè)簡單的消費(fèi)者流程如下：

 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());
    }
}

注意點(diǎn)：

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

常見問題與註意事項(xiàng)

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

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

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

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

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