Kafka工作原理学习

Linux运维技术之路

发布于 2025-02-25 02:30:02

16900

代码可运行

文章被收录于专栏：Linux运维技术之路Linux运维技术之路

运行总次数：0

代码可运行

Kafka 工作原理

Apache Kafka 是一个分布式流处理平台，最初由 LinkedIn 开发，并于 2011 年开源。Kafka 的设计目标是提供高吞吐量、低延迟的消息传递系统，适用于大规模数据流处理和实时分析。它广泛应用于日志收集、监控数据聚合、流式数据处理等场景。

核心概念

1. Producer（生产者）：负责将消息发布到 Kafka 集群中的某个 Topic。
2. Consumer（消费者）：从 Kafka 集群中订阅 Topic，并消费消息。
3. Topic（主题）：消息的分类或通道，生产者将消息发送到特定的 Topic，消费者从 Topic 中读取消息。
4. Partition（分区）：每个 Topic 可以分为多个 Partition，Partition 是 Kafka 实现并行处理和扩展性的关键。每个 Partition 是一个有序的、不可变的消息序列。
5. Broker（代理）：Kafka 集群中的每个节点称为 Broker，负责存储和传递消息。
6. Offset（偏移量）：每个 Partition 中的消息都有一个唯一的 Offset，表示消息在 Partition 中的位置。
7. Consumer Group（消费者组）：一组消费者共同消费一个 Topic 的消息，每个 Partition 只能被同一个 Consumer Group 中的一个消费者消费。

工作流程

1. 消息生产：Producer 将消息发送到指定的 Topic，Kafka 会根据 Partition 的策略（如轮询、哈希等）将消息分配到不同的 Partition。
2. 消息存储：每个 Partition 的消息会被持久化存储在 Broker 上，Kafka 使用顺序写入的方式提高性能。
3. 消息消费：Consumer 从指定的 Topic 和 Partition 中拉取消息，并通过 Offset 记录消费进度。Kafka 支持消费者组的概念，允许多个消费者协同工作。
4. 消息删除：Kafka 会根据配置的保留策略（如时间或大小限制）定期删除旧的消息。

特性

• 高吞吐量：Kafka 能够处理每秒数百万条消息，适合大规模数据流处理。
• 持久化存储：Kafka 将消息持久化到磁盘，支持消息的重放和容错。
• 分布式架构：Kafka 是分布式的，支持水平扩展，能够处理大规模数据。
• 多消费者支持：Kafka 支持多个消费者组同时消费同一个 Topic 的消息，互不影响。

案例说明

场景：电商平台订单处理系统

假设我们有一个电商平台，用户下单后需要进行一系列处理，包括订单验证、库存检查、支付处理、物流安排等。为了提高系统的可扩展性和解耦，我们可以使用 Kafka 来构建一个基于消息队列的订单处理系统。

系统架构

1. 订单服务（Order Service）：当用户下单时，订单服务将订单信息发送到 Kafka 的 order-topic。
2. 验证服务（Validation Service）：从 order-topic 中消费订单消息，验证订单的合法性（如用户信息、商品信息等），并将验证结果发送到 validation-result-topic。
3. 库存服务（Inventory Service）：从 validation-result-topic 中消费验证通过的订单，检查库存是否充足，并将结果发送到 inventory-result-topic。
4. 支付服务（Payment Service）：从 inventory-result-topic 中消费库存检查通过的订单，处理支付逻辑，并将支付结果发送到 payment-result-topic。
5. 物流服务（Logistics Service）：从 payment-result-topic 中消费支付成功的订单，安排物流配送。

数据流图

用户下单 -> Order Service -> [Kafka: order-topic] 
                            -> Validation Service -> [Kafka: validation-result-topic]
                            -> Inventory Service -> [Kafka: inventory-result-topic]
                            -> Payment Service -> [Kafka: payment-result-topic]
                            -> Logistics Service

实现步骤

1. 订单服务（Producer）：

• 当用户下单时，订单服务将订单信息（如订单号、用户ID、商品ID、数量等）作为消息发送到 Kafka 的 order-topic。
• 示例代码（Java）：Properties props=newProperties(); 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"); Producer<String, String> producer = newKafkaProducer<>(props); Stringtopic="order-topic"; StringorderId="order-123"; StringorderDetails="{\"orderId\":\"" + orderId + "\", \"userId\":\"user-456\", \"productId\":\"product-789\", \"quantity\":2}"; ProducerRecord<String, String> record = newProducerRecord<>(topic, orderId, orderDetails); producer.send(record); producer.close();

2. 验证服务（Consumer）：

• 验证服务订阅 order-topic，消费订单消息，验证订单的合法性（如检查用户是否存在、商品是否有效等），并将验证结果发送到 validation-result-topic。
• 示例代码（Java）：Properties props=newProperties(); props.put("bootstrap.servers", "localhost:9092"); props.put("group.id", "validation-group"); props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); KafkaConsumer<String, String> consumer = newKafkaConsumer<>(props); consumer.subscribe(Arrays.asList("order-topic")); while (true) { ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100)); for (ConsumerRecord<String, String> record : records) { StringorderId= record.key(); StringorderDetails= record.value(); // 验证订单逻辑 booleanisValid= validateOrder(orderDetails); if (isValid) { // 发送验证结果到 validation-result-topic sendToKafka("validation-result-topic", orderId, orderDetails); } } }