如何控制每个队列的并发性？答案

【问题标题】：How to control concurrency per queue?如何控制每个队列的并发性？
【发布时间】：2015-03-01 09:56:27
【问题描述】：

Sidekiq 文档建议我只能控制 sidekiq 的全局并发，而不是每个队列。我在这里提出一个问题，希望有针对每个队列并发设置的解决方案。一些 3rd 方服务不接受高并发，并且仅仅为了这些限制整个 sidekiq 是很痛苦的。

我正在使用 sidekiq 3.3

【问题讨论】：

【解决方案1】：

使用 Heroku，我可以通过在 Procfile 中设置环境变量然后在 sidekiq.rb 初始化程序中使用它来控制每个队列的并发：

Sidekiq.configure_server do |config|
 config.options[:concurrency] = (ENV['SIDEKIQ_WORKERS_PROCFILE'] || ENV['SIDEKIQ_WORKERS'] || 1).to_i
 ...
end

SIDEKIQ_WORKERS_PROCFILE 在 Procfile 中为一个队列设置 - 其他队列使用在 Heroku 设置中设置的 SIDEKIQ_WORKERS。

我不确定这是否对您的场景有帮助。

更新

为了澄清这一点，这个想法涉及在 Heroku 上的部署，并且每个队列都在单独的 dyno 中处理。 这使得它仍然使用全局 sidekiq 并发设置，dynos 只是一种解决方法，可以在我的用例。

我的Procfile 看起来像这样：

web: bundle exec unicorn -p $PORT -c ./config/unicorn.rb
default: env HEROKU_PROCESS=default bundle exec sidekiq -c 5
important: env HEROKU_PROCESS=important bundle exec sidekiq -q important -c 5
instant: env HEROKU_PROCESS=instant bundle exec sidekiq -q instant -c 5
matrices: env HEROKU_PROCESS=matrices SIDEKIQ_WORKERS_PROCFILE=1 bundle exec sidekiq -q matrices -c 1

您可以看到matrices worker 将SIDEKIQ_WORKERS_PROCFILE 变量设置为 1 - 这使得可以在队列中以不同的并发运行 worker。该变量由sidekiq.rb 初始化程序读取。请注意，还有 -c 1 选项 - 不过我不知道这是否重要。

初始化程序已经在那里了。

所有设置完毕，在 sidekiq 仪表板中，我可以看到 matrices 队列正在运行 1 个线程，而其他使用 3 个线程（SIDEKIQ_WORKERS 变量在 Heroku 设置环境变量中设置为 3）：

【讨论】：

我看不出这是如何在特定队列上设置并发的？
@Dty 我稍后会更新响应，但在 Procfile 中为给定队列设置了SIDEKIQ_WORKERS_PROCFILE，例如：important: env HEROKU_PROCESS=important SIDEKIQ_CONCURRENCY=1 bundle exec sidekiq -q important -c 1，而其他人则为：instant: env HEROKU_PROCESS=instant bundle exec sidekiq -q instant -c 5
抱歉，我错过了什么吗？我仍然看不到这如何指定特定队列的并发性。我真的很想能够做到这一点，但 sidekiq 作者说这在 2014 年是不可能的 stackoverflow.com/questions/27537799/…
非常感谢您的澄清。这可能非常适合我的用例。
据我了解，该解决方案仅适用于 heroku ？

【解决方案2】：

Sidekiq 企业版增强了并发控制：

https://github.com/mperham/sidekiq/wiki/Ent-Rate-Limiting

特别是，您可以通过计数或间隔限制任意作业的并发性。例如一次最多 20 个工作；或最多 20 个作业（60 秒/1 小时等）。间隔可以滚动或时钟对齐。

这可以通过正确的用法满足您询问的每个队列控制。但它可以更加灵活，通过您自己的分组来控制并发。例如，您可以指定每秒最多 30 个作业可以访问 PayPal。或者，您可以指定每个州每秒 30 个工作可以访问 Paypay，每个州 15 个工作可以同时访问 Stripe。（当然，假设“状态”是您数据中的一个属性。）

您希望定义组的细粒度没有限制。

【讨论】：

【解决方案3】：

我只是从 yml 文件中删除并发，并在启动 worker 时更改它。

sidekiq.yml

:queues:
  - default
  - long_running

我在 docker 环境中运行我的工人。我只是为每个队列运行单独的工作人员：

默认工作人员以

开头

sidekiq -q default -c 5

长时间运行的工作人员以

开头

sidekiq -q long_running -c 2

或者，您可以使用单独的 yml 配置并使用 -C 选项传递不同的配置

【讨论】：

【解决方案4】：

要处理队列明智的并发，您可以使用这个 gem sidekiq-limit-fetch。它修复了队列可以使用的最大线程数。

【讨论】：