Java Kafka 代码示例：生产者和消费者

以下是一个简单的 Java 代码示例，用于生产和消费 Kafka 消息：

生产者：

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class KafkaProducerExample {
   public static void main(String[] args) throws Exception{

      String topicName = 'test-topic';
      String key = 'sample-key';
      String value = 'sample-value';

      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<>(topicName,key,value);
      producer.send(record);
      producer.close();
   }
}

消费者：

import org.apache.kafka.clients.consumer.Consumer;
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.Collections;
import java.util.Properties;

public class KafkaConsumerExample {
   public static void main(String[] args) throws Exception {

      String topicName = 'test-topic';
      String groupName = 'test-group';

      Properties props = new Properties();
      props.put('bootstrap.servers', 'localhost:9092');
      props.put('group.id', groupName);
      props.put('key.deserializer', StringDeserializer.class.getName());
      props.put('value.deserializer', StringDeserializer.class.getName());

      Consumer<String, String> consumer = new KafkaConsumer<>(props);
      consumer.subscribe(Collections.singletonList(topicName));

      while (true) {
         ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
         records.forEach(record -> {
            System.out.printf('offset = %d, key = %s, value = %s%n', record.offset(), record.key(), record.value());
         });
      }
   }
}

这个例子使用 Kafka 的 Java 客户端 API，生产一个包含 'sample-key' 和 'sample-value' 的消息，并将其发送到名为 'test-topic' 的主题中。然后，消费者订阅此主题并从中接收消息。在消费者中，我们使用了一个无限循环，不断从 Kafka 主题中拉取消息。当收到消息时，我们只是简单地打印它们的偏移量，键和值。