使用Java编写Kafka生产者和消费者示例

标题：使用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();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

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());
            }
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

4. 运行示例

将以上两个Java文件编译并运行，即可在Kafka中发送和接收消息。确保Kafka服务已经启动，并且Topic已经创建。

5. 总结

本文介绍了如何使用Java编写Kafka的生产者和消费者示例。通过这些示例代码，我们可以轻松地在Java应用中集成Kafka，实现消息的生产和消费。Kafka具有高性能、高可靠性的特点，适用于各种场景的消息处理需求。