我已经在Kafka生产者中传递了一个Protobuf对象,并且在消费者端接收到一个字节数组。现在我想再次将该响应反序列化回Protobuf对象,但是我无法做到这一点。我该如何做到呢?
这是我的消费者:
from email import message
from kafka import KafkaConsumer
import json
from simple import simple_message
import check_pb2
consumer = KafkaConsumer('latest', api_version=(0, 10, 1),
group_id='my-group', enable_auto_commit=False,
bootstrap_servers=['localhost:9092'])#,
#value_deserializer=lambda m: json.loads(m.decode('ascii')))
for message in consumer:
print(message)
我尝试使用Confluent,但Kafka Exception : {} KafkaError{code=_VALUE_DESERIALIZATION,val=-159,str="Unknown magic byte. This message was not produced with a Confluent Schema Registry serializer"
1条答案
按热度按时间nhaq1z211#
如果您没有使用Confluent Schema Registry,请使用
ParseFromString
。如果(且仅当)您正在使用Schema Registry,则Confluent有一个Protobuf反序列化器类。请参阅example,它恰好也在内部使用
ParseFromString
。