Kafka 生产者消费者架构

2025年5月14日 | 阅读18分钟

Kafka 生产者-消费者架构简介

Kafka 生产者-消费者架构是 Apache Kafka 的核心组成部分，Kafka 是一个分布式流处理平台，用于构建实时数据管道和流处理应用程序。Kafka 允许生产者将记录（消息）发送到主题（topic），然后由消费者进行消费。Kafka 的架构设计用于以可扩展、分布式的方式处理大量数据，确保容错性和持久性。

Kafka 作为分布式提交日志运行，其中记录按顺序存储，并且可以多次消费，为数据流和实时分析提供了可靠的机制。Kafka 的架构将生产、存储和消费数据的职责分开，使其能够高效地管理大规模分布式系统中的数据流。

关键概念

生产者：将数据发布到 Kafka 主题的实体。
消费者：从 Kafka 主题读取数据的实体。
主题：记录被发布的类别或馈送名称。
分区：主题的子划分，允许并行处理。
代理（Broker）：管理记录存储和检索的 Kafka 服务器。

Kafka 生产者-消费者架构的关键组件

Kafka 的生产者-消费者架构是其处理分布式、可扩展且容错的实时数据流能力的核心。理解该架构的关键组件对于设计、部署和管理基于 Kafka 的系统至关重要。主要组件包括生产者、消费者、主题、分区、代理和 Zookeeper。

1. 生产者

生产者负责将数据（记录）发送到 Kafka 主题。它们在生成数据流方面起着至关重要的作用，无论是日志、指标、事件还是其他形式的数据。

特性

异步通信：生产者可以异步地将记录发送到 Kafka，从而允许它们在不等待 Kafka 的确认的情况下继续生产数据。这可以提高吞吐量并降低延迟。
分区：生产者可以通过自定义逻辑或依赖 Kafka 的默认分区器来决定将记录发送到主题中的哪个分区。分区对于负载平衡和并行处理至关重要。
确认和持久性：在将记录视为成功生产之前，生产者可以配置所需的 Kafka 确认级别。选项包括
- acks=0：无需确认。生产者不等待任何确认，提供最低的延迟，但可能面临数据丢失的风险。
- acks=1：生产者等待分区领导者的确认。这在性能和持久性之间提供了平衡。
- acks=all：生产者等待所有同步副本（ISR）的确认，确保最高级别的持久性。

示例配置

 
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("acks", "all");
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);   

输出

2. 消费者

消费者从 Kafka 主题读取和处理数据。它们被分组到消费者组中，每个组订阅一个或多个主题。Kafka 在组内的消费者之间分配分区，确保每个分区仅由组内的单个消费者消费，从而实现并行处理。

特性

消费者组：消费者在组内工作，这使得可扩展。如果某个消费者失败，Kafka 会自动将其分区重新分配给组内的其他消费者。
偏移量管理：Kafka 会跟踪每个消费者组的偏移量（即最后一个已消费记录的位置）。这确保了在重新启动或故障的情况下，每个消费者都能从上次中断的地方继续。
负载平衡：Kafka 会在组内的消费者之间自动平衡负载，确保每个分区都能得到有效处理。
重新平衡：当消费者加入或离开组时，Kafka 会触发重新平衡，在剩余的消费者之间重新分配分区。虽然重新平衡是必要的，但它可能会引入延迟，因此需要仔细管理。

示例程序

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-group");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("my-topic"));   

输出

3. 主题和分区

主题是生产者发送记录和消费者读取的主题类别或馈送名称。Kafka 中的每个主题都分为多个分区。

特性

分区：一个主题被划分为多个分区，这些分区分布在 Kafka 代理之间。每个分区都是一条有序的、不可变的记录序列，是 Kafka 中并行处理的基本单元。
复制：Kafka 将分区复制到多个代理以确保容错。副本数量是可配置的，Kafka 确保数据在副本之间保持一致。
顺序保证：Kafka 保证分区内的记录是严格有序的，但不保证跨分区的顺序。这在设计依赖数据顺序的系统时非常重要。

