问题标签 [job-queue]

我想使用 node.js Bull处理预定的作业。基本上我有两个处理器来处理两种类型的工作。有一个配置器可以配置将使用 cron 添加到公牛队列的作业。

调度程序将位于一个微服务中，每个处理器将是一个单独的微服务。所以我将拥有 3 个微服务。

我的问题是我在公牛上使用正确的模式吗？

index.js

处理器配置器.js

fetcher-configurator.js

fetcher.js

警报处理器.js

将有三个微服务 -

1. 节点索引.js
2. 节点 fetcher-configurator.js
3. 节点处理器配置器.js

我看到公牛的行为不一致。有时我收到错误缺少作业类型的流程处理程序

我正在使用以下代码提交 5 个作业，但第一个作业比其余 4 个作业需要大约一个小时，尽管处理的记录数相同。

假设我有一个简单的昂贵函数，可以将一些结果存储到文件中：

现在假设我想在 dask 中安排一些这样的任务，然后异步接收这些请求并一个一个地运行这些函数。我目前正在设置一个 dask 客户端对象...

...然后以交互方式（来自 IPython）调度这些作业：

我遇到的问题是这些任务不是连续运行而是并行运行，在我的特殊情况下会导致并发问题。

所以我的问题是：像我上面在 dask 中描述的那样设置作业队列的正确方法是什么？

resolve()下面的函数和函数是否真的是then()ES6 Promise 机制中唯一查看并添加履行处理程序到作业队列的函数（并将它们标记为已添加且不再添加）？

上述代码的替代方法resolve()是直接完成，而不是在setTimeout(). 是真的then(fn)会为这个promise添加fn到履行处理程序表（不是作业队列）中，还是如果promise已经解决了，直接添加fn到作业队列中，并且fn标记为“已经添加到作业队列中” " 并且不需要再次添加？

另一方面，这resolve()是一个 JS ES6 提供的函数，它将一个 Promise 标记为“已解决”（状态），并检查该 Promise 的履行处理程序表是否为空。如果有这样的处理程序（并且之前不应该被标记为已添加到作业队列中），那么这些处理程序将被添加到作业队列中并标记为已添加。

在任何情况下，假设履行处理程序表中有fn1, fn2, fn3, ... 等，每个都添加到作业队列中fn1(resolvedValue)，调用resolvedValue内部会记住resolve(resolveValue)?

并且根本没有其他情况可以访问履行处理程序表。作业队列作业由 JS 事件循环执行，因此当所有 GUI“点击事件处理程序”等都完成后，作业队列中的履行（和拒绝）处理程序被执行，然后是 setTimeout 和 setInterval处理程序也被执行？这就是履行处理程序表和作业队列的工作方式吗？

以下代码：

将“点 B”和“点 C”的打印添加到作业队列中，之后，打印出“去喝一杯”的处理程序也添加到作业队列中。但是在 Node 和 Google Chrome 的结果中，“去喝一杯”在“点 B”和“点 C”之前打印。是不是promise有单独的job queue和setTimeoutandsetInterval有单独的job queue，promise job queue是在timed job queue之前执行的？似乎 ES6 规范提到了两个作业队列： ScriptJobs 和 PromiseJobs。我们可以指望在定时作业队列之前执行的承诺作业队列吗？

我的项目中有多个公牛队列，NodeJS如果前一个队列成功执行，它们将运行。我正在尝试在这里验证一些电子邮件地址。

1. 检查电子邮件格式（formatQueue）

2. 使用包的电子邮件存在npm email-existence(existenceQueue)

该formatQueue过程花费的时间更少，它将运行RegEx并验证Email格式。但该email-existence包大约需要 5-10 秒才能完成。

formatQueueexistenceQueue如果少于jobs20-100 ，则可以正常工作。但是当我一次添加超过 1000 个工作时，existenceQueue失败并出现以下错误

我检查了HERE和HERE的问题，我认为该过程需要很长时间才能响应，所以limiter按照HERE的引用添加。但这对我没有帮助。

如果任何队列中的作业失败，则不会处理下一个作业。它将停在那里，其他工作将保持waiting状态。

我的代码类似于下面的代码。请帮我解决这个问题。

队列.js

- - - - - 更新 - - - - -

