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

以下是Java实现Kafka生产者消费者代码实例的完整攻略，包含两个示例。

简介

Kafka是一种高性能、分布式、可扩展的消息队列系统，可以实现实时数据传输和处理。本攻略将详细讲解Java实现Kafka生产者消费者的代码实例，并提供两个示例。

示例一：Kafka生产者

以下是Java实现Kafka生产者的代码示例：

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;

public class KafkaProducerExample {
    public static void main(String[] args) {
        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");
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        for (int i = 0; i < 10; i++) {
            producer.send(new ProducerRecord<>("test", Integer.toString(i), Integer.toString(i)));
        }
        producer.close();
    }
}

在这个示例中，我们使用KafkaProducer类来创建Kafka生产者，使用Properties类来设置Kafka生产者的配置，使用ProducerRecord类来创建消息记录，使用send方法来发送消息，使用close方法来关闭Kafka生产者。

示例二：Kafka消费者

以下是Java实现Kafka消费者的代码示例：

import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.util.Collections;
import java.util.Properties;

public class KafkaConsumerExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("test"));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            records.forEach(record -> {
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            });
        }
    }
}

在这个示例中，我们使用KafkaConsumer类来创建Kafka消费者，使用Properties类来设置Kafka消费者的配置，使用subscribe方法来订阅主题，使用poll方法来拉取消息，使用forEach方法来遍历消息记录，使用printf方法来输出消息记录的偏移量、键和值。

总结

通过本攻略的介绍，我们了解了Java实现Kafka生产者消费者的代码实例，并提供了两个示例。在实际开发中，我们可以根据具体的业务需求和场景来选择合适的API和配置，以提高系统的性能和可靠性。