主题创建示例

kafka-topics.sh --create --topic my-topic --bootstrap-server localhost:9092 --partitions 3 --replication-factor 2   

此命令创建了一个名为“my-topic”的主题，包含 3 个分区和 2 个复制因子。

4. 代理

代理是管理和存储数据的 Kafka 服务器。每个 Kafka 集群由多个代理组成，每个代理负责一部分分区。

特性

领导者和跟随者：对于每个分区，一个代理充当领导者，处理所有读写请求。其他代理将分区的副本存储为跟随者，并从领导者复制数据。
可扩展性：Kafka 的架构允许您向集群添加更多代理以处理增加的负载，从而使 Kafka 具有高度可扩展性。
容错性：如果某个代理发生故障，Kafka 会自动从同步副本中选举新的领导者，从而确保系统继续正常运行。

配置示例 (server.properties)

broker.id=0
log.dirs=/tmp/kafka-logs
zookeeper.connect=localhost:2181   

5. Zookeeper

Zookeeper 由 Kafka 用于管理和协调集群。它处理诸如领导者选举、配置管理以及代理之间的状态同步等任务。

特性

集群协调：Zookeeper 跟踪 Kafka 代理的状态，并协调它们之间的分区分发。
领导者选举：Zookeeper 为分区选举领导者，确保数据可用且一致。
配置管理：Kafka 将配置数据存储在 Zookeeper 中，代理使用这些数据来维护集群之间的一致性。

示例配置

这些关键组件构成了 Kafka 生产者-消费者架构的支柱。它们共同使 Kafka 能够提供一个可扩展、容错且高吞吐量的数据流处理平台，该平台在各个行业中被广泛采用，用于实时数据处理和分析。理解每个组件及其交互方式对于有效设计和管理基于 Kafka 的系统至关重要。

Kafka 生产者-消费者架构的类型

Kafka 的生产者-消费者架构非常灵活，并支持各种配置，具体取决于系统的要求。这些配置或类型决定了生产者和消费者如何与 Kafka 主题和分区进行交互，影响诸如可扩展性、吞吐量和容错能力等因素。主要的 Kafka 生产者-消费者架构类型是

单生产者、单消费者 (SPSC)
多生产者、单消费者 (MPSC)
单生产者、多消费者 (SPMC)
多生产者、多消费者 (MPMC)

这些架构中的每一种都有独特的特征、优点和挑战，将在下面详细探讨。

1. 单生产者、单消费者 (SPSC)

在单生产者、单消费者 (SPSC) 架构中，一个生产者将数据发送到 Kafka 主题，一个消费者从此主题读取数据。这是最简单的 Kafka 架构，通常用于数据流从一个源到另一个目的地的简单场景。

特性

简洁性：SPSC 模型易于实现和管理。不存在涉及多个生产者或消费者的复杂性。
顺序保证：由于只有一个生产者和一个消费者，因此消息的顺序自然得到保留。
低开销：只有一个生产者和一个消费者，在协调和资源管理方面开销最小。

用例

点对点通信：适用于单个数据生产者（如传感器或日志生成器）需要将数据发送到单个消费者（如监控应用程序）的场景。
测试和原型设计：SPSC 常用于开发环境，其中简单的设置足以测试 Kafka 的功能。

示例实现

// Producer Configuration
Properties producerProps = new Properties();
producerProps.put("bootstrap.servers", "localhost:9092");
producerProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
producerProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps);
producer.send(new ProducerRecord<>("my-topic", "key", "value"));
Properties consumerProps = new Properties();
consumerProps.put("bootstrap.servers", "localhost:9092");
consumerProps.put("group.id", "my-group");
consumerProps.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
consumerProps.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
consumer.subscribe(Arrays.asList("my-topic"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("Consumed record with key %s and value %s%n", record.key(), record.value());
    }
}   

输出

挑战

