发送消息到Kafka时，是否需要包含密钥？

Question

发送消息到Kafka时，是否需要包含密钥？

apache-kafkakafka-consumer-apikafka-producer-api

184

KeyedMessage<String, byte[]> keyedMessage = new KeyedMessage<String, byte[]>(request.getRequestTopicName(), SerializationUtils.serialize(message)); 
producer.send(keyedMessage);

目前，我正在作为键控消息的一部分发送没有任何键的消息，如果使用delete.retention.ms，它仍然能正常工作吗？我需要在消息中发送一个键吗？将键作为消息的一部分是否好？

- gaurav

3个回答

98

tl;dr 发送消息到Kafka不需要键，但默认情况下会使用消息的键来选择主题的分区。

除了接受的答案外，我想添加一些细节。

分区

默认情况下，Kafka使用消息的键来选择要写入的主题的分区。这是由DefaultPartitioner完成的。

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 消息的键用于选择分区，返回值（类型为 int）是分区号。如果没有键，则需要依赖值进行处理，这可能更加复杂。

排序

正如给出的答案所述，Kafka 只对分区级别的消息排序做出保证。

假设您想在 Kafka 主题中使用两个分区存储客户的财务交易记录。消息可能如下所示 (key:value)：

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中的所有消息。

如果使用有意义的键（例如客户ID），则可以避免这种情况，分区将如下所示：

// 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。根据文档，“日志压实确保Kafka始终保留每个消息键在单个主题分区数据日志中的最后已知值。”。

没有任何键，这个好又有用的设置将不可用。

键的用法

在实际使用情况中，Kafka消息的键可以对业务逻辑的性能和清晰度产生巨大影响。

例如，可以自然地使用键来分区数据。由于可以控制消费者从特定分区读取，因此这可能作为有效的过滤器。此外，键可以包含有关消息实际值的某些元数据，以帮助您控制后续处理。键通常比值小，因此解析键而不是整个值更方便。同时，您可以像处理值一样处理键，进行所有序列化和模式注册。

需要注意的是，还存在头部的概念可用于存储信息，请参阅文档。

- Michael Heil

请问您能否解释一下如何控制消费者从特定分区读取数据？ - Guy_g23

4

带有消息键（key）的目的是为了获取指定字段的消息排序。

如果 key=null，数据将被轮流发送到不同的分区和不同的代理服务器（在分布式环境中），当然也会发送到相同的主题。
如果发送了一个 key，那么所有该 key 的消息都将始终发送到同一个分区。

解释和例子

key 可以是任何字符串或整数等。例如以一个整数 employee_id 作为 key。
因此，employee_id 123 将始终被分配到分区 0，employee_id 345 将始终被分配到分区 1。这取决于 key 哈希算法，其取决于分区数量。
如果您不发送任何 key，则使用轮询技术可以将消息发送到任何分区。

- Pradeep Singh

将消息发送到相同的分区并不能保证始终按顺序进行排序。 - Nag

1

它保证了同一分区内的顺序。 - Pradeep Singh

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- kuujo · Accepted Answer

如果您需要在键上具有强制顺序并且正在开发类似状态机的东西，则键大多是有用/必要的。如果您需要确保具有相同键（例如唯一 ID）的消息始终以正确的顺序出现，则将键附加到消息将确保具有相同键的消息始终进入主题中的同一分区。 Kafka 保证分区内有序，但主题中的分区间无序，因此不提供键 - 这将导致轮询在分区间分配 - 将无法维护此类顺序。

在状态机的情况下，可以使用键和log.cleaner.enable来去重具有相同键的条目。在这种情况下，Kafka 假设您的应用程序仅关心给定键的最新实例，而日志清理器仅在键不为空时删除给定键的旧副本。此形式的日志压缩由log.cleaner.delete.retention属性控制，并且需要键。

或者，更常见的log.retention.hours属性默认启用，通过删除过期的完整日志段来工作。在这种情况下，不必提供键。 Kafka 将简单地删除比给定保留期旧的日志块。

也就是说，如果您已启用日志压缩或需要具有相同键的消息的严格顺序，则应该肯定使用键。否则，空键可能提供更好的分布，并防止某些键可能比其他键更常出现的热点问题。