原创：Jesse Anderson

翻译：Sijie Guo

使用 Kafka 或 Pulsar 的一个常见用例是创建工作队列。这两种技术为实现此用例提供了不同的实现。我将讨论在 Kafka 和 Pulsar 中实现工作队列的方法以及它们的相对优势。

什么是工作队列

工作队列（Work Queue）通过使用消息中间件系统发布消息来添加工作任务。一个任务消息通常将由一个进程（最好是一组进程）消费，然后对其进行某种处理。

工作队列在消息处理时间上通常与其他处理方式略有不同。大多数常规数据处理（如 ETL 或简单处理）在毫秒到秒级之间完成。但是工作队列用例下的消息处理时间通常相对更长，从几秒到几分钟，甚至几个小时。

工作队列也被称为分布式工作队列（distributed work queue）。因为单台机器或者单个进程不足以满足相应的处理需求。我们必须通过在许多不同的进程和计算机之间分发工作任务来满足相应的处理需求。因为引入了分布式的技术，所以任务处理系统的复杂性也增加了 10-15 倍。

工作队列的示例

为了帮助您理解工作队列，让我举几个我在现实世界中看到的简单示例。所有示例的共同点是：任务的处理需要更长的时间，同时必须尽快获得结果。

视频转码

在视频网站，用户需要上传视频。此视频将保存到对象存储中。完成视频上传后，Web 服务将发布一条关于该视频的消息，包含要转码的视频的存储地址。这条消息将由一组消费者消费，这些消费者将对此视频进行转码，以便以网络友好的格式使用。转码可能需要几分钟到几小时才能完成。一旦视频转码完成，处理转码的消费者将发布一条新的消息告知此视频已经可以使用。

语音识别与场景分析

通常自动客服系统需要处理来自于呼叫中心的电话呼叫数据。当一个电话呼叫完成后，有多个步骤会发生。首先，语音识别的程序需要处理呼叫的语音对话，将音频转化为文本。接下来，使用各种 NLP 和场景分析工具将对文本进行分析。整个处理过程需要 1-60 分钟。当整条电话呼叫数据被处理后，系统还需要发布一条消息来告知该呼叫数据已被处理完成。

工作队列的挑战

工作队列的最根本的挑战是如何分发和均衡工作任务。你需要确保一个长时间运行的进程不会导致队列中的其他工作任务大量累积；其他的工作进程能够持续的拿到工作任务进行处理。同时，你还需要能够随着流量的变化自动地扩展工作处理能力。

工作队列的最大挑战在于如何处理错误：

• 你如何知道一个工作进程是否挂掉或者何时挂掉？

• 你如何重新开始处理？

• 你如何检测一个工作进程是否挂掉？

这些问题的答案都是针对特定技术的。当你的应用场景中需要使用工作队列的时候，这些答案对你选择何种技术至关重要。

为什么不批量处理

一个常见问题是为什么要使用实时系统而不是批处理系统？批处理系统具有固有的启动时间。对于 30 秒的处理时间，你可能需要花费 5-10 秒等待分布式系统分配和启动资源。实时工作队列的关键之一是得到处理结果的速度。批处理系统处理此类数据的效率太低。

使用 Kafka 实现工作队列

既然您已经了解了工作队列以及与它们相应的挑战，那么接下来，我们来讨论一下如何使用 Apache Kafka 创建工作队列。

High Watermark 和工作队列

在了解如何使用 Kafka 创建工作队列之前，你需要了解 Kafka 消费者如何标记他们已经消费了消息。Kafka 消费者通过提交偏移量来执行此任务。Kafka 消费者可以使用 commitSync 或 commitAsync 来提交消费偏移量。

Kafka 消费者使用 High Watermark 来代表偏移量。这也就意味着消费者只能说“我已经处理到这个偏移量”，而不是“我已经处理过这个消息”。这是 Kafka 跟其他消息中间件的一个重要区别。Kafka 没有内置的方式来确认单条消息。