单点故障：如果生产者或消费者发生故障，整个数据流将中断。
可扩展性有限：此架构由于仅限于单个生产者和单个消费者，因此可扩展性不佳。

2. 多生产者、单消费者 (MPSC)

在多生产者、单消费者 (MPSC) 架构中，多个生产者将数据发送到单个 Kafka 主题，一个消费者从此主题读取数据。此设置用于当来自各种源的数据需要由单个消费者聚合和处理时。

特性

数据聚合：MPSC 非常适合需要由单个消费者集体处理或分析多个源生成数据的场景。
提高吞吐量：通过多个生产者，数据生产速率可以显著提高，从而增加系统的整体吞吐量。
数据倾斜的可能性：由于多个生产者将数据发送到同一个主题，如果管理不当，可能会导致分区之间数据分布不均。

用例

日志聚合：来自多个应用程序或服务器的日志可以发送到单个 Kafka 主题，然后由一个消费者进行处理以进行分析或存储。
物联网数据收集：在物联网场景中，来自各种传感器的数据可以发送到 Kafka，由单个消费者聚合和处理数据。

示例实现

// Producer 1
KafkaProducer<String, String> producer1 = new KafkaProducer<>(producerProps);
producer1.send(new ProducerRecord<>("my-topic", "key1", "value1"));

// Producer 2
KafkaProducer<String, String> producer2 = new KafkaProducer<>(producerProps);
producer2.send(new ProducerRecord<>("my-topic", "key2", "value2"));
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
consumer.subscribe(Arrays.asList("my-topic"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("Consumed record with key %s and value %s%n", record.key(), record.value());
    }
}   

输出

挑战

生产者之间的协调：确保多个生产者的数据在分区之间均匀分布可能具有挑战性。
单消费者瓶颈：如果消费者无法跟上所有生产者的组合数据速率，它可能会成为瓶颈。

3. 单生产者、多消费者 (SPMC)

在单生产者、多消费者 (SPMC) 架构中，一个生产者将数据发送到 Kafka 主题，多个消费者从此主题读取数据，通常作为消费者组的一部分。Kafka 确保主题的每个分区仅由组内的单个消费者消费，从而实现并行处理。

特性

并行处理：SPMC 通过在组内的多个消费者之间分配分区，实现了数据并行消费。
可扩展性：此架构高度可扩展，通过向组添加更多消费者来增加处理能力。
容错性：如果某个消费者发生故障，Kafka 会自动将其分区重新分配给组内的其他消费者，确保数据处理继续进行。

用例

流处理：在需要实时处理数据的场景中很有用，例如交易处理、日志分析或监控系统。
负载平衡：SPMC 通常用于在多个消费者之间平衡处理负载，确保资源的有效利用。

示例实现

// Single Producer (Same as in SPSC)
KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps);
producer.send(new ProducerRecord<>("my-topic", "key", "value"));

// Consumer 1
KafkaConsumer<String, String> consumer1 = new KafkaConsumer<>(consumerProps);
consumer1.subscribe(Arrays.asList("my-topic"));

// Consumer 2
KafkaConsumer<String, String> consumer2 = new KafkaConsumer<>(consumerProps);
consumer2.subscribe(Arrays.asList("my-topic"));

// Processing loop for each consumer
while (true) {
    ConsumerRecords<String, String> records = consumer1.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("Consumer 1 consumed record with key %s and value %s%n", record.key(), record.value());
    }
}   

输出

挑战

分区重新平衡：当消费者加入或离开组时，Kafka 会触发重新平衡，这可能会导致处理延迟。
消费者滞后：如果数据生产速率很高，消费者可能会滞后，导致处理延迟。

4. 多生产者、多消费者 (MPMC)

多生产者、多消费者 (MPMC) 架构是最复杂和最强大的 Kafka 设置。多个生产者将数据发送到 Kafka 主题，多个消费者（组织成消费者组）从主题读取数据。Kafka 在每个组内的消费者之间有效地分配分区，实现了高并行度和吞吐量。

特性

