如何在.NET Core Kafka流应用中添加多消费者消费同一主题？

阿华AIGC实验室

2026-5-13

解决.NET Core Kafka多消费者消费同一主题的问题

首先，要实现多个消费者消费同一主题，核心需要两个条件：主题有多个分区，以及所有消费者属于同一个消费组（相同group-id）。下面是具体的操作步骤和代码调整方案：

1. 确保Kafka主题具备多个分区

Kafka的消费组机制是将主题的分区分配给组内的消费者，每个分区只能被组内的一个消费者消费。所以首先要保证你的orderrequests主题有足够的分区（数量至少等于你要启动的消费者数量）。

创建/修改主题分区

如果主题还未创建，用Kafka命令行工具创建带多分区的主题（以3个分区为例）：

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

如果主题已经存在，只能增加分区数（分区数无法减少）：

kafka-topics.sh --alter --topic orderrequests --bootstrap-server localhost:9092 --partitions 3

2. 调整消费者配置与代码

你的appsettings.json里的groupid已经设置为csharp-consumer，这个保持不变——所有要加入同一消费组的消费者必须使用相同的group-id。接下来需要修正代码里的消费者实例创建逻辑：

问题点分析

你当前的ProcessOrdersService在循环里每次创建新的ConsumerWrapper，这会导致重复订阅、资源浪费，而且无法稳定实现多消费者负载均衡。正确的做法是每个消费者实例作为独立的后台服务，且每个实例只初始化一次消费者。

修改ProcessOrdersService

调整ExecuteAsync方法，只初始化一次消费者，并持续消费，同时添加异常处理和资源释放：

namespace Api.Services {
  public class ProcessOrdersService : BackgroundService {
    private readonly ConsumerConfig consumerConfig;
    private readonly ProducerConfig producerConfig;
    public ProcessOrdersService(ConsumerConfig consumerConfig, ProducerConfig producerConfig) {
      this.producerConfig = producerConfig;
      this.consumerConfig = consumerConfig;
    }
    protected override async Task ExecuteAsync(CancellationToken stoppingToken) {
      Console.WriteLine($"OrderProcessing Service Started (Thread: {Thread.CurrentThread.ManagedThreadId})\n\n");
      // 仅初始化一次消费者
      using var consumerHelper = new ConsumerWrapper(consumerConfig, "orderrequests");
      
      while (!stoppingToken.IsCancellationRequested) {
        try {
          string orderRequest = consumerHelper.readMessage();
          OrderRequest order = JsonConvert.DeserializeObject<OrderRequest>(orderRequest);
          Console.WriteLine($"Info: OrderHandler => Processing order for {order.productname} (Consumer Thread: {Thread.CurrentThread.ManagedThreadId})\n\n");
          order.status = OrderStatus.COMPLETED;
          
          // 建议复用生产者实例，而不是每次创建（可选优化）
          using var producerWrapper = new ProducerWrapper(producerConfig,"readytoship");
          await producerWrapper.writeMessage(JsonConvert.SerializeObject(order));
        } catch (ConsumeException ex) {
          Console.WriteLine($"Consume error: {ex.Error.Reason}");
        } catch (Exception ex) {
          Console.WriteLine($"Unexpected error: {ex.Message}");
        }
      }
    }
  }
}

给ConsumerWrapper添加资源释放

实现IDisposable接口，确保消费者能正确关闭和释放资源：

namespace Api {
  public class ConsumerWrapper : IDisposable {
    private string _topicName;
    private ConsumerConfig _consumerConfig;
    private Consumer<string,string> _consumer;
    private static readonly Random rand = new Random();
    
    public ConsumerWrapper(ConsumerConfig config,string topicName) {
      this._topicName = topicName;
      this._consumerConfig = config;
      this._consumer = new Consumer<string,string>(this._consumerConfig);
      this._consumer.Subscribe(topicName);
    }
    
    public string readMessage(){
      var consumeResult = this._consumer.Consume();
      return consumeResult.Value;
    }
    
    public void DisplayMessage() {
      var consumeResult = this._consumer.Consume();
      Console.WriteLine(consumeResult.Value);
      Console.WriteLine($"Info: OrderHandler => Delivered the order for {consumeResult.Value}\n\n");
    }
    
    // 实现IDisposable释放消费者资源
    public void Dispose() {
      _consumer.Unsubscribe();
      _consumer.Close();
      _consumer.Dispose();
    }
  }
}

3. 启动多个消费者实例

要让多个消费者同时运行，你需要在应用中注册多个ProcessOrdersService实例。在Program.cs里添加多次注册（数量和主题分区数一致即可）：

// 注册3个消费者实例，对应3个分区
builder.Services.AddHostedService<ProcessOrdersService>();
builder.Services.AddHostedService<ProcessOrdersService>();
builder.Services.AddHostedService<ProcessOrdersService>();