标题：使用Java编写Kafka生产者和消费者示例
Kafka是一个分布式流处理平台，具有高性能、高吞吐量的特点，被广泛应用于消息队列、日志收集、事件处理等场景。在本文中，我们将介绍如何使用Java编写Kafka的生产者和消费者，以实现消息的生产和消费。
1. 准备工作
在开始之前，确保已经安装并配置了Kafka，并启动了Zookeeper和Kafka服务。可以从官方网站https://kafka.apache.org/downloads下载Kafka。
2. 编写Kafka生产者
首先，我们来编写一个简单的Kafka生产者，用于向指定的Topic发送消息。
import org.apache.kafka.clients.producer.*; import java.util.Properties; public class KafkaProducerExample { public static void main(String[] args) { String bootstrapServers = "localhost:9092"; String topic = "test-topic"; Properties props = new Properties(); props.put("bootstrap.servers", bootstrapServers); 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); for (int i = 0; i < 10; i++) { String key = "key-" + i; String value = "message-" + i; ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value); producer.send(record, new Callback() { @Override public void onCompletion(RecordMetadata metadata, Exception exception) { if (exception == null) { System.out.println("Message sent successfully! Topic: " + metadata.topic() + ", Partition: " + metadata.partition() + ", Offset: " + metadata.offset()); } else { System.err.println("Error sending message: " + exception.getMessage()); } } }); } producer.close(); } }
3. 编写Kafka消费者
接下来，我们编写一个简单的Kafka消费者，用于从指定的Topic消费消息。
import org.apache.kafka.clients.consumer.*; import org.apache.kafka.common.TopicPartition; import java.time.Duration; import java.util.Collections; import java.util.Properties; public class KafkaConsumerExample { public static void main(String[] args) { String bootstrapServers = "localhost:9092"; String topic = "test-topic"; String groupId = "test-group"; Properties props = new Properties(); props.put("bootstrap.servers", bootstrapServers); props.put("group.id", groupId); props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); Consumer<String, String> consumer = new KafkaConsumer<>(props); consumer.subscribe(Collections.singletonList(topic)); while (true) { ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100)); for (ConsumerRecord<String, String> record : records) { System.out.println("Received message: " + "Key = " + record.key() + ", Value = " + record.value() + ", Topic = " + record.topic() + ", Partition = " + record.partition() + ", Offset = " + record.offset()); } } } }
4. 运行示例
将以上两个Java文件编译并运行，即可在Kafka中发送和接收消息。确保Kafka服务已经启动，并且Topic已经创建。
5. 总结
本文介绍了如何使用Java编写Kafka的生产者和消费者示例。通过这些示例代码，我们可以轻松地在Java应用中集成Kafka，实现消息的生产和消费。Kafka具有高性能、高可靠性的特点，适用于各种场景的消息处理需求。