高吞吐量和并行度：MPMC 通过允许多个生产者以高速率生成数据，同时多个消费者并行处理这些数据，从而最大化 Kafka 的能力。
可扩展性：此架构在数据生产和消费方面都高度可扩展。您可以根据需要增加生产者或消费者的数量来处理更多数据或处理任务。
容错性和负载平衡：Kafka 自动处理故障并将负载在消费者之间重新平衡的能力，确保 MPMC 设置的高可用性和可靠性。

用例

分布式数据处理：MPMC 非常适合分布式系统，其中数据从多个源生成，需要由多个消费者并行处理，例如大数据分析、机器学习管道或微服务架构。
实时分析：在需要从多个源进行实时数据分析的场景中（例如，金融交易、社交媒体监控），MPMC 提供了必要的可扩展性和吞吐量。

示例实现

// Producer 1
KafkaProducer<String, String> producer1 = new KafkaProducer<>(producerProps);
producer1.send(new ProducerRecord<>("my-topic", "key1", "value1"));

// Producer 2
KafkaProducer<String, String> producer2 = new KafkaProducer<>(producerProps);
producer2.send(new ProducerRecord<>("my-topic", "key2", "value2"));

// Consumer 1
KafkaConsumer<String, String> consumer1 = new KafkaConsumer<>(consumerProps);
consumer1.subscribe(Arrays.asList("my-topic"));

// Consumer 2
KafkaConsumer<String, String> consumer2 = new KafkaConsumer<>(consumerProps);
consumer2.subscribe(Arrays.asList("my-topic")
// Consumer processing loop for consumer 1
while (true) {
    ConsumerRecords<String, String> records = consumer1.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("Consumer 1 consumed record with key %s and value %s%n", record.key(), record.value());
    }
}

// Consumer processing loop for consumer 2
while (true) {
    ConsumerRecords<String, String> records = consumer2.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("Consumer 2 consumed record with key %s and value %s%n", record.key(), record.value());
    }
}   

输出

挑战

复杂性：管理 MPMC 架构可能很复杂，尤其是在确保数据在分区之间均匀分布、处理重新平衡以及避免数据重复等问题方面。
协调开销：对于多个生产者和消费者，需要增加协调。例如，确保每个消费者被分配适当数量的分区，并且生产者有效地将数据发送到正确的分区。
资源利用：在多个生产者和消费者之间有效利用资源（例如，网络带宽、CPU、内存）需要仔细规划，尤其是在大规模部署中。

Kafka 生产者-消费者架构的实现

Kafka 的生产者-消费者架构是一个强大的框架，用于构建可扩展、容错且高吞吐量的数据处理系统。在实际场景中实现 Kafka 涉及配置和编码生产者和消费者，以有效地与 Kafka 代理进行交互、处理数据以及管理数据流的各个方面。本节介绍了几种关键的实现策略，包括

使用线程和锁
使用信号量
使用条件变量
使用消息队列

每种方法都配有代码示例和实际实现注意事项。

1. 使用线程和锁

在基于 Kafka 的系统中，可以使用多个线程来处理并发数据生产和消费。使用线程和锁可以确保数据得到有效处理，并防止竞态条件或数据不一致。

生产者实现

可以使用多个线程来并发发送数据的 Kafka 生产者。例如，如果您有高吞吐量要求，可以使用多个线程同时向 Kafka 主题生产消息。

示例：多线程 Kafka 生产者

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

import java.util.Properties;

public class MultiThreadedProducer {
    private static final String TOPIC = "my-topic";

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        Runnable producerTask = () -> {
            for (int i = 0; i < 100; i++) {
                ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, "key-" + i, "value-" + i);
                producer.send(record);
            }
        };

        Thread thread1 = new Thread(producerTask);
        Thread thread2 = new Thread(producerTask);

        thread1.start();
        thread2.start();
    }
}   

输出

在此示例中，使用两个线程并发发送消息。由于 Kafka 的生产者 API 内部处理线程安全，因此此处不需要显式锁定。

消费者实现

