Kafka的发布-订阅功能: Java实现与应用场景解析

GeekLiHua

发布于 2025-01-21 16:30:57

12500

代码可运行

文章被收录于专栏：JavaJava

运行总次数：0

代码可运行

Kafka的发布-订阅功能: Java实现与应用场景解析

Apache Kafka 是一个流行的分布式消息传递系统，广泛用于构建实时数据管道和流应用程序。它以高吞吐量、内置的分区、副本和容错性著称。Kafka 主要提供发布-订阅模式，这里我们将深入探讨 Kafka 的发布-订阅功能，通过一个 Java 代码示例和一个具体的应用场景。

Kafka的核心概念

Producer（生产者）：向 Kafka 主题发送消息的实体。
Consumer（消费者）：从 Kafka 主题读取消息的实体。
Topic（主题）：消息的分类，生产者向主题发送消息，消费者从主题读取消息。
Broker（代理）：Kafka 集群中的服务器节点。

Java 代码示例

环境配置

添加依赖：在 Maven 的 pom.xml 文件中添加 Kafka 客户端依赖。

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.8.0</version>
</dependency>

Kafka Producer 示例

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class SimpleProducer {
    public static void main(String[] args) {
        String bootstrapServers = "127.0.0.1:9092";
        String topicName = "testTopic";

        // 设置生产者属性
        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

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

        // 创建消息
        ProducerRecord<String, String> record = new ProducerRecord<>(topicName, "Hello, Kafka!");

        // 发送消息
        producer.send(record);

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

Kafka Consumer 示例

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Collections;
import java.util.Properties;

public class SimpleConsumer {
    public static void main(String[] args) {
        String bootstrapServers = "127.0.0.1:9092";
        String groupId = "test-group";
        String topic = "testTopic";

        // 设置消费者属性
        Properties properties = new Properties();
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

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

        // 订阅主题
        consumer.subscribe(Collections.singleton(topic));

        // 拉取并处理消息
        while(true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("Received message: key = %s, value = %s, offset = %d%n", 
                                  record.key(), record.value(), record.offset());
            }
        }

        // 注意：实际应用中需要合适的退出机制
    }
}