MQTT Kafka源连接器：有趣的字节字符

Question

mqttpublisher.py

while v3 < 3:
             data3 = {
                      "time": str(datetime.datetime.now().time()),
                       "val": v3
                      }
             client.publish("sensor/dist", json.dumps(data3), qos=2)

             v3 += 1
             time.sleep(2)

mqttsubscriber.py

def on_message_print(client, userdata, message):
            print(message.topic,message.payload)

subscribe.callback(on_message_print, "sensor/#", hostname="localhost")

py

consumer = KafkaConsumer('mqtt.',
                     bootstrap_servers=['localhost:9092'])

for message in consumer:
   print(message)

ConsumerRecord(Topic='mqtt.',partition=0,offset=225,timestamp=1545117270870,timestamp_type=0,key=b'\x00\x00\x00\x01\x16sensor/dist“,value=b'\x00\x00\x00\x00\x01\x16sensor/dist{”time“：”12:44:0.817462“,”val“：0}”,headers=[('mqtt.message.id“,b'0'),('mqtt.qos‘,b'0')b'false')],checksum=none,serialized_key_size=17,serialized_value_size=43,serialized_header_size=62)

ConsumerRecord(Topic='mqtt.',partition=0,offset=226,timestamp=1545117272821,timestamp_type=0,key=b'\x00\x00\x00\x01\x16sensor/dist“,value=b'\x00\x00\x00\x01\x16sensor/dist{”time“：”12:44:32.820040“,”val“：1}”,headers=[('mqtt.message.id',b'0'),('mqtt.qos',b'0'),'false')],checksum=none,serialized_key_size=17,serialized_value_size=43,serialized_header_size=62)

ConsumerRecord(Topic='mqtt.',partition=0,offset=227,timestamp=1545117274824,timestamp_type=0,key=B'\x00\x00\x00\x01\x16sensor/dist“,value=B'\x00\x00\x00\x00\x01\x16sensor/dist{”time“：”12:44:34.822657“,”val“：2}”,headers=[('mqtt.message.id',B'0'),('mqtt.qos',B'0'),,b'false')],checksum=none,serialized_key_size=17,serialized_value_size=43,serialized_header_size=62)

是什么原因导致了Kafka消费者中额外字节的上述预置？提前谢了。

Answer 1

作为演示的一部分，您将启动一个模式注册表

启动Kafka连接和依赖项(Kafka、Zookeeper、模式注册表）：

合流开始连接