Kafka+Avro+Python实现生产者消费者收发Avro格式消息问题求助

现有代码存在的问题 #

• 语法错误：Schema字典定义未闭合，末尾缺少}]}符号，无法正常解析为JSON格式。
• value_serializer参数使用错误：KafkaProducer的value_serializer需要传入可调用的序列化函数，负责将Python对象转换为二进制字节，不能直接传入解析后的Avro Schema对象。
• 缺少完整Avro序列化逻辑：仅传入Schema无法完成消息到Avro二进制格式的转换，需要搭配Avro的DatumWriter、BinaryEncoder等组件实现序列化逻辑。
• 无发送确认逻辑：循环发送消息后没有调用flush()等待发送完成，可能程序退出时消息还未实际发送到Kafka broker。

正确实现方法 #

方式1：原生Avro实现（无需Schema Registry，适合测试/简单场景） #

先安装依赖：

pip install kafka-python avro

生产者代码

import json
import io
import avro.schema
from avro.io import DatumWriter, BinaryEncoder
from kafka import KafkaProducer

# 修正后的完整Schema定义
schema = {
    "type": "record",
    "name": "myrecord",
    "fields": [
        {"name": "typ", "type": "string"},
        {"name": "pred", "type": "int"}
    ]
}
parsed_schema = avro.schema.parse(json.dumps(schema))

# 自定义Avro序列化函数
def avro_serializer(msg):
    writer = DatumWriter(parsed_schema)
    bytes_writer = io.BytesIO()
    encoder = BinaryEncoder(bytes_writer)
    writer.write(msg, encoder)
    return bytes_writer.getvalue()

# 初始化生产者
producer = KafkaProducer(
    bootstrap_servers=['xxxx:xxxx'],
    value_serializer=avro_serializer
)

# 发送消息
for i in range(100):
    message = {"typ": "sth", "pred": i}
    producer.send("your_topic_name", value=message)
# 等待所有消息发送完成
producer.flush()
producer.close()

消费者代码

import io
import json
import avro.schema
from avro.io import DatumReader, BinaryDecoder
from kafka import KafkaConsumer

# 和生产者完全一致的Schema
schema = {
    "type": "record",
    "name": "myrecord",
    "fields": [
        {"name": "typ", "type": "string"},
        {"name": "pred", "type": "int"}
    ]
}
parsed_schema = avro.schema.parse(json.dumps(schema))

# 自定义Avro反序列化函数
def avro_deserializer(bytes_data):
    reader = DatumReader(parsed_schema)
    bytes_reader = io.BytesIO(bytes_data)
    decoder = BinaryDecoder(bytes_reader)
    return reader.read(decoder)

# 初始化消费者
consumer = KafkaConsumer(
    'your_topic_name',
    bootstrap_servers=['xxxx:xxxx'],
    auto_offset_reset='earliest',
    value_deserializer=avro_deserializer
)

# 消费消息
for msg in consumer:
    print(f"接收消息：{msg.value}")

方式2：Confluent Avro实现（带Schema Registry，生产环境推荐） #

生产环境通常搭配Schema Registry管理Schema，避免生产者消费者手动维护Schema一致性，先安装依赖：

pip install confluent-kafka confluent-kafka[avro]

生产者代码

from confluent_kafka import Producer
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer
from confluent_kafka.serialization import SerializationContext, MessageField

# Schema Registry配置
schema_registry_client = SchemaRegistryClient({'url': 'http://你的SchemaRegistry地址:8081'})

schema_str = """
{
    "type": "record",
    "name": "myrecord",
    "fields": [
        {"name": "typ", "type": "string"},
        {"name": "pred", "type": "int"}
    ]
}
"""

avro_serializer = AvroSerializer(schema_registry_client, schema_str)

# 初始化生产者
producer = Producer({'bootstrap.servers': 'xxxx:xxxx'})

# 发送消息
for i in range(100):
    message = {"typ": "sth", "pred": i}
    producer.produce(
        topic="your_topic_name",
        value=avro_serializer(message, SerializationContext("your_topic_name", MessageField.VALUE))
    )
producer.flush()

消费者代码

from confluent_kafka import Consumer
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroDeserializer
from confluent_kafka.serialization import SerializationContext, MessageField

# Schema Registry配置
schema_registry_client = SchemaRegistryClient({'url': 'http://你的SchemaRegistry地址:8081'})

schema_str = """
{
    "type": "record",
    "name": "myrecord",
    "fields": [
        {"name": "typ", "type": "string"},
        {"name": "pred", "type": "int"}
    ]
}
"""

avro_deserializer = AvroDeserializer(schema_registry_client, schema_str)

# 初始化消费者
consumer = Consumer({
    'bootstrap.servers': 'xxxx:xxxx',
    'group.id': '你的消费组ID',
    'auto.offset.reset': 'earliest'
})
consumer.subscribe(['your_topic_name'])

# 消费消息
while True:
    msg = consumer.poll(1.0)
    if msg is None:
        continue
    if msg.error():
        print(f"消费错误: {msg.error()}")
        continue
    value = avro_deserializer(msg.value(), SerializationContext(msg.topic(), MessageField.VALUE))
    print(f"接收消息：{value}")

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