如何在Python的prepare_producer()中配置多个Schema Registry URL

方法1：使用反向代理/负载均衡（推荐生产环境） #

在Schema Registry集群前端部署反向代理（如Nginx、HAProxy），将多个Registry节点地址统一为一个代理地址。代码中直接使用这个代理地址即可，无需修改现有逻辑：

from kafka_schema_registry import prepare_producer

SAMPLE_SCHEMA = {
    "type": "record",
    "name": "TestType",
    "fields" : [
        {"name": "age", "type": "int"},
        {"name": "name", "type": ["null", "string"]}
    ]
}

# 替换为反向代理地址
producer = prepare_producer(
        ['localhost:9092'],
        f'http://schema-registry-proxy:8081',  # 代理地址
        topic_name,
        1,
        1,
        value_schema=SAMPLE_SCHEMA,
)

producer.send(topic_name, {'age': 34})

这种方式的优势是客户端无需改动，代理会自动处理节点故障转移和负载分配，符合生产环境高可用要求。

方法2：手动初始化SchemaRegistryClient（绕过prepare_producer封装） #

如果kafka_schema_registry库底层依赖的Schema Registry客户端支持多URL（比如基于confluent-kafka的实现通常支持逗号分隔的URL列表），可以跳过prepare_producer封装函数，手动创建客户端和生产者：

from kafka import KafkaProducer
from kafka_schema_registry.client import SchemaRegistryClient
from kafka_schema_registry.serializer import KafkaSerializer

SAMPLE_SCHEMA = {
    "type": "record",
    "name": "TestType",
    "fields" : [
        {"name": "age", "type": "int"},
        {"name": "name", "type": ["null", "string"]}
    ]
}

# 初始化支持多URL的SchemaRegistryClient
client = SchemaRegistryClient(urls=['http://schemaregistry1:8081', 'http://schemaregistry2:8081'])
# 序列化器
serializer = KafkaSerializer(client, SAMPLE_SCHEMA)

# 创建Kafka生产者并指定序列化器
producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=serializer
)

producer.send(topic_name, {'age': 34})
producer.flush()

注意：不同版本的kafka_schema_registry库API可能略有差异，需根据实际版本调整代码。这种方式直接操作底层客户端，支持配置多个Registry URL实现故障转移。

注意事项 #

• 确认kafka_schema_registry库的底层客户端是否支持多URL配置，部分轻量封装库可能未暴露该参数，此时反向代理是更稳妥的方案。
• 多URL配置下，客户端通常会自动尝试故障转移，但需确保Registry集群配置了跨节点的Schema同步。

内容的提问来源于stack exchange，提问作者pradip.parkar