类似地，可以在消费者应用程序中使用多个线程来并发处理消息。每个线程都可以处理来自 Kafka 主题的消息，从而提高处理效率。

示例：多线程 Kafka 消费者

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

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class MultiThreadedConsumer {
    private static final String TOPIC = "my-topic";
    private static final String GROUP_ID = "my-group";

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
        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(TOPIC));

        Runnable consumerTask = () -> {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("Consumed record with key %s and value %s%n", record.key(), record.value());
                }
            }
        };

        Thread thread1 = new Thread(consumerTask);
        Thread thread2 = new Thread(consumerTask);

        thread1.start();
        thread2.start();
    }
}   

输出

在此实现中，使用两个线程来并发消耗和处理消息。每个线程都会轮询记录并独立处理它们。

注意事项

并发控制：确保您的应用程序设计考虑了并发问题，例如竞态条件，通过使用适当的同步机制（如果需要）。
线程管理：正确管理线程以避免诸如过度资源消耗或线程争用之类的问题。

2. 使用信号量

信号量可用于控制对共享资源的访问，例如 Kafka 主题或缓冲区。它们对于管理并发访问和防止过载很有用。

生产者实现

可以使用信号量来限制并发生产者的数量或控制对共享资源的访问。

示例：带信号量的生产者

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

import java.util.Properties;
import java.util.concurrent.Semaphore;

public class SemaphoreProducer {
    private static final String TOPIC = "my-topic";
    private static final Semaphore semaphore = new Semaphore(10); // Limit to 10 concurrent messages

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        Runnable producerTask = () -> {
            try {
                semaphore.acquire(); // Acquire a permit
                for (int i = 0; i < 100; i++) {
                    ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, "key-" + i, "value-" + i);
                    producer.send(record);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                semaphore.release(); // Release the permit
            }
        };

        Thread thread1 = new Thread(producerTask);
        Thread thread2 = new Thread(producerTask);

        thread1.start();
        thread2.start();
    }
}

输出

在此示例中，使用信号量来限制并发生产者的数量。这有助于管理对 Kafka 代理的负载并防止过载。

消费者实现

信号量也可在消费者中使用，以控制对共享资源的访问或管理并发。

示例：带信号量的消费者

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

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.Semaphore;

public class SemaphoreConsumer {
    private static final String TOPIC = "my-topic";
    private static final String GROUP_ID = "my-group";
    private static final Semaphore semaphore = new Semaphore(5); // Limit to 5 concurrent processes

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
        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(TOPIC));

        Runnable consumerTask = () -> {
            try {
                semaphore.acquire(); // Acquire a permit
                while (true) {
                    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                    for (ConsumerRecord<String, String> record : records) {
                        System.out.printf("Consumed record with key %s and value %s%n", record.key(), record.value());
                    }
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                semaphore.release(); // Release the permit
            }
        };

        Thread thread1 = new Thread(consumerTask);
        Thread thread2 = new Thread(consumerTask);

        thread1.start();
        thread2.start();
    }
}

输出

注意事项

资源管理：确保正确使用信号量来管理资源并防止死锁。
并发控制：使用信号量来控制对共享资源的访问并防止竞态条件。

3. 使用条件变量

条件变量用于基于某些条件同步线程。当线程在继续之前需要等待满足特定条件时，它们很有用。

生产者实现

条件变量可用于指示生产者何时应发送数据，或何时缓冲区已准备好接受新数据。

示例：带条件变量的生产者

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

import java.util.Properties;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class ConditionVariableProducer {
    private static final String TOPIC = "my-topic";
    private static final BlockingQueue<String> buffer = new LinkedBlockingQueue<>();
    private static final Lock lock = new ReentrantLock();
    private static final Condition bufferNotFull = lock.newCondition();

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        Runnable producerTask = () -> {
            try {
                while (true) {
                    lock.lock();
                    try {
                        bufferNotFull.await(); // Wait for the buffer to be not full
                        String message = "message-" + System.currentTimeMillis();
                        buffer.put(message);
                        producer.send(new ProducerRecord<>(TOPIC, message));
                        bufferNotFull.signalAll(); // Signal other threads that the buffer has been updated
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally {
                        lock.unlock();
                    }
                }
            } finally {
                producer.close();
            }
        };

        Thread thread1 = new Thread(producerTask);
        Thread thread2 = new Thread(producerTask);

        thread1.start();
        thread2.start();
    }
}   

