使用Confluent平台实现C#与ksql Avro格式的序列化/反序列化

核心问题根因 #

• Expecting data framing of length 5 bytes or more报错：你给消息Key配置了AvroDeserializer，但ksqlDB表的主键（对应Kafka消息Key）默认使用原生Int32序列化，没有Avro协议要求的「1字节魔法位+4字节Schema ID」5字节前缀，只有4字节的Int32值，触发长度校验失败
• 发送无报错但消息未投递：多为Key序列化不匹配、Value Schema和类结构不匹配、producer提前释放未完成发送导致

完整实现方案 #

前置依赖 #

需安装以下NuGet包：

• Confluent.Kafka
• Confluent.SchemaRegistry
• Confluent.SchemaRegistry.Serdes.Avro
  如果使用AvroConvert简化方案，额外安装AvroConvert包即可

方案1：基于Confluent官方Schema Registry实现 #

步骤1：对齐Currency类和Avro Schema

首先你可以通过Schema Registry接口GET /subjects/currency-value/versions/latest获取ksqlDB自动生成的Value Schema，保证C#类的字段名、类型和Schema完全匹配，若Schema字段为小写，可通过别名属性适配：

using Avro;

public class Currency
{
    [AvroAlias("id")]
    public int Id { get; set; }
    [AvroAlias("name")]
    public string Name { get; set; }
}

步骤2：写入逻辑实现

// 基础配置
var producerConfig = new ProducerConfig
{
    BootstrapServers = kafkaConfig.ServerUrl,
    Acks = Acks.All // 配置所有副本确认后返回，避免消息丢失
};
var schemaRegistryConfig = new SchemaRegistryConfig
{
    Url = kafkaConfig.SchemaRegistryUrl
};

using var schemaRegistry = new CachedSchemaRegistryClient(schemaRegistryConfig);
using var producer = new ProducerBuilder<int, Currency>(producerConfig)
    .SetKeySerializer(Serializers.Int32) // Key使用原生Int32序列化，不要用Avro序列化
    .SetValueSerializer(new AvroSerializer<Currency>(schemaRegistry).AsSyncOverAsync())
    .Build();

var msg = new Currency { Id = 1, Name = "3" };
var message = new Message<int, Currency>
{
    Key = msg.Id,
    Timestamp = new Timestamp(DateTime.UtcNow, TimestampType.CreateTime),
    Value = msg
};
// 等待发送结果，不要提前释放producer
var deliveryResult = await producer.ProduceAsync("currency", message);
// 可打印结果确认发送状态
Console.WriteLine($"消息发送成功，偏移量：{deliveryResult.Offset}，分区：{deliveryResult.Partition}");

步骤3：读取逻辑实现

var consumerConfig = new ConsumerConfig
{
    GroupId = kafkaConfig.ConsumerGroup,
    BootstrapServers = kafkaConfig.ServerUrl,
    AutoOffsetReset = AutoOffsetReset.Earliest
};
var schemaRegistryConfig = new SchemaRegistryConfig
{
    Url = kafkaConfig.SchemaRegistryUrl
};

using var schemaRegistry = new CachedSchemaRegistryClient(schemaRegistryConfig);
using var consumer = new ConsumerBuilder<int, Currency>(consumerConfig)
    .SetKeyDeserializer(Deserializers.Int32) // Key使用原生Int32反序列化，不要用Avro反序列化
    .SetValueDeserializer(new AvroDeserializer<Currency>(schemaRegistry).AsSyncOverAsync())
    .Build();

consumer.Subscribe("currency");
var result = consumer.Consume();
var currency = result.Message.Value;
Console.WriteLine($"读取到数据：Id={currency.Id}, Name={currency.Name}");

方案2：基于AvroConvert实现（无需手动管理Schema） #

AvroConvert的Schema默认从传入的C#类自动反射生成，也可配置从Schema Registry拉取，无需手动处理Schema注册逻辑，代码更简洁：

写入逻辑

var producerConfig = new ProducerConfig
{
    BootstrapServers = kafkaConfig.ServerUrl,
    Acks = Acks.All
};
using var producer = new ProducerBuilder<int, byte[]>(producerConfig)
    .SetKeySerializer(Serializers.Int32)
    .Build();

var msg = new Currency { Id = 1, Name = "3" };
// 直接序列化对象为Avro字节
var avroBytes = AvroConvert.Serialize(msg);
var message = new Message<int, byte[]>
{
    Key = msg.Id,
    Value = avroBytes
};
await producer.ProduceAsync("currency", message);

读取逻辑

var consumerConfig = new ConsumerConfig
{
    GroupId = kafkaConfig.ConsumerGroup,
    BootstrapServers = kafkaConfig.ServerUrl,
    AutoOffsetReset = AutoOffsetReset.Earliest
};
using var consumer = new ConsumerBuilder<int, byte[]>(consumerConfig)
    .SetKeyDeserializer(Deserializers.Int32)
    .Build();

consumer.Subscribe("currency");
var result = consumer.Consume();
var currency = AvroConvert.Deserialize<Currency>(result.Message.Value);

消息投递失败排查点 #

• 发送时必须等待ProduceAsync返回结果，不要在发送完成前释放producer；若使用无回调的Produce方法，必须调用producer.Flush()等待所有消息发送完成
• 可通过ksqlDB CLI执行PRINT 'currency' FROM BEGINNING;直接查看主题内的原始消息，确认消息是否真的未投递
• 检查Schema Registry中currency-value的Schema是否和你的Currency类结构完全一致，类型不匹配、字段名不匹配都会导致消息被ksqlDB忽略
• 确认你使用的Kafka、Schema Registry、ksqlDB版本和Confluent.Kafka NuGet包版本兼容，建议使用1.8.x及以上稳定版本

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