Spring Cloud Streams 没有在消息中设置 kafka 键？

Question

故事是这样的。我有一个 kafka 代理和一个特定对象（我将其 jsonify 以通过我的主题发送），它有一个我想用作键的 ID。

目前我正在使用“partitionKeyExtractorClass”配置来设置提取ID并将其作为键返回的类。

看起来像这样：

def extractKey(Message<?> message) {
    log.info('Extracting key from message')
    String id = new JsonSlurper().parseText(new String(message.payload)).properties.id
    log.info("Got = ${id}")

    return id
}

我的实际问题是，当我浏览有关该主题的消息时，保存我的消息的 ConsumerRecord 说密钥为空...

这是一个错误吗？难道我做错了什么？这方面的文档并没有比这更进一步。

Answer 1

你看，你把partition和key混在一起了。

目前KafkaMessageChannelBinder 不提供用于确定key 与Message 的选项。

只有您可以强大使用的现有功能是KafkaHeaders.MESSAGE_KEY：

    Object messageKey = this.messageKeyExpression != null
            ? this.messageKeyExpression.getValue(this.evaluationContext, message)
            : message.getHeaders().get(KafkaHeaders.MESSAGE_KEY);

因此，在output 消息之前，您应该计算密钥并将其放入该标头中。