输出

消费者实现

消费者可以使用条件变量来指示何时有数据可供处理。

示例：带条件变量的消费者

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

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class ConditionVariableConsumer {
    private static final String TOPIC = "my-topic";
    private static final String GROUP_ID = "my-group";
    private static final BlockingQueue<String> buffer = new LinkedBlockingQueue<>();
    private static final Lock lock = new ReentrantLock();
    private static final Condition bufferNotEmpty = lock.newCondition();

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
        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(TOPIC));

        Runnable consumerTask = () -> {
            try {
                while (true) {
                    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                    for (ConsumerRecord<String, String> record : records) {
                        lock.lock();
                        try {
                            buffer.put(record.value());
                            bufferNotEmpty.signalAll(); // Signal that there is data in the buffer
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        } finally {
                            lock.unlock();
                        }
                    }
                }
            } finally {
                consumer.close();
            }
        };

        Thread thread1 = new Thread(consumerTask);
        Thread thread2 = new Thread(consumerTask);

        thread1.start();
        thread2.start();
    }
}   

输出

注意事项

同步：使用条件变量来处理同步问题并有效管理线程协调。
复杂性：条件变量增加了实现的复杂性。确保它们得到适当使用，以避免诸如信号丢失或死锁之类的问题。

4. 使用消息队列

消息队列可用于解耦生产者和消费者，在它们之间提供缓冲区。Kafka 本身就是一种分布式消息队列，但额外的消息队列系统可用于特定场景与 Kafka 结合使用。

生产者实现

生产者可以使用本地消息队列在将消息发送到 Kafka 之前临时存储它们。

示例：带本地消息队列的生产者

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

import java.util.Properties;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class QueueProducer {
    private static final String TOPIC = "my-topic";
    private static final BlockingQueue<String> messageQueue = new LinkedBlockingQueue<>();

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        Runnable producerTask = () -> {
            try {
                while (true) {
                    String message = messageQueue.take(); // Take a message from the queue
                    producer.send(new ProducerRecord<>(TOPIC, message));
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        };

        // Enqueue messages
        for (int i = 0; i < 100; i++) {
            messageQueue.add("message-" + i);
        }

        Thread producerThread = new Thread(producerTask);
        producerThread.start();
    }
}   

输出

消费者实现

消费者可以使用本地消息队列在处理它们之前缓冲来自 Kafka 的消息。

示例：带本地消息队列的消费者

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

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class QueueConsumer {
    private static final String TOPIC = "my-topic";
    private static final String GROUP_ID = "my-group";
    private static final BlockingQueue<String> messageQueue = new LinkedBlockingQueue<>();

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
        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(TOPIC));

        Runnable consumerTask = () -> {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    try {
                        messageQueue.put(record.value()); // Add record to the queue
                    } catch (InterruptedException e) {
                        e.printStackTrace();   }
                } }
        };

        Runnable processorTask = () -> {
            while (true) {
                try {
                    String message = messageQueue.take(); // Process message from the queue
                    System.out.printf("Processed record: %s%n", message);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };

        Thread consumerThread = new Thread(consumerTask);
        Thread processorThread = new Thread(processorTask);

        consumerThread.start();
        processorThread.start();
    }
}   

输出

注意事项

解耦：消息队列有助于解耦生产者和消费者，使系统对故障更具弹性。
性能：确保消息队列不会成为系统中的瓶颈。

Kafka 生产者-消费者架构的实际应用

Apache Kafka 的生产者-消费者架构被广泛应用于各种行业，用于构建可扩展的实时数据处理系统。以下是一些关键的实际应用

