在Apache Kafka 2.0中有没有一种方法来区分消息的优先级？

Question

编辑

如果其他人在这种特殊的情况下，我得到了一些类似于我在调整消费者配置后所寻找的东西。我创建了一个生产者，它将优先级消息发送到三个单独的主题（针对高/中/低优先级），然后创建了三个单独的消费者来消费每一个主题。然后我频繁地轮询优先级较高的主题，而不轮询优先级较低的主题，除非高的主题是空的：

    while(true) {
        final KafkaConsumer<String,String> highPriConsumer = createConsumer(TOPIC1);
        final KafkaConsumer<String,String> medPriConsumer = createConsumer(TOPIC2);

        final ConsumerRecords<String, String> consumerRecordsHigh = highPriConsumer.poll(100);
        if (!consumerRecordsHigh.isEmpty()) {
            //process high pri records
        } else {
            final ConsumerRecords<String, String> consumerRecordsMed = medPriConsumer.poll(100);
            if (!consumerRecordsMed.isEmpty()) {
                //process med pri records

poll timeout（.poll()方法的参数）确定没有要轮询的记录时等待的时间。我为每个主题设置了一个非常短的时间，但您可以为较低的优先级设置更低的时间，以确保在存在高pri消息时不会消耗宝贵的等待周期

max.poll.records配置显然决定了单个轮询中要抓取的最大记录数。对于较高的优先级，这一数字也可以设置得更高。

max.poll.interval.ms配置确定轮询之间的时间-处理max.poll.records消息所需的时间。这里澄清一下。

另外，我认为暂停/恢复整个使用者/主题可以这样实现：

    kafkaConsumer.pause(kafkaConsumer.assignment())
    if(kafkaConsumer.paused().containsAll(kafkaConsumer.assignment())) {
        kafkaConsumer.resume(kafkaConsumer.assignment());
    }

我不确定这是不是最好的方法，但我在别处找不到一个好的例子

我同意下面吴老师的观点，这并不是Kafka真正的正确用法。这是单线程处理，每个主题都有一个专用的使用者，但我将从这里着手改进这个过程。

处理器API

当我尝试处理器API时，我尝试通过检查poll()是否为空来确定高优先级KafkaConsumer当前是否正在处理任何内容，然后希望使用Med优先级使用者poll()，但第二个主题poll返回为空。似乎也没有一种简单的方法来获取主题上的所有TopicPartition以便调用KafkaConsumer.pause(partitions)。

Kafka溪流

当我尝试kafkastreams时，我设置了一个流以从我的每个“优先级”主题使用，但是没有办法检查连接到较高优先级主题的kstream或kafkastreams实例当前是否空闲或正在处理。

我的代码基于这个文件

其他

我也尝试了这里的代码:priority-kafka-client，但它并没有像预期的那样工作，因为运行下载的测试文件有混合的优先级。

Answer 1

这听起来像是应用程序中的一个设计问题--kafka最初被设计为提交日志，其中每条消息都以偏移量写入代理，并且各种使用者以提交它们的顺序使用它们，具有非常低的延迟和高吞吐量。考虑到分区而不是主题是Kafka中工作分配的基本单位，具有主题级别的优先级将很难在本地实现。

我建议你调整你的设计，使用Kafka以外的其他建筑构件，而不是试图削减你的脚，以适应鞋。您可以做的一件事是让您的生产者上传文件到适当的文件存储，并通过Kafka发送链接，包括元数据。然后，根据带宽状态，您的用户可以根据大文件的元数据来决定下载是否明智。这样，您可能更有可能拥有一个健壮的设计，而不是错误地使用Kafka。

如果您真的想要坚持只使用Kafka，一个解决方案是将大文件发送到一些固定数量的硬编码分区，只有当带宽良好时，用户才会使用这些分区。