基于.NET C#的Kafka带元数据文件流生产消费技术咨询

一、大文件分片处理（针对100MB文件） #

Kafka单条消息过大会影响性能，建议对100MB文件做分片处理：

• 按固定大小（比如1MB）拆分文件，每个分片作为一条Kafka消息
• 每条消息携带核心元数据：文件唯一ID、分片索引、总分片数、文件类型、原始文件名等
• 消费者端通过文件ID聚合分片，拼接成完整文件后再处理

示例分片逻辑代码：

// 文件分片方法
public IEnumerable<(byte[] Chunk, FileMetadata Metadata)> SplitFile(string filePath, int chunkSize = 1024 * 1024)
{
    var fileInfo = new FileInfo(filePath);
    var metadata = new FileMetadata
    {
        FileId = Guid.NewGuid().ToString(),
        FileType = GetFileType(filePath), // 自定义方法判断邮件/物理文件等类型
        TotalChunks = (int)Math.Ceiling(fileInfo.Length / (double)chunkSize),
        OriginalFileName = fileInfo.Name
    };

    using var stream = new FileStream(filePath, FileMode.Open);
    var buffer = new byte[chunkSize];
    int bytesRead;
    int chunkIndex = 0;

    while ((bytesRead = stream.Read(buffer, 0, chunkSize)) > 0)
    {
        var chunk = new byte[bytesRead];
        Array.Copy(buffer, chunk, bytesRead);
        metadata.ChunkIndex = chunkIndex++;
        yield return (chunk, metadata);
    }
}

// 元数据类（对应Schema Registry的Schema结构）
public class FileMetadata
{
    public string FileId { get; set; }
    public string FileType { get; set; }
    public int TotalChunks { get; set; }
    public int ChunkIndex { get; set; }
    public string OriginalFileName { get; set; }
}

二、Schema Registry 元数据设计 #

建议用Avro作为元数据的序列化格式（Schema Registry对Avro支持最完善）：

• 定义Avro Schema描述元数据结构：

{
  "type": "record",
  "name": "FileMetadata",
  "namespace": "YourAppNamespace",
  "fields": [
    {"name": "FileId", "type": "string"},
    {"name": "FileType", "type": "string"},
    {"name": "TotalChunks", "type": "int"},
    {"name": "ChunkIndex", "type": "int"},
    {"name": "OriginalFileName", "type": "string"}
  ]
}

• 生产者将Schema注册到Registry，发送消息时用Schema序列化元数据；消费者通过Registry获取Schema反序列化元数据
• 可以把分片字节和元数据封装成一个Avro记录，统一序列化发送

三、.NET生产者实现要点 #

使用Confluent.Kafka和Confluent.SchemaRegistry.Serdes.Avro包，重点配置：

• 调整MessageMaxBytes，和Broker端的message.max.bytes、replica.fetch.max.bytes匹配（确保能容纳分片消息）
• 用文件ID作为消息Key，确保同一文件的分片发送到同一个Partition，方便消费者聚合

示例生产者代码片段：

var producerConfig = new ProducerConfig
{
    BootstrapServers = "localhost:9092",
    MessageMaxBytes = 1024 * 1024 + 100000, // 略大于分片大小，预留元数据空间
    SchemaRegistryUrl = "localhost:8081",
    CompressionType = CompressionType.Gzip // 启用压缩减少带宽消耗
};

// FileChunk是包含元数据和分片字节的Avro记录
using var producer = new ProducerBuilder<string, FileChunk>(producerConfig)
    .SetValueSerializer(new AvroSerializer<FileChunk>(producerConfig))
    .Build();

foreach (var (chunk, metadata) in SplitFile("path/to/target/file"))
{
    var fileChunk = new FileChunk { Metadata = metadata, ChunkData = chunk };
    await producer.ProduceAsync("file-processing-topic", new Message<string, FileChunk> { Key = metadata.FileId, Value = fileChunk });
}

四、多消费者按文件类型处理 #

所有消费者订阅同一个主题，通过消息过滤实现分类型处理：

• 消费者配置AutoOffsetReset = AutoOffsetReset.Earliest，避免丢失消息
• 收到消息后先判断Metadata.FileType，只处理自身负责的类型（比如邮件消费者只处理FileType = "Email"的消息）
• 维护本地缓存聚合分片，当分片数达到总分片数时，拼接成完整文件再处理

示例消费者代码片段：

var consumerConfig = new ConsumerConfig
{
    BootstrapServers = "localhost:9092",
    GroupId = "email-file-consumer-group",
    AutoOffsetReset = AutoOffsetReset.Earliest,
    SchemaRegistryUrl = "localhost:8081"
};

// 缓存待聚合的文件分片
var fileAggregator = new Dictionary<string, (FileMetadata Metadata, List<byte[]> Chunks)>();

using var consumer = new ConsumerBuilder<string, FileChunk>(consumerConfig)
    .SetValueDeserializer(new AvroDeserializer<FileChunk>(consumerConfig).AsSyncOverAsync())
    .Build();

consumer.Subscribe("file-processing-topic");

while (true)
{
    var consumeResult = consumer.Consume();
    var fileChunk = consumeResult.Message.Value;
    var metadata = fileChunk.Metadata;

    // 只处理邮件类型文件
    if (metadata.FileType != "Email") continue;

    if (!fileAggregator.ContainsKey(metadata.FileId))
    {
        fileAggregator[metadata.FileId] = (metadata, new List<byte[]>());
    }

    var entry = fileAggregator[metadata.FileId];
    entry.Chunks.Insert(metadata.ChunkIndex, fileChunk.ChunkData);

    // 检查是否收集完所有分片
    if (entry.Chunks.Count == metadata.TotalChunks)
    {
        // 拼接完整文件
        var fullFileBytes = entry.Chunks.SelectMany(c => c).ToArray();
        await ProcessEmailFile(fullFileBytes, metadata); // 自定义邮件业务处理逻辑

        // 清理缓存，避免内存占用过高
        fileAggregator.Remove(metadata.FileId);
    }
}

五、Docker Compose环境优化 #

调整Kafka和Schema Registry的配置，适配大分片消息：

services:
  kafka:
    image: confluentinc/cp-kafka:latest
    environment:
      KAFKA_MESSAGE_MAX_BYTES: 1572864 # 设置为1.5MB，大于1MB分片大小
      KAFKA_REPLICA_FETCH_MAX_BYTES: 1572864
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092,PLAINTEXT_INTERNAL://kafka:29092
    ports:
      - "9092:9092"
  schema-registry:
    image: confluentinc/cp-schema-registry:latest
    environment:
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: PLAINTEXT://kafka:29092
      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081
    ports:
      - "8081:8081"
    depends_on:
      - kafka

六、关键注意事项 #

• 分片丢失处理：给消息设置合理的过期时间，定期清理未完成的聚合缓存；可以将已处理的文件ID记录到Redis，避免消费者重启后重复处理
• 性能优化：启用消息压缩，根据服务器配置调整分片大小，避免过大或过小
• 错误处理：生产者要处理发送失败的重试逻辑，消费者要处理消息反序列化失败、分片缺失等异常

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