无法在 Kafka Python 中向主题发送消息

Question

我有一个生产者代码，用于向 Kafka 发送消息。直到昨天我都能发送消息。从今天起，我无法发送消息。不确定其版本是否兼容问题。没有任何失败或错误消息，代码被执行，但它不发送消息。

以下是python模块版本，
kafka-python==2.0.1
Python 3.8.2

下面是我的代码，

from kafka import KafkaProducer
import logging
logging.basicConfig(level=logging.INFO)

producer = KafkaProducer(bootstrap_servers='127.0.0.1:9092')
producer.send('Jim_Topic', b'Message from PyCharm')
producer.send('Jim_Topic', key=b'message-two', value=b'This is Kafka-Python')

任何建议都会有所帮助。

我也尝试记录该行为，但不知道为什么生产者被关闭，

INFO:kafka.conn:<BrokerConnection node_id=bootstrap-0 host=127.0.0.1:9092 <connecting> [IPv4 ('127.0.0.1', 9092)]>: connecting to 127.0.0.1:9092 [('127.0.0.1', 9092) IPv4]
INFO:kafka.conn:<BrokerConnection node_id=bootstrap-0 host=127.0.0.1:9092 <connecting> [IPv4 ('127.0.0.1', 9092)]>: Connection complete.
INFO:kafka.producer.kafka:Closing the Kafka producer with 0 secs timeout.
INFO:kafka.producer.kafka:Proceeding to force close the producer since pending requests could not be completed within timeout 0.
INFO:kafka.producer.kafka:Kafka producer closed

Process finished with exit code 0

Answer 1

最后添加producer.flush() 帮助我解决了这些问题。在实际提交事务之前，任何未完成的消息都将被刷新（传递）

from kafka import KafkaProducer
import logging
logging.basicConfig(level=logging.INFO)

producer = KafkaProducer(bootstrap_servers='127.0.0.1:9092')
producer.send('Jim_Topic', b'Message from PyCharm')
producer.send('Jim_Topic', key=b'message-two', value=b'This is Kafka-Python')
producer.flush()

Answer 2

你能试试下面的代码吗？我从 kafka-python 文档中选择了它，并在我本地的 Kafka 实例上进行了尝试

from kafka import KafkaProducer
import json
def on_send_success(record_metadata):
    print(record_metadata.topic)
    print(record_metadata.partition)
    print(record_metadata.offset)


def on_send_error(excp):
    log.error('I am an errback', exc_info=excp)


if __name__ == "__main__":
    target_topic = "Jim_Topic"
    producer = KafkaProducer(bootstrap_servers=['127.0.0.1:9092'],retries=5, key_serializer=lambda x: json.dumps(x).encode("ascii"), value_serializer=lambda x: json.dumps(x).encode("ascii"))
    messages = [{key:"message_one",value:"Message from PyCharm"},{key:"message_two",value:"This is Kafka-Python"}]
    for msg in messages:
      producer.send(target_topic,msg).add_callback(on_send_success).add_errback(on_send_error)

    producer.flush(timeout=10) # this forcibly sends any messages that are stuck.
    producer.close(timeout=5)