Java实现Kafka生产者消费者代码实例

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

package com.example.kafkatest;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class Producer {
  public static void main(String[] args) {
    //配置信息
    Properties props = new Properties();
    //kafka服务器地址
    props.put("bootstrap.servers", "localhost:9092");
    //设置数据key和value的序列化处理类
    props.put("key.serializer", StringSerializer.class);
    props.put("value.serializer", StringSerializer.class);
    //创建生产者实例
    KafkaProducer<String,String> producer = new KafkaProducer<>(props);
    ProducerRecord record = new ProducerRecord<String, String>("topic1", "userName", "lc");
    //发送记录
    producer.send(record);
    producer.close();
  }
}

package com.example.rabbittest;

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.time.Duration;
import java.util.Arrays;
import java.util.Properties;

public class Consumer {
  public static void main(String[] args) {
    //配置信息
    Properties props = new Properties();
    //kafka服务器地址
    props.put("bootstrap.servers", "localhost:9092");
    //必须指定消费者组
    props.put("group.id", "test");
    //设置数据key和value的序列化处理类
    props.put("key.deserializer", StringDeserializer.class);
    props.put("value.deserializer", StringDeserializer.class);
    //创建消息者实例
    KafkaConsumer<String,String> consumer = new KafkaConsumer<>(props);
    //订阅topic1的消息
    consumer.subscribe(Arrays.asList("topic1"));
    //到服务器中读取记录
    while (true){
      ConsumerRecords<String,String> records = consumer.poll(Duration.ofMillis(100));
      for(ConsumerRecord<String,String> record : records){
        System.out.println("key:" + record.key() + "" + ",value:" + record.value());
      }
    }
  }
}