Python Kafka消费者无法获取消息Key问题求助

嘿，我一眼就瞅出问题所在啦！你当前的生产者代码发送消息时，完全没有指定key参数，只传了消息的value部分，所以消费者那边拿到的key自然就是默认的None咯。

问题根源 #

Kafka的producer.send()方法中，key是可选参数——如果你不主动指定，它就不会被设置。你看你原来的代码：

producer.send('collect-response-devices', {'deviceId': deviceId})

这里的第二个参数其实是被当作value传入的，根本没传key，所以消息里自然没有key信息。

解决方案 #

我们需要两步调整：一是给生产者配置key的序列化器，二是发送消息时显式指定key参数，同时消费者也要对应处理key的反序列化。

1. 修正生产者代码

首先，创建KafkaProducer时同时配置key_serializer（因为Kafka要求key和value都要是字节类型，所以需要把字符串key转成字节），并且发送消息时显式传入key参数：

from kafka import KafkaProducer
from kafka.errors import KafkaError
import json

# 合并所有配置到一个Producer实例，避免重复创建覆盖配置
producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    key_serializer=lambda k: k.encode('ascii'),  # 字符串key转字节
    value_serializer=lambda m: json.dumps(m).encode('ascii'),
    retries=5  # 把重试配置也加进来
)

deviceId = "4bc03533ccc94065"
responseId = "c03c4851-701f-4265-aafd-eb133c09c08e"

print(deviceId)
print(responseId)

# 发送消息时显式指定key和value
producer.send(
    'collect-response-devices',
    key=deviceId,
    value={'deviceId': deviceId}
)

producer.send(
    'collect-response-responses',
    key=responseId,
    value={'responseId': responseId}
)

# 回调函数保留
def on_send_success(record_metadata):
    print(record_metadata.topic)
    print(record_metadata.partition)
    print(record_metadata.offset)

def on_send_error(excp):
    # 建议用logging模块代替print，这里保留你的逻辑
    print(excp)

# 确保所有异步消息都发送完成
producer.flush()

2. 修正消费者代码

消费者需要添加key_deserializer来解码生产者发送的key（对应生产者的编码方式）：

from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    bootstrap_servers='localhost:9092',
    auto_offset_reset='earliest',
    key_deserializer=lambda k: k.decode('ascii') if k else None,  # 解码key，兼容key为None的情况
    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)

consumer.subscribe(['collect-response-devices'])

for message in consumer:
    print(message.key, message.value)

效果验证 #

修改完成后，消费者的输出就会变成类似这样：

('4bc03533ccc94065', {'deviceId': '4bc03533ccc94065'})

内容的提问来源于stack exchange，提问作者hasherBaba