使用 MassTransit 和 Kafka 配置 Avro

Question

在生产和消费 Confluent Kafka 主题时，如何配置 MassTransit 以使用 Avro 进行序列化/反序列化？我看到 Avro 序列化器/解串器在包Confluent.SchemaRegistry.Serdes 中。欢迎提供一些代码示例。

Answer 1

要将 MassTransit 配置为使用 Avro，我的做法是使用生成的类文件 (avrogen)，然后配置生产者和主题端点，如下所示：

首先，您需要为模式注册表创建客户端：

var schemaRegistryClient = new CachedSchemaRegistryClient(new Dictionary<string, string>
{
    {"schema.registry.url", "localhost:8081"},
});

然后，您可以配置骑手：

services.AddMassTransit(x =>
{
    x.UsingInMemory((context, cfg) => cfg.ConfigureEndpoints(context));
    x.AddRider(rider =>
    {
        rider.AddConsumer<KafkaMessageConsumer>();

        rider.AddProducer<string, KafkaMessage>(Topic, context => context.MessageId.ToString())
            .SetKeySerializer(new AvroSerializer<string>(schemaRegistryClient).AsSyncOverAsync())
            .SetValueSerializer(new AvroSerializer<KafkaMessage>(schemaRegistryClient).AsSyncOverAsync());

        rider.UsingKafka((context, k) =>
        {
            k.Host("localhost:9092");

            k.TopicEndpoint<string, KafkaMessage>("topic-name", "consumer-group", c =>
            {
                c.SetKeyDeserializer(new AvroDeserializer<string>(schemaRegistryClient).AsSyncOverAsync());
                c.SetValueDeserializer(new AvroDeserializer<KafkaMessage>(schemaRegistryClient).AsSyncOverAsync());
                c.AutoOffsetReset = AutoOffsetReset.Earliest;
                c.ConfigureConsumer<KafkaMessageConsumer>(context);

                c.CreateIfMissing(m =>
                {
                    m.NumPartitions = 2;
                });
            });
        });
    });
});

您可以查看working unit test 以查看更多详细信息。我可能应该将此添加到文档中。

我刚才写这篇文章是为了回答这个问题，直到一个小时前我才使用 Avro。

另外，我使用this article from Confluent 来启动和运行。链接单元测试项目中的docker-compose.yml 配置了所有需要的服务。