全解析Kafka Java生产者和消费者

2024-01-02 04:16:25

Kafka Java：轻松掌握生产者和消费者编程

前言

Apache Kafka 是一种革命性的分布式流处理平台，以其惊人的高吞吐量、超低延迟和强大的容错性而闻名。对于任何希望构建高性能数据管道或流媒体应用程序的开发者来说，Kafka 都是必不可少的工具。在这篇全面指南中，我们将深入探究 Kafka Java 生产者和消费者，带你了解它们的配置、使用方法和实际用例。

Kafka 生产者

1. 创建生产者

创建 Kafka 生产者非常简单。只需创建一个 Properties 对象，设置引导服务器（即 Kafka 集群中代理的地址）、键序列化程序（用于将键转换为字节数组）和值序列化程序（用于将值转换为字节数组）。然后使用这些属性创建 KafkaProducer 实例。

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

import java.util.Properties;

public class KafkaProducerExample {

    public static void main(String[] args) {
        // 设置生产者配置
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        // 创建生产者
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        // 发送消息
        producer.send(new ProducerRecord<String, String>("my-topic", "hello, Kafka!"));

        // 关闭生产者
        producer.close();
    }
}

2. 发送消息

要发送消息，只需创建 ProducerRecord 实例，指定主题（消息发送到的目标主题）和消息（键值对）。然后使用 producer.send() 方法发送消息。

3. 关闭生产者

发送完消息后，请务必调用 producer.close() 方法关闭生产者，释放所有资源。

Kafka 消费者

1. 创建消费者

与生产者类似，创建 Kafka 消费者也很容易。同样，创建一个 Properties 对象并设置引导服务器、组 ID（识别属于同一消费组的消费者）、键反序列化程序和值反序列化程序。然后使用这些属性创建 KafkaConsumer 实例。

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
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 properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        // 创建消费者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);

        // 订阅主题
        consumer.subscribe(Collections.singletonList("my-topic"));

        // 拉取消息
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                System.out.println(record.key() + ": " + record.value());
            }
        }

        // 关闭消费者
        consumer.close();
    }
}