【Kafka】Java整合Kafka

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