1. 操作系统

在操作系统中，Kafka 用于高效地管理日志和系统指标。

应用示例：系统日志聚合

现代操作系统会生成大量与系统事件、用户活动和应用程序错误相关的日志。需要实时收集、处理和分析这些日志以进行监控和故障排除。

示例配置

// Producer configuration for logging
Properties producerProps = new Properties();
producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps);

// Send log messages
ProducerRecord<String, String> logRecord = new ProducerRecord<>("system-logs", "INFO", "System started successfully.");
producer.send(logRecord);
// Consumer configuration for processing logs
Properties consumerProps = new Properties();
consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "log-processing-group");
consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
consumer.subscribe(Collections.singletonList("system-logs"));

// Process log messages
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("Received log: %s%n", record.value());
    }
}   

输出

2. 日志处理系统

Kafka 被广泛用于日志处理系统，以聚合和分析来自各种源的日志。

应用示例：集中式日志管理

大型组织通常有多个应用程序和服务生成日志。Kafka 提供了一个集中式平台来聚合这些日志，从而更容易进行分析和生成洞察。

3. 实时数据流

Kafka 的架构非常适合实时数据流应用程序，在这些应用程序中，数据需要被立即处理和响应。

应用示例：金融市场数据

在金融市场中，股票价格、交易量和其他指标等数据是实时生成的，需要进行处理以做出交易决策。

4. 分布式系统

在分布式系统中，Kafka 提供了一个可靠且可扩展的消息传递系统，用于处理服务间通信和数据集成。

应用示例：微服务通信

微服务通常需要相互通信并共享数据。Kafka 作为消息代理，实现了微服务之间可靠的通信和数据共享。

示例配置

// Producer for microservices events
Properties producerProps = new Properties();
producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps);

// Send event data
ProducerRecord<String, String> eventRecord = new ProducerRecord<>("service-events", "serviceA", "Event occurred");
producer.send(eventRecord);
java
Copy code
// Consumer for microservices events
Properties consumerProps = new Properties();
consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "service-consumer-group");
consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
consumer.subscribe(Collections.singletonList("service-events"));

// Process events
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        // Handle event
    }
}   

输出

下一主题Kafka 复制

Python

Java

.Net Framework

AI, ML and Data Science

Cloud Technology

B.Tech and MCA

Web Technology

PHP

Software Testing

Technical Interview

Java Interview

Python

Web Interview

Database Interview

B.Tech / MCA

Important Interview

Software Testing Interview

Company Interviews

Online Compilers

Multiple Choice Questions

Kafka 教程

Kafka 安装

Kafka CLI

Kafka 编程

实时示例

Kafka 监控

Kafka Connect

Kafka Streams

杂项

Kafka 生产者消费者架构

Kafka 生产者-消费者架构简介

关键概念

Kafka 生产者-消费者架构的关键组件

1. 生产者

2. 消费者

3. 主题和分区

4. 代理

5. Zookeeper

Kafka 生产者-消费者架构的类型

1. 单生产者、单消费者 (SPSC)

特性

用例

示例实现

2. 多生产者、单消费者 (MPSC)

特性

用例

示例实现

3. 单生产者、多消费者 (SPMC)

特性

用例

示例实现

挑战

4. 多生产者、多消费者 (MPMC)

特性

用例

示例实现

挑战

Kafka 生产者-消费者架构的实现

1. 使用线程和锁

生产者实现

消费者实现

注意事项

2. 使用信号量

生产者实现

消费者实现

3. 使用条件变量

消费者实现

4. 使用消息队列

生产者实现

消费者实现

Kafka 生产者-消费者架构的实际应用

1. 操作系统

2. 日志处理系统

3. 实时数据流

4. 分布式系统

相关帖子

ZooKeeper 摘要身份验证

Apache Kafka vs Apache Storm

使用 Kafka 进行混合云数据编排

Kafka-to-Cosmos DB 管道中的模式演进和序列化