1.引入依賴
<dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>2.3.1</version></dependency>
2.搭建生產者
package com.wen.kafka;import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;import java.util.Properties;
import java.util.concurrent.ExecutionException;public class MyProducer {public static void main(String[] args) throws ExecutionException, InterruptedException {//配置信息Properties prop = new Properties();prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.117.80:9092");prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());//創建生產者Producer<String,String> producer = new KafkaProducer<String, String>(prop);//創建消息ProducerRecord<String,String> record = new ProducerRecord<>("test", "hello kafka-client");//同步發送消息
// RecordMetadata metadata = producer.send(record).get();
// System.out.println("同步消息——topic:"+metadata.topic()+"partition"+metadata.partition()+"offset"+metadata.offset());//異步發送消息producer.send(record, new Callback() {@Overridepublic void onCompletion(RecordMetadata recordMetadata, Exception e) {if (e != null) {System.out.println(e.getMessage());}if (recordMetadata != null) {System.out.println("異步消息——topic:"+recordMetadata.topic()+"partition"+recordMetadata.partition()+"offset"+recordMetadata.offset());}}});Thread.sleep(1000);}
}
3.搭建消費者
package com.wen.kafka;import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;public class MyConsumer {public static void main(String[] args) {//參數信息Properties prop = new Properties();prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.117.80:9092");prop.put(ConsumerConfig.GROUP_ID_CONFIG,"testGroup");prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());//創建消費者Consumer<String,String> consumer = new KafkaConsumer<String, String>(prop);//訂閱主題consumer.subscribe(Arrays.asList("test"));//拉取消息while (true){ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));for (ConsumerRecord<String, String> record : records) {System.out.println(record.value());}}}
}
)