这种以 High Watermark 记录消费者偏移量的方式意味着应用程序无法单条地记录和发现错误。例如，如果消费者正在处理来自同一分区的两条工作任务而其中一件失败，那么 Kafka 缺乏内置的方式来告诉应用程序这一条消息失败了但另外一件消息成功了。Kafka 的客户端可以告知应用程序已经确认处理到了这一个偏移量，但是没法告知哪条消息成功和哪条消息失败。

要绕过这个限制，你需要让每个消费者将每个分区视为它自己的工作“线程”。每个分区将被限制为一次处理一件事。当消费者完成这项工作时，它将调用 commitSync 标记处理完成。

由于您要在分区中保持长时间运行，因此您必须创建更多分区来有效地处理数据。虽然您可能已经开始使用 20-30 个分区，但您最终需要 100 个分区。这个分区数量是受限于 Kafka 的消费方式，因为整个消费者群需要足够的分区来有效地分配负载。

可能不言而喻，您需要根据您要处理的工作量来分配恰当的分区数量来相应地扩展您的消费组。

管理自己的提交

你会注意到我多次使用“内置”这个词。这是因为还有另一种选择，它不是 Kafka 内置的。你必须编写自己的代码来实现单条确认。

如您所见，Kafka 消费者的问题是受限于 High Watermark 的消费方式。您可以通过编程的方式来自己处理消费者的偏移量。最简单的方法是使用数据库：你必须关闭 Kafka 的自动偏移提交；然后你可以在数据库而不是 Kafka 中记录你处理过消息的偏移量。

当消费者重新启动时，消费者需要知道它相应的分区分配，然后进行数据库查找以查找最后一个偏移量及其状态。如果最后一个偏移量出错，则消费者将重新处理该消息。

虽然这会增加更多的开销，但这是我向使用 Kafka 的团队推荐的一种方法。

使用 Pulsar 实现工作队列

既然您已经了解了如何使用 Kafka 创建工作队列，那么让我们来看看如何使用 Apache Pulsar 来实现，并进行对比。

选择性确认

我们在上一节了解过 Kafka 的 High Watermark。Pulsar 在支持这种类型的确认的同时，还支持另外一种类型的确认，叫做选择性确认。选择性确认允许消费者仅确认单条消息。您可以访问Pulsar的官方文档中关于 Ack 的部分，了解更多有关选择性确认的细节。

当谈到工作队列时，选择性确认让工作队列变得简单。在一个工作队列中，我们可以通过使用选择性 acknowledge 仅仅确认我们已经确认处理过的消息。

如果要获取那些处理失败了的消息，你可以通过调用 redeliverUnacknowledgedMessages 方法来获取。这将使 Pulsar 重新投递那些未经确认的消息。你也可以通过设置另外一个参数 ackTimeout，让 Pulsar 自动重新传递那些超时未确认的消息。

Pulsar 在解决分布式工作队列上还有另外一个优势。在一个拥有很多分布的分布式工作队列中，一些分区仍然有可能成为热点，或者接收大量的任务消息。Pulsar 通过使用共享订阅更好地解决了这个问题。共享订阅允许跨消费者进行循环分发。这样可以比 Kafka 更均匀地分配工作。

在 Pulsar 中，你会将任务消息发布到工作队列中。此消息将一个共享订阅中的许多不同消费者进程中进行消费，然后消费者开始实际的数据处理。一旦完成该处理，消费者将选择性地确认该消息。接着，它会产生一条新的消息，说明改消息已经完成处理。

注意： Pulsar 在开始设计之初就把工作队列的应用场景考虑在内，并针对它进行优化，因为 Yahoo 在内部有大量的工作队列应用场景。这也是我们看到 Pulsar 和 Kafka 之间有如此巨大差异的重要原因。

创建分布式工作队列

选择不同的消息中间件系统将确实改变你实现分布式工作队列的方式。虽然你可以选择任何一种解决方案创建分布式工作队列，但是 Kafka 和 Pulsar 有不同的创建方法。使用 Pulsar 创建分布式工作队列要容易得多。

如果你有工作队列的使用场景，请多多对比确保使用最合适的技术来实现。

点击 链接，查看英文原文