在.net中将多个事件类型添加到同一个kafka主题答案

【问题标题】：Adding multiple event type to same kafka topic in .net在.net中将多个事件类型添加到同一个kafka主题
【发布时间】：2021-08-02 12:42:32
【问题描述】：

我正在尝试将多个模式添加到模式注册表中的同一主题，因此我已将 ValueSubjectNameStrategy 设置为 SubjectNameStrategy.TopicRecord，还将注册表自动设置为 AutomaticRegistrationBehavior.Always。但是在自动注册架构时，它仍然使用 SubjectNameStrategy.Topic 策略。

 var schemaRegistryConfig = new SchemaRegistryConfig { Url = "http://localhost:8081", ValueSubjectNameStrategy = SubjectNameStrategy.TopicRecord };
        var registry = new CachedSchemaRegistryClient(schemaRegistryConfig);
        var builder = new ProducerBuilder<string, SplitLineKGN>(KafkaConfig.Producer.GetConfig(_config.GetSection("KafkaProducer")))
                    .SetAvroValueSerializer(registry, registerAutomatically: AutomaticRegistrationBehavior.Always)
                    .SetErrorHandler((_, error) => Console.Error.WriteLine(error.ToString()));
        _producerMsg = builder.Build();
await _producerMsg.ProduceAsync("MyTopic", new Message<string, SampleMessage> { Key = key, Value = line });

如何将多个模式自动注册到一个主题？

【问题讨论】：

标签： c# .net avro

【解决方案1】：

确保您更改了subject naming strategy for a topic
SchemaRegistryConfig.ValueSubjectNameStrategy 已弃用，现在应使用序列化程序的配置进行配置：code

要使用单个生产者生成多种事件类型，您必须使用 AvroSerializer，如下所述：

var schemaRegistryConfig = new SchemaRegistryConfig { Url = "http://localhost:8081" };

using var schemaRegistryClient = new CachedSchemaRegistryClient(schemaRegistryConfig);

var avroSerializerConfig = new AvroSerializerConfig
{
    SubjectNameStrategy = SubjectNameStrategy.TopicRecord,
    AutoRegisterSchemas = true // (the default)
};

// Assuming this is your own custom code because the Confluent 
// producer doesn't have anything like this.
var producerConfig = KafkaConfig.Producer.GetConfig(_config.GetSection("KafkaProducer"));

using var producer = new ProducerBuilder<string, ISpecificRecord>(producerConfig)
        .SetValueSerializer(new AvroSerializer<ISpecificRecord>(schemaRegistryClient, avroSerializerConfig))
        .SetErrorHandler((_, error) => Console.Error.WriteLine(error))
        .Build();

var deliveryResult = await producer.ProduceAsync("MyTopic", new Message<string, ISpecificRecord> 
{ 
    Key = key, 
    Value = line 
});

Console.WriteLine($"Delivered to: {deliveryResult.TopicPartitionOffset}");

【讨论】：