KeyedMessage<String, byte[]> keyedMessage = new KeyedMessage<String, byte[]>(request.getRequestTopicName(), SerializationUtils.serialize(message));
producer.send(keyedMessage);
目前，我正在发送没有任何密钥的消息作为密钥消息的一部分，它仍然可以与delete.retention.ms 一起使用吗？我需要在消息中发送密钥吗？将密钥作为消息的一部分这样好吗？
【问题讨论】：
标签： apache-kafka kafka-consumer-api kafka-producer-api
如果您需要强大的密钥顺序并且正在开发诸如状态机之类的东西，则密钥通常是有用/必要的。如果您要求始终以正确的顺序查看具有相同键（例如，唯一 id）的消息，则将键附加到消息将确保具有相同键的消息始终进入主题中的同一分区。 Kafka 保证分区内的顺序，但不保证主题中的分区之间的顺序，因此，或者不提供密钥（这将导致跨分区的循环分配）将不会保持这种顺序。
在状态机的情况下，密钥可以与 log.cleaner.enable 一起使用，以对具有相同密钥的条目进行重复数据删除。在这种情况下，Kafka 假设您的应用程序只关心给定键的最新实例，并且日志清理器仅在该键不为空时才删除给定键的旧副本。这种形式的日志压缩由 log.cleaner.delete.retention 属性控制，并且需要密钥。
或者，默认启用的更常见的属性 log.retention.hours 通过删除已过期的完整日志段来工作。在这种情况下，不必提供密钥。 Kafka 将简单地删除超过给定保留期的日志块。
就是这么说，如果您启用了log compaction 或要求对具有相同密钥的消息进行严格排序，那么您绝对应该使用密钥。否则，在某些键可能比其他键出现得更多的情况下，空键可能会提供更好的分布并防止潜在的热点问题。
【讨论】：
ProducerRecord时可以显式指定分区@
tl;博士 不，在向 Kafka 发送消息时不需要密钥。但是……
除了非常有用的公认答案之外，我还想补充一些细节
默认情况下，Kafka 使用消息的键来选择它写入的主题的分区。这是在DefaultPartitioner by
kafka.common.utils.Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
如果没有提供密钥，则 Kafka 将以循环方式对数据进行分区。
在 Kafka 中，可以通过扩展 Partitioner 类来创建自己的 Partitioner。为此，您需要覆盖具有签名的partition 方法：
int partition(String topic,
Object key,
byte[] keyBytes,
Object value,
byte[] valueBytes,
Cluster cluster)
通常，Kafka消息的key用于选择分区，返回值（int类型）为分区号。如果没有密钥，您需要依赖可能更复杂处理的值。
如给定答案中所述，Kafka 仅在分区级别保证消息的排序。
假设您想将客户的金融交易存储在具有两个分区的 Kafka 主题中。消息可能看起来像（键：值）
null:{"customerId": 1, "changeInBankAccount": +200}
null:{"customerId": 2, "changeInBankAccount": +100}
null:{"customerId": 1, "changeInBankAccount": +200}
null:{"customerId": 1, "changeInBankAccount": -1337}
null:{"customerId": 1, "changeInBankAccount": +200}
由于我们没有定义键，所以两个分区可能看起来像
// partition 0
null:{"customerId": 1, "changeInBankAccount": +200}
null:{"customerId": 1, "changeInBankAccount": +200}
null:{"customerId": 1, "changeInBankAccount": +200}
// partition 1
null:{"customerId": 2, "changeInBankAccount": +100}
null:{"customerId": 1, "changeInBankAccount": -1337}
您阅读该主题的消费者最终可能会告诉您该帐户的余额在特定时间为 600，尽管情况并非如此！只是因为它在分区 1 中的消息之前读取了分区 0 中的所有消息。
使用有意义的键（如 customerId）可以避免这种情况，因为分区是这样的：
// partition 0
1:{"customerId": 1, "changeInBankAccount": +200}
1:{"customerId": 1, "changeInBankAccount": +200}
1:{"customerId": 1, "changeInBankAccount": -1337}
1:{"customerId": 1, "changeInBankAccount": +200}
// partition 1
2:{"customerId": 2, "changeInBankAccount": +100}
请记住，只有将生产者配置 max.in.flight.requests.per.connection 设置为 1 才能保证分区内的排序。但是，该配置的默认值是 5，它被描述为：
"客户端在阻塞前将在单个连接上发送的未确认请求的最大数量。注意，如果此设置设置为大于 1 并且有发送失败的情况，则存在重新发送消息的风险由于重试而排序（即，如果启用重试）。"
您可以在 Kafka - Message Ordering Guarantees 上的另一篇 Stackoverflow 帖子中找到更多详细信息。
如果没有密钥作为消息的一部分，您将无法将主题配置 cleanup.policy 设置为 compacted。根据documentation，“日志压缩确保 Kafka 将始终为单个主题分区的数据日志中的每个消息键至少保留最后一个已知值。”。
如果没有任何密钥，这个漂亮而有用的设置将无法使用。
在实际用例中，Kafka 消息的密钥会对您的性能和业务逻辑的清晰度产生巨大影响。
例如，密钥可以自然地用于对数据进行分区。由于您可以控制您的消费者从特定分区中读取，这可以作为一个有效的过滤器。此外，密钥可以包含一些关于消息实际值的元数据，以帮助您控制后续处理。键通常比值小，因此解析键而不是整个值更方便。同时，您也可以使用密钥应用所有序列化和模式注册。
作为说明，还有Header的概念，可以用来存储信息，见documentation。
【讨论】：
hash(key) % 10 分布在 10 个分区中。
带有消息的键基本上是为了获取特定字段的消息排序而发送的。
解释和举例
【讨论】：