1

我的用例是创建动态延迟作业。（我正在使用可用于创建延迟作业的Bulls Queue 。）

根据某些事件，向延迟间隔添加更多延迟（进一步延迟作业）。

由于我找不到任何function更新作业的延迟间隔，我想出了以下步骤：

onEvent(jobId):
  // queue is of Type Bull.Queue
  // job is of type bull.Job
  job = queue.getJob(jobId)
  data = job.data
  delay = job.toJSON().delay
  job.remove()
  queue.add("jobName", {value: 1}, {jobId: jobId, delayed: delay + someValue})

这几乎解决了我的问题。

但我担心这些操作会发生的规模。

我预计在不久的将来每分钟会有近50K事件甚至更多。

我的队列大小预计会根据唯一的JobId增长。

我期待的不仅仅是：

  • 每日100万次入场
  • 每周约 4-5 百万次入场
  • 10-12 百万每月进入。

此外，在60-70 天后，作业的延迟间隔将达到，并且较旧的作业将被一一删除。

我可以运行多个处理器来处理这些延迟的工作，这不是问题。

我的队列大小将在 60-70 天后稳定下来，我的队列或多或少会有大约 1000 万个工作岗位。

我可以根据需要垂直缩放我的REDIS。

但我想了解以下查询的时间复杂度：

 queue.getJob(jobId)  // Get Job By Id
 job.remove() // remove job from queue
 queue.add(name, data, opts) // add a delayed job to this queue

如果这些操作中的任何一个是 O(N) 或者QUEUE可以保留一些小于 1000 万的最大作业数。

然后我可能不得不放弃这个设计并想出一些完全不同的东西。

需要有经验的人的建议，他们可以指导我如何解决这个问题。

任何形式的帮助表示赞赏。

4

1 回答 1

1

参考源代码：

queue.getJob(jobId)

这应该是 O(1)，因为它主要使用使用 hmget 的基于哈希的解决方案。您只请求一份工作，根据官方 redis 文档，时间复杂度为 O(N)，其中 N 是请求的键数，将按 O(1) 的顺序排列，因为我预计公牛存储很少哈希键的字段数。

作业.remove()

考虑到您的大量工作将被延迟，并且其中一小部分被移动到等待或活动队列。这应该是摊销水平上的 O(logN)，因为它主要使用zrem进行这些操作。

queue.add（名称，数据，选择）

对于延迟队列中的作业添加，bull 正在使用zadd，所以这又是 O(logN)。

于 2021-08-13T11:18:38.403 回答