处理器花费了太多时间来响应，因此stalled由于我使用超时，因此工作正在获得或失败。

我正在尝试process在不同的processor文件中运行 itef 作为其在bull文档中，我添加了如下文件。

现在在我之前使用的作业完成时，我曾经使用更新的作业参数获取作业数据。现在我没有得到更新的工作数据。文档中的第二个参数result是undefined. 现在在这种情况下如何获取更新的作业数据。

我们有多个作业消息排队到云存储队列。GetMessage()来自各个作业队列（我们有多个队列）的工作角色并处理它们。

我们正在尝试使用出列可见性和实例标识符从同一队列创建多个工作人员处理消息，但是对于这些工作人员，GetMessage()我无法找到一种方法来有条件地过滤仅分配给该工作人员处理的作业消息。

例如，如果 Worker A 正在寻找具有WorkerName='WorkerA'作为其属性或参数之一的 JobMessages，而不是使用JobQueue.GetMessage()然后阅读每条消息以了解是否WorkerName='WorkerA'有一种方法可以过滤 JobQueue 以仅获取具有 WorkerName = 'WorkerA 的消息'？

我正在尝试process在不同的processor文件 itef 中运行它，因为它在bull 文档中，我添加了如下文件。

现在在我之前使用的作业完成时，我曾经使用更新的作业参数获取作业数据。现在我没有得到更新的工作数据。文档中的第二个参数result是undefined. 现在在这种情况下如何获取更新的作业数据。

和工作正常，我运行job如下。processorsprocess

在这种情况下，作业数据本身正在更新，并且效果也很好。

我有多个队列在做不同的工作，一旦前一个工作成功完成，那么只有我需要另一个工作来开始工作。我还有另一个问题，我已经提到了那里的用例。请在这里查看。

我们每天有很多版本，每个版本都有自己的文件夹。该文件夹current指向最新发布的文件夹。我们将 Laravel 的作业队列功能与 Redis 一起使用。现在有时会发生作业在队列中停留的时间更长，然后发布存在：

• 释放100
• 发布101
• 发布102
• 当前-> release103

使用 创建的作业release103。

release103在从服务器上清除了四个新部署之后：

• 发布104
• 发布105
• 发布106
• 当前-> release107

当现在准备好处理作业时，我收到错误“文件 release103/vendor/.../foo.php 不存在”。

有什么好的方法可以避免这种行为吗？

问题

我需要一个比标准 FIFO 或优先级队列复杂得多的作业队列系统。队列主要需要像标准的 FIFO 队列一样工作，但在作业的出队和运行方面有更复杂的逻辑。在我的情况下，只有在不超过某些用户和系统并发限制的情况下，才应该出队作业（例如，用户只能有 10 个特定类型的并发作业，整个系统只能有 100 个并发作业某种类型等）

目标是让一个 kubernetes 集群成为这个队列的消费者，如果满足所有必要条件，k8 将把作业出列并启动一个新容器来运行它。我不是 k8 的专家，但我认为我们不能让 k8 在对给定作业出队之前运行这些并发检查。所以，我认为我需要的是一个作业队列系统，它通过只允许通过某些检查的作业出列来将这些检查放入队列本身。

我们尝试了什么

我们已经使用 sql 数据库实现了我们自己的作业队列系统。此数据库中有一个主作业队列表，其中包含每个作业的信息。然后，我们创建了自己的应用程序，该应用程序定期（每 10 秒左右）在该表上运行复杂的查询，以确定哪些作业应该出列并运行。然后这个应用程序启动工作进程来运行这些作业（不是在容器中，只是标准进程）。

这种方法有几个问题。首先，查找准备运行的作业的查询非常复杂且缓慢。此外，当系统上有大量活动时，作业队列表可能成为整个系统的巨大瓶颈。此外，由于我们希望开始在它们自己的 docker 容器中运行这些工作进程，因此如果可能的话，我们希望 Kubernetes 集群成为队列的直接消费者，而不是让我们自己的应用程序作为中介。

问题

有哪些流行的方法来处理复杂的作业队列？我无法想象我们是唯一需要施加并发限制的作业队列的人，我也无法想象我们的 SQL 方法是实现我们需要的最佳方式。在这种情况下，我们可以做些什么来使我们的作业队列系统尽可能高效？