pom.xml
? ? ? ? <dependency>
? ? ? ? ? ? <groupId>org.apache.kafka</groupId>
? ? ? ? ? ? <artifactId>kafka-clients</artifactId>
? ? ? ? ? ? <version>2.3.1</version>
? ? ? ? </dependency>

注意調試通過后要添加log-back.xml并設置日志級別到info及以上，否則debug日志實在是多啊……

///ProducerTest.java

package com.whq.demo; import java.util.Properties; import java.util.Random; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.serialization.StringSerializer; public class ProducerTest {public static String topic = "whq_test";//定義主題public static void main(String[] args) throws InterruptedException {Properties p = new Properties();p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.81.145:9092,192.168.81.145:9093,192.168.81.145:9094");//kafka地址，多個地址用逗號分割p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(p);try {while (true) {String msg = "Hello,你好," + new Random().nextInt(100);ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic, msg);kafkaProducer.send(record);System.out.println("消息發送成功:" + msg);Thread.sleep(500);}} finally {kafkaProducer.close();}} }

///ConsumerTest.java

package com.whq.demo; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.common.serialization.StringDeserializer;import java.util.Collections; import java.util.Properties;public class ConsumerTest {public static void main(String[] args) {Properties p = new Properties();p.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.81.145:9092,192.168.81.145:9093,192.168.81.145:9094");p.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);p.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);p.put(ConsumerConfig.GROUP_ID_CONFIG, "3");//組編號，不同的組提交偏移不同p.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,true);//自動提交，提交后數據在同組無法再次獲取,否則同組另外消費者接入還會收到同樣的消息p.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");/*** AUTO_OFFSET_RESET_CONFIG 重置偏移設置* earliest* 當各分區下有已提交的offset時，從提交的offset開始消費；無提交的offset時，從頭開始消費* latest* 當各分區下有已提交的offset時，從提交的offset開始消費；無提交的offset時，消費新產生的該分區下的數據* none* topic各分區都存在已提交的offset時，從offset后開始消費；只要有一個分區不存在已提交的offset，則拋出異常*/KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(p);kafkaConsumer.subscribe(Collections.singletonList(ProducerTest.topic));// 訂閱消息while (true) {ConsumerRecords<String, String> records = kafkaConsumer.poll(100);for (ConsumerRecord<String, String> record : records) {System.out.println(String.format("topic:%s,offset:%d,消息:%s", //record.topic(), record.offset(), record.value()));}}} }

//出現問題：生產者第二次發消息時連接到了127.0.0.1:9093，后來發現是kafka服務器配置問題：
vi config/server.properties
修改

listeners=PLAINTEXT://192.168.81.145:9092 ? ?#注意，這里一定要有客戶端可訪問的ip，否則非本機連接調用第二次會異常

總結

以上是生活随笔為你收集整理的java调用kafka的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。