### 使用 Protocol Buffers 和 Kafka 的 Python 实现 为了在 Python 中实现通过 Protocol Buffers 编解码消息并使用 Kafka 进行传输，可以按照以下方法设置环境和编写代码。 #### 安装依赖库 首先安装 `confluent-kafka` 库用于连接到 Kafka 集群以及处理消息收发操作；另外还需要安装 `protobuf` 来支持序列化/反序列化的功能： ```bash pip install confluent_kafka protobuf ``` #### 创建 .proto 文件定义结构体 创建一个 `.proto` 文件来描述要发送的数据格式。例如名为 `animal.proto`: ```protobuf syntax = "proto3"; package animal; message Unicorn { string name = 1; int32 age = 2; } ``` 接着利用命令行工具编译此文件生成对应的 Python 类： ```bash protoc --python_out=. ./animal.proto ``` 这将会产生一个新的 Python 模块叫做 `animal_pb2.py`,其中包含了基于上述 proto 文件自动生成的消息类。 #### 生产者端编码逻辑 生产者负责构建对象实例并通过 Kafka 发送出去，在这里展示了一个简单的例子说明怎样做： ```python from confluent_kafka import Producer, SerializingProducer import json from google.protobuf.json_format import MessageToJson from animal_pb2 import Unicorn as AnimalUnicorn def delivery_report(err, msg): """ Called once for each message produced to indicate delivery result. Triggered by poll() or flush(). """ if err is not None: print('Message delivery failed: {}'.format(err)) else: print('Message delivered to {} [{}]'.format(msg.topic(), msg.partition())) conf = {'bootstrap.servers': 'localhost:9092', 'key.serializer': lambda k: str(k).encode(), 'value.serializer': lambda v: MessageToJson(v)} producer = SerializingProducer(conf) topic = "unicorns_protobuffed" for i in range(5): # Send five messages unicorn = AnimalUnicorn(name=f'Uni{i}', age=i * 10) producer.produce(topic=topic, key="my_key", value=unicorn, on_delivery=delivery_report) # Wait for any outstanding messages to be delivered and delivery report callbacks to occur producer.flush() ``` 这段脚本会向指定的主题发布五个不同名字年龄的独角兽实体，并且每条记录都会被转换成 JSON 字符串形式再传送给 broker[^1]。 #### 消费者端解码逻辑 消费者接收到来自主题中的数据包之后将其解析回原始的对象表示方式以便进一步加工处理： ```python from confluent_kafka import Consumer, DeserializingConsumer from google.protobuf.json_format import ParseDict from animal_pb2 import Unicorn as AnimalUnicorn import json class ProtobufDeserializer(object): def __init__(self, cls): self.cls = cls def __call__(self, s, ctx=None): obj_dict = json.loads(s.decode()) return ParseDict(obj_dict, self.cls()) consumer_conf = {'bootstrap.servers': 'localhost:9092', 'group.id': 'test_group', 'auto.offset.reset': 'earliest'} deserializer_config = consumer_conf.copy() deserializer_config.update({ 'key.deserializer': lambda k, _: k.decode(), 'value.deserializer': ProtobufDeserializer(AnimalUnicorn), }) consumer = DeserializingConsumer(deserializer_config) consumer.subscribe(['unicorns_protobuffed']) try: while True: msg = consumer.poll(timeout=1.0) if msg is None: continue error = msg.error() if error: raise Exception(f"Error occurred during consumption: {error}") print("Received message:") received_unicorn = msg.value() print(received_unicorn) except KeyboardInterrupt: pass finally: consumer.close() ``` 以上程序片段展示了如何订阅特定主题监听新到达的信息流，每当有新的项目加入队列时就会触发回调函数执行相应的业务流程。