如何在.NET Confluent Kafka中定义自定义主题名称策略

核心思路 #

要生成env.RecordTypeFullName格式的主题名称，只需实现Confluent Kafka .NET提供的ITopicNameStrategy接口，若需同步自定义Schema命名，可结合ISchemaNameStrategy接口一起实现。

具体实现步骤 #

1. 实现自定义主题名称策略类 #

创建类实现ITopicNameStrategy，重写Serialize方法拼接环境标识与记录完整类型名：

using Confluent.Kafka;
using Confluent.SchemaRegistry.Serdes;

public class EnvRecordTypeTopicNameStrategy : ITopicNameStrategy
{
    private readonly string _env;

    public EnvRecordTypeTopicNameStrategy(string env)
    {
        _env = env ?? throw new ArgumentNullException(nameof(env));
    }

    public string Serialize<T>(string topic, T data, SerializationContext context)
    {
        var recordFullName = typeof(T).FullName ?? typeof(T).Name;
        return $"{_env}.{recordFullName}";
    }
}

2. （可选）自定义Schema名称策略 #

如果需要Schema也遵循关联格式，实现ISchemaNameStrategy接口：

public class EnvRecordTypeSchemaNameStrategy : ISchemaNameStrategy
{
    private readonly string _env;

    public EnvRecordTypeSchemaNameStrategy(string env)
    {
        _env = env ?? throw new ArgumentNullException(nameof(env));
    }

    public string GetSchemaName<T>(bool isKey)
    {
        var recordFullName = typeof(T).FullName ?? typeof(T).Name;
        return $"{_env}.{recordFullName}.{(isKey ? "key" : "value")}";
    }
}

3. 配置使用自定义策略 #

在生产者/消费者初始化时指定自定义策略，以Avro生产者为例：

var producerConfig = new ProducerConfig
{
    BootstrapServers = "localhost:9092"
};

var avroSerializerConfig = new AvroSerializerConfig();
var schemaRegistryConfig = new SchemaRegistryConfig
{
    Url = "http://localhost:8081"
};

using var producer = new ProducerBuilder<string, MyRecord>(producerConfig)
    .SetValueSerializer(new AvroSerializer<MyRecord>(
        schemaRegistryConfig,
        avroSerializerConfig,
        topicNameStrategy: new EnvRecordTypeTopicNameStrategy("dev"),
        schemaNameStrategy: new EnvRecordTypeSchemaNameStrategy("dev")
    ))
    .Build();

4. 注意事项 #

• 环境标识env建议从配置文件（如appsettings.json）读取，避免硬编码，适配多环境切换。
• 其他序列化器（Protobuf、JsonSchema等）的自定义策略配置逻辑一致，只需替换对应序列化器的构造参数。
• 主题名称需符合Kafka规范：仅支持字母、数字、下划线、连字符、点号，长度不超过255字符。

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