Bull Queue并发问题

Question

我需要帮助理解Bull Queue (bull.js)如何处理并发作业。

假设我有10个Node.js实例，每个实例都实例化一个连接到同一Redis实例的Bull Queue：

const bullQueue = require('bull');
const queue = new bullQueue('taskqueue', {...})
const concurrency = 5;
queue.process('jobTypeA', concurrency, job => {...do something...});

这是否意味着在全局范围内，所有10个节点实例中将有最多5个(并发)并发运行jobTypeA类型的作业？或者是我误解了，并发设置为per-Node instance？

如果一个Node实例指定了不同的并发值，会发生什么情况？

我能确定作业不会被多个Node实例处理吗？

Answer 1

TL;DR是:在正常情况下，作业只被处理一次。如果出现问题(比如Node.js进程崩溃)，作业可能会被双重处理。

引用公牛官方README.md

重要说明

队列的目标是“至少一次”的工作策略。这意味着在某些情况下，一个作业可能会被处理多次。这主要发生在工作进程在整个处理过程中未能锁定给定作业的情况下。

当一个工人在处理一个工作时，它将保持该工作“锁定”，这样其他工人就不能处理它。

重要的是要理解锁是如何工作的，以防止作业失去锁--变得停滞--并因此被重启。锁定是通过在间隔lockRenewTime (通常是一半lockDuration)上为lockDuration创建一个锁来在内部实现的。如果在更新锁之前lockDuration已经过去，则该作业将被视为已停止并自动重新启动；它将是double processed。在以下情况下可能会发生这种情况：

1. 运行你的job processor的节点进程意外地terminates.
2. Your job processor太耗费CPU资源，导致节点事件循环停滞，因此，Bull无法更新作业锁(参见#488了解如何更好地检测到此问题)。您可以通过将job processor分解为较小的部分来修复此问题，以便任何单个部分都无法阻止节点事件循环。或者，您可以为lockDuration设置传递一个更大的值(但代价是需要更长的时间才能识别真正停止的作业)。

因此，您应该始终侦听stalled事件，并将其记录到错误监视系统中，因为这意味着您的作业可能会被双重处理。

作为一种保护措施，有问题的作业不会无限期地重新启动(例如，如果作业处理器总是使其节点进程崩溃)，作业将从停滞状态恢复最多maxStalledCount时间(默认值：1)。

Answer 2

由于要面对a problem with too many processor threads，我花了很多时间深入研究它。

简而言之，公牛的并发性是在队列对象级别，而不是队列级别。

如果深入研究代码，就会在对队列对象调用.process时调用并发性设置。这意味着，即使在同一个Node应用程序中，如果您创建多个队列并多次调用.process，它们也会增加可以处理的并发作业的数量。

一位贡献者发布了以下内容：

是的，当我第一次使用公牛的时候，我也感到有点惊讶。队列选项在Redis中永远不会持久化。每个应用程序可以有任意多个队列实例，每个实例可以有不同的设置。并发设置是在注册处理器时设置的，它实际上特定于每个process()函数调用，而不是队列。如果使用命名处理器，则可以多次调用process()。每个调用将按照并发数量(默认值为1)注册N个事件循环处理程序(具有节点的process.nextTick())。

因此，您的问题的答案是:是的，如果您在多个节点实例中注册进程处理程序，您的流程将由多个节点实例处理。

Answer 3

Bull被设计为与“至少一次”语义并发处理作业，尽管如果处理器工作正常，即没有停顿或崩溃，它实际上是在交付“恰好一次”。但是，您可以将最大延迟重试次数设置为0 (maxStalledCount https://github.com/OptimalBits/bull/blob/develop/REFERENCE.md#queue)，然后语义将为“至多一次”。

话虽如此，我将尝试回答发帖者提出的两个问题：

如果一个节点实例指定了不同的并发值，会发生什么情况？

我假设你指的是“队列实例”。如果是，则在处理器中指定并发性。如果并发数是X，那么给定的处理器最多只能并发处理X个作业。

我能确定作业不会被多个节点实例处理吗？

可以，只要您的作业没有崩溃，或者您的最大延迟作业数设置为0。