Kafka 简单生产者不发送消息也不报错

Question

我正在编写一个简单的生产者，我只想将原始数据发送到一个主题中。出于某种原因，我需要指定序列化器，它将消息转换为 json，然后转换为 utf-8，然后发送 json 消息..

此代码不起作用（没有错误，但在主题中没有任何内容）

from kafka import KafkaProducer

producer = KafkaProducer(bootstrap_servers=['localhost:9092'])
topic = "my_new_topic5"

producer.send(topic, b'test message')

此代码有效

producer = KafkaProducer(bootstrap_servers=['localhost:9092'],
                     value_serializer=lambda x:
                     dumps(x).encode('utf-8'))

for e in range(2):
    data = {'number': e}
    producer.send('numtest', value=data)
    sleep(5)

Answer 1

在终止程序之前尝试在send() 和producer.close() 之后调用producer.flush()。以下应该可以解决问题：

from kafka import KafkaProducer

producer = KafkaProducer(bootstrap_servers=['localhost:9092'])
topic = "my_new_topic5"

producer.send(topic, b'test message')
producer.flush()
producer.close()