使用python将列表发送给Kafka生产者

Question

我有一些我希望通过Kafka制作人发送的列表。

listA [1,2,3]
listB ["cat", "dog", "fish"]

生产者将消息作为字节发送，因此我不确定如何正确设置消息，以便列表发送，当需要引号发送消息时。这就是我现在拥有的。

producer = KafkaProducer(bootstrap_servers='localhost:9092')
for _ in range(1):
        print(producer.send('test', b'"worker_id": listA ,  "worker_name" : listB'))

这个方法只给我一个SyntaxError。我也尝试过这种方法，我得到了类似的结果

print(producer.send('test', b("worker_uuid": worker_uuid))

Answer 1

这是通过首先将列表转换为字节数组来解决的。

rList = [1, 2, 3, 4, 5]

arr = bytes(rList)
print(arr)

结果是

b'\x01\x02\x03\x04\x05'

Answer 2

你考虑过JSON编码吗？如果你配置你的KafkaProducer w / value_serializer如下：

KafkaProducer(value_serializer=lambda v: json.dumps(v).encode('utf-8'))

那么你应该能够直接发送你的列表：

producer.send('test', [1, 2, 3])
producer.send('test', ["cat", "dog", "fish"])

消费者也需要配置为通过json进行解码。如果您使用kafka-python，您可能会执行以下操作：

KafkaConsumer(value_deserializer=lambda v: json.loads(v.decode('utf-8')))