Kafka使用consumer.Assign指定分区消费无数据问题求助

搭建了一套Kafka架构，涉及多台产生电力、气压、除尘及设备状态数据的设备。为每台设备创建一个Topic，每个Topic设置4个分区，其中气压数据写入分区0、电力数据写入分区1、除尘数据写入分区2、设备状态数据写入分区3。编写了4个消费者，通过consumer.Assign()分别订阅对应Topic的指定分区，但无法接收到数据。已确认生产者已将正确数据写入对应分区，若订阅整个Topic则能正常接收数据。

气压消费者服务代码 #

using System.IO.Compression;
using Confluent.Kafka;
using EnergyMeasurement.Models;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
 public class AirPressureConsumerService : BackgroundService
    {
        private readonly ILogger<AirPressureConsumerService> logger;
        private readonly List<string> topics;
        private readonly string groupId = "coldPath";
        private readonly string bootstrapServers = "localhost:9092";

        public AirPressureConsumerService(ILogger<AirPressureConsumerService> logger)
        {
            this.logger = logger;
            this.topics = new List<string> { "thingId1", "thingId2" };
        }

      protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        var config = new ConsumerConfig
        {
            BootstrapServers = this.bootstrapServers,
            GroupId = this.groupId,
            AutoOffsetReset = AutoOffsetReset.Earliest
        };

        using var consumer = new ConsumerBuilder<Ignore, byte[]>(config).Build();

        foreach (var topic in this.topics)
        {
            var topicPartition = new TopicPartition(topic, 0);
            consumer.Assign(topicPartition);

            while (!stoppingToken.IsCancellationRequested)
            {
                var consumeResult = consumer.Consume(stoppingToken);

                var compressedDto = CompressedDTO.Parser.ParseFrom(consumeResult.Message.Value);
                var contentBytes = Decompress(compressedDto.CompressedData.ToByteArray());

                var airPressureDTO = AirPressureDTO.Parser.ParseFrom(contentBytes);

                Console.WriteLine($"Received message at {consumeResult.TopicPartitionOffset}: {airPressureDTO}");
                await Task.Delay(1000, stoppingToken);
            }
        }

        consumer.Close();
    }

        private byte[] Decompress(byte[] data)
        {
            using var compressedStream = new MemoryStream(data);
            using var decompressStream = new GZipStream(compressedStream, CompressionMode.Decompress);
            using var resultStream = new MemoryStream();
            decompressStream.CopyTo(resultStream);
            return resultStream.ToArray();
        }
    }

主程序代码 #

using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;

public class Program
{
    public static void Main(string[] args)
    {
        CreateHostBuilder(args).Build().Run();
    }

    public static IHostBuilder CreateHostBuilder(string[] args) =>
        Host.CreateDefaultBuilder(args)
            .ConfigureServices((hostContext, services) =>
            {
                services.AddHostedService<ElectricConsumerService>();
                services.AddHostedService<AirPressureConsumerService>();
                services.AddHostedService<MachineStatusConsumerService>();
                services.AddHostedService<DustExtractionConsumerService>();
                services.AddHostedService<TestConsumer>();
            });
}

核心问题 #

代码中的foreach循环逻辑存在两处关键错误：

1. 分区分配被覆盖：每次调用consumer.Assign()会替换之前的分区分配，而非追加，最终只有最后一次调用的分区会被保留，但实际循环在第一个Topic的while循环中就陷入无限执行，永远无法处理后续Topic的分区。
2. 无法同时监听多Topic分区：原逻辑试图逐个分配分区并消费，导致同一时间只能监听单个Topic的单个分区，不符合多Topic分区监听的需求。

修正方案 #

需要一次性收集所有目标TopicPartition，通过一次consumer.Assign()完成所有分区的分配，再进入统一的消费循环：

修正后的气压消费者服务代码 #

using System.IO.Compression;
using Confluent.Kafka;
using EnergyMeasurement.Models;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using System.Collections.Generic;

public class AirPressureConsumerService : BackgroundService
{
    private readonly ILogger<AirPressureConsumerService> logger;
    private readonly List<string> topics;
    private readonly string groupId = "coldPath";
    private readonly string bootstrapServers = "localhost:9092";

    public AirPressureConsumerService(ILogger<AirPressureConsumerService> logger)
    {
        this.logger = logger;
        this.topics = new List<string> { "thingId1", "thingId2" };
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        var config = new ConsumerConfig
        {
            BootstrapServers = this.bootstrapServers,
            GroupId = this.groupId,
            AutoOffsetReset = AutoOffsetReset.Earliest
        };

        using var consumer = new ConsumerBuilder<Ignore, byte[]>(config).Build();

        // 收集所有需要监听的TopicPartition：每个Topic的分区0
        var topicPartitions = new List<TopicPartition>();
        foreach (var topic in this.topics)
        {
            topicPartitions.Add(new TopicPartition(topic, 0));
        }

        // 一次性分配所有目标分区
        consumer.Assign(topicPartitions);

        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                var consumeResult = consumer.Consume(stoppingToken);

                var compressedDto = CompressedDTO.Parser.ParseFrom(consumeResult.Message.Value);
                var contentBytes = Decompress(compressedDto.CompressedData.ToByteArray());

                var airPressureDTO = AirPressureDTO.Parser.ParseFrom(contentBytes);

                Console.WriteLine($"Received message at {consumeResult.TopicPartitionOffset}: {airPressureDTO}");
                await Task.Delay(1000, stoppingToken);
            }
            catch (ConsumeException ex)
            {
                logger.LogError(ex, "消费消息时发生错误");
            }
        }

        consumer.Close();
    }

    private byte[] Decompress(byte[] data)
    {
        using var compressedStream = new MemoryStream(data);
        using var decompressStream = new GZipStream(compressedStream, CompressionMode.Decompress);
        using var resultStream = new MemoryStream();
        decompressStream.CopyTo(resultStream);
        return resultStream.ToArray();
    }
}

额外注意事项 #

1. 手动提交偏移量：使用Assign()手动分配分区时，Kafka不会自动提交偏移量，若需要持久化消费进度，需在消费成功后调用consumer.Commit(consumeResult)手动提交。
2. 其他消费者同步修正：电力、除尘、设备状态三类消费者需按相同逻辑修改，分别对应分区1、2、3。
3. 消费者组分区冲突：同一消费者组内的消费者不要重复分配相同分区，避免重复消费或消费冲突。

内容的提问来源于stack exchange，提问作者tebrex_rtk