.NET Core + Kafka 开发指南
https://cdn.nlark.com/yuque/0/2025/jpeg/359374/1735804678982-fdba62e2-0bda-4198-82d9-6848357435b5.jpeg什么是Kafka
Apache Kafka是一个分布式流处理平台，由LinkedIn开发并开源，后来成为Apache软件基金会的顶级项目。Kafka主要用于构建实时数据管道和流式应用程序。
Kafka 架构
从下面3张架构图中可以看出Kafka Server 实际扮演的是Broker的角色， 一个Kafka Cluster由多个Broker组成， 或者可以说是多个Topic组成。
图 1
https://cdn.nlark.com/yuque/0/2025/png/359374/1735803149887-a2bad129-d707-4090-885b-87687d78f76f.png
图 2
https://cdn.nlark.com/yuque/0/2025/svg/359374/1735803646595-0f47a20f-9ee5-409f-b3ca-82d02a689627.svg
图 3
https://cdn.nlark.com/yuque/0/2025/png/359374/1735803258104-67e3ad27-ca07-48a5-99e8-8cdab9b6b17a.png
主要概念(Main Concepts)和术语(Terminology)
Kafka Cluster
一个Kafka集群是一个由多个Kafka代理组成的分布式系统，它们协同工作以处理实时流数据的存储和处理。它为大规模应用程序中高效的数据流和消息传递提供了容错性、可扩展性和高可用性。
Broker
Broker是构成Kafka集群的服务器。 每个Broker负责接收、存储和提供数据。 它们处理来自生产者和消费者的读写操作。 Broker还管理数据的复制以确保容错性。
Topic and Partitions
Kafka中的数据被组织成主题（Topics），这些是生产者发送数据和消费者读取数据的逻辑通道。每个主题被划分为分区（partitions），它们是Kafka中并行处理的基本单位。分区允许Kafka通过在多个Broker 之间分布数据来水平扩展。
Producers
生产者是发布（写入）数据到Kafka主题的客户端应用程序。它们根据分区策略将记录发送到适当的主题和分区，分区策略可以是基于键（key-based）或轮询（round-robin）。
Consumers
消费者是订阅Kafka主题并处理数据的客户端应用程序。它们从主题中读取记录，并且可以是消费者组的一部分，这允许负载均衡和容错。每个组中的消费者从一组独特的分区中读取数据。
Zookeeper
ZooKeeper是一个集中式服务，用于维护配置信息、命名、提供分布式同步和提供群组服务。在Kafka中，ZooKeeper用于管理和协调Kafka Broker。ZooKeeper被展示为与Kafka集群交互的独立组件。
Offsets
偏移量（offsets）是分配给分区中每条消息的唯一标识符。消费者将使用这些偏移量来跟踪他们在消费主题中消息的进度。
Kafka vs RabbitMQ
相同点
[*]消息队列功能
[*]Kafka和RabbitMQ都是流行的消息队列工具，支持生产者-消费者模式，能够解耦系统，提高系统的可扩展性和可靠性。
[*]异步通信
[*]两者都支持异步通信，允许生产者发送消息后立即返回，消费者可以异步处理消息。
[*]多种消息传递模式
[*]均支持点对点（P2P）和发布/订阅（Pub/Sub）模式。
[*]持久化支持
[*]Kafka和RabbitMQ都支持消息的持久化，以确保在系统故障或重启后消息不会丢失。
[*]高可用性
[*]两者都支持集群部署，具有高可用性和容错能力。
[*]语言支持
[*]提供多种语言的客户端库，支持不同编程语言的集成。
<hr>不同点
架构和设计
[*]数据存储
[*]Kafka：基于日志分区存储设计，适合高吞吐量的顺序读写。
[*]RabbitMQ：基于AMQP协议，消息存储在队列中，适合低延迟的场景。
[*]消息消费模式
[*]Kafka：消息由消费者主动拉取，支持多次消费。
[*]RabbitMQ：消息通过推送方式传递给消费者，消费后消息默认从队列中移除。
[*]使用场景
[*]Kafka：适用于大数据场景（日志聚合、流式处理），擅长处理高吞吐量、大规模消息传递。
[*]RabbitMQ：适用于需要复杂路由和消息确认的场景（如事务性消息和实时通信）。
性能与延迟
[*]高吞吐量
[*]Kafka：设计针对高吞吐量场景优化，能够支持百万级消息每秒。
[*]RabbitMQ：吞吐量相对较低，但延迟更低。
[*]延迟
[*]Kafka：适合高吞吐量但对实时性要求不高的应用。
[*]RabbitMQ：更适合低延迟应用，提供实时性支持。
协议支持
[*]协议类型
[*]Kafka：自定义的二进制协议。
[*]RabbitMQ：基于AMQP协议，支持丰富的消息功能（如TTL、优先级）。
[*]兼容性
[*]Kafka：需要Kafka专用客户端。
[*]RabbitMQ：支持AMQP标准协议，兼容性较强。
开发一个Producer和一个Consumer
本地docker环境启动一个kafka
version: '2'services:zookeeper: image: confluentinc/cp-zookeeper:7.4.4 environment: ZOOKEEPER_CLIENT_PORT: 2181 ZOOKEEPER_TICK_TIME: 2000 ports: - 22181:2181 kafka: image: confluentinc/cp-kafka:7.4.4 depends_on: - zookeeper ports: - 29092:29092 environment: KAFKA_BROKER_ID: 1 KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181 KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092 KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1使用.NET CORE + Kafka开发一个消息生产者， 一个消息消费者， 客户端需要安装组件** Confluent.Kafka**
InventoryUpdateProducer
public class ProducerService{ private readonly IConfiguration _configuration; private readonly IProducer<Null, string> _producer; private readonly ILogger<ProducerService> _logger; public ProducerService(IConfiguration configuration, ILogger<ProducerService> logger) { _configuration = configuration; _logger = logger; var config = new ProducerConfig { BootstrapServers = _configuration["Kafka:BootstrapServers"], }; _producer = new ProducerBuilder<Null, string>(config).Build(); } public async Task ProductAsync(string topic, string message) { var orderPlacedMessage = new Message<Null, string> { Value = message }; await _producer.ProduceAsync(topic, orderPlacedMessage); _logger.LogInformation("Message sent to topic: {Topic}", topic); }}")]public class InventoryController : ControllerBase{ private readonly ProducerService _producerService; public InventoryController(ProducerService producerService) { _producerService = producerService; } public async Task<IActionResult> Post( InventoryUpdateRequest request) { var message = System.Text.Json.JsonSerializer.Serialize(request); await _producerService.ProductAsync("inventory-update", message); return Ok("Inventory Updated Successfully..."); }}启动项目，查看Swagger
https://cdn.nlark.com/yuque/0/2025/png/359374/1735876009405-4b883e17-5f7d-46da-834d-1a8ad6e67009.png
InventoryUpdateConsumer
消息消费者程序使用.net core BackgroundService开发， 这个类需要在程序启动时注入进去，不要忘记。
public class ConsumerService : BackgroundService{ private readonly ILogger<ConsumerService> _logger; private readonly IConfiguration _configuration; private readonly IConsumer<Ignore, string> _consumer; public ConsumerService(ILogger<ConsumerService> logger, IConfiguration configuration) { _logger = logger; _configuration = configuration; var consumerConfig = new ConsumerConfig { BootstrapServers = configuration["Kafka:BootstrapServers"], GroupId = "InventoryConsumerGroup", AutoOffsetReset = AutoOffsetReset.Earliest }; _consumer = new ConsumerBuilder<Ignore, string>(consumerConfig).Build(); } protected override async Task ExecuteAsync(CancellationToken stoppingToken) { _consumer.Subscribe("inventory-update"); try { while (!stoppingToken.IsCancellationRequested) { HandleMessage(stoppingToken); await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken); } } catch (OperationCanceledException) { _logger.LogInformation("Consumer service has been cancelled."); } catch (Exception ex) { _logger.LogError($"Error in consuming messages: {ex.Message}"); } finally { _consumer.Close(); } } public void HandleMessage(CancellationToken cancellation) { try { var consumeResult = _consumer.Consume(cancellation); var message = consumeResult.Message.Value; _logger.LogInformation($"Received inventory update: {message}"); } catch (Exception ex) { _logger.LogError($"Error processing Kafka message: {ex.Message}"); } }}var builder = WebApplication.CreateBuilder(args);builder.Services.AddHostedService<ConsumerService>();运行程序
Publish Message
https://cdn.nlark.com/yuque/0/2025/png/359374/1735876252374-5f2e13db-2087-49ab-8f9b-d873c35d6c52.png
Consume Message
https://cdn.nlark.com/yuque/0/2025/png/359374/1735876252325-57845904-230e-43f3-8e1c-e2ccf0967691.png
总结
Apache Kafka不是消息中间件的一种实现。相反，它只是一种分布式流式系统。 不同于基于队列和交换器的RabbitMQ，Kafka的存储层是使用分区事务日志来实现的。Kafka也提供流式API用于实时的流处理以及连接器API用来更容易的和各种数据源集成。
页:
[1]