1.问题解决

json.dumps()参数添加ensure_ascii=False

msg = json.dumps(json_msg,ensure_ascii=False).encode()

2.问题描述

python向Kafka发送json格式的中文消息消费端显示异常，发送字符串中文显示正常
生产者代码：

import json
from kafka import KafkaProducer

producer = KafkaProducer(bootstrap_servers='node001:9092')

json_msg = {
    "user_id":"001",
    "user_name":"悟净",
    "user_age":"23",
    "remark":"11111"
}

msg = json.dumps(json_msg).encode()
msg1 = "message--中文测试--python".encode()
producer.send('test-topic1', msg)
producer.send('test-topic1', msg1)
producer.close()

消费者代码：

from kafka import KafkaConsumer

consumer = KafkaConsumer('test-topic1',
                         group_id="test-group1",
                         bootstrap_servers='node001:9092,node002:9092')
for message in consumer:
    message
    print(message.value.decode())

消费端收到消息，json格式的中文异常

json.dumps()参数添加ensure_ascii=False后中文显示正常