试图为每次迭代生成Kafka主题的消息，但看起来我最终没有向消费者发送消息

Question

当使用循环调用kakfa生产类时，无法将消息写入kafka主题（生产者）。

我对Python和Kafka很陌生。我正在尝试编写一个python程序，将消息写入Kafka主题并生成，以便Kafka消费者可以订阅该主题以发布消息。

我不确定我的程序中缺少了什么，它限制了我将信息写入主题。

请注意：我正在读取一个JSON文件，并使用for循环来准备键值。然后将其分配给一个变量，并使用Kafka product with arg为msg引用该变量。

附件是Kafka制片人节目。

输入：Json_smpl。json

文件内容：

{
"transaction":{
"Accnttype":"Saving"
,"Branch":"West"
,"id":"WS"
}
}

程序：

from confluent_kafka import Producer
import json

def acked(err, msg):
    if err is not None:
        print("Failed to deliver message: {0}: {1}"
              .format(msg.value(), err.str()))
    else:
        print("Message produced: {0}".format(msg.value()))

p = Producer({'bootstrap.servers': 'localhost:9092'})
try:
    with open('json_smpl.json') as read_j:
        data = json.load(read_j)
        get_data = data.get("transactions")
    print(get_data)
    for i in get_data:
        a = list(get_data.items()[0])
        p.produce(topic='mytopic12', 'myvalue #{0}'.format(a), callback=acked)
except KeyboardInterrupt:
    pass
p.flush(1)

预期结果：消息（JSON键

实际结果：主题中没有消息。因此消费者没有收到任何消息。

Answer 1

您的文件没有事务键，也没有循环，因此您的JSON没有被解析，您也没有捕捉到KeyError或ValueError

从这个开始

p = Producer({'bootstrap.servers': 'localhost:9092'})
try:
    with open('json_smpl.json') as read_j:
        data = json.load(read_j).get("transaction")
        tosend = json.dumps(data)
        print("Ready to send : {}".format(tosend))
        p.produce(topic='mytopic12', tosend, callback=acked)
except:
    print("There was some error")