15

我们最近被迫用 RQ 替换 celery，因为它更简单，而且 celery 给我们带来了太多问题。现在，我们无法找到动态创建多个队列的方法，因为我们需要同时完成多个作业。因此，基本上对我们的一个路由的每个请求都应该启动一项工作，让多个用户等待一个用户的工作完成，然后我们才能继续下一个工作是没有意义的。我们会定期向服务器发送请求，以获取作业的状态和一些元数据。这样我们可以用进度条更新用户（这可能是一个漫长的过程，所以为了用户体验必须这样做）

我们正在使用 Django 和 Python 的rq库。我们没有使用django-rq（请让我知道使用它是否有优势）

到目前为止，我们在其中一个控制器中开始了一项任务，例如：

redis_conn = Redis()
q = Queue(connection=redis_conn)  
job = django_rq.enqueue(render_task, new_render.pk, domain=domain, data=csv_data, timeout=1200)

然后在我们的render_task方法中，我们根据长任务的状态将元数据添加到作业中：

current_job = get_current_job()
current_job.meta['state'] = 'PROGRESS'
current_job.meta['process_percent'] = process_percent
current_job.meta['message'] = 'YOUTUBE'
current_job.save()

现在我们有了另一个端点，它获取当前任务及其元数据并将其传递回客户端（这通过 oeriodic AJAX 请求发生）

我们如何在不阻塞其他作业的情况下同时运行作业？我们应该动态地创建队列吗？有没有办法利用工人来实现这一目标？

4

3 回答 3

9

据我所知，RQ 没有任何设施来管理多个工人。您必须启动一个新的工作进程来定义它将使用哪个队列。一种对我来说效果很好的方法是使用Supervisor。在主管中，您为给定的队列和进程数量配置您的工作人员以具有并发性。例如，您可以让 5 个工作人员的“高优先级”队列和 1 个工作人员的“低优先级”队列。

于 2016-05-11T18:05:25.883 回答
7

运行多个工人不仅是可能的而且是理想的。我使用 bash 文件作为启动命令进入虚拟环境，并使用自定义 Worker 类启动。

这是一个主管配置，在生产工作负载下对我来说非常适合 RQ 工作人员。请注意，startretries很高，因为它在 AWS 上运行并且需要在部署期间重试。

[program:rq-workers]
process_name=%(program_name)s_%(process_num)02d
command=/usr/local/bin/start_rq_worker.sh
autostart=true
autorestart=true
user=root
numprocs=5
startretries=50
stopsignal=INT
killasgroup=true
stopasgroup=true
stdout_logfile=/opt/elasticbeanstalk/tasks/taillogs.d/super_logs.conf
redirect_stderr=true

start_rq_worker.sh 的内容

#!/bin/bash
date > /tmp/date
source /opt/python/run/venv/bin/activate
source /opt/python/current/env
/opt/python/run/venv/bin/python /opt/python/current/app/manage.py
rqworker --worker-class rq.SimpleWorker default
于 2019-05-09T15:29:02.523 回答
3

我想建议一个使用django-rq的非常简单的解决方案：

示例设置.py

...

RQ_QUEUES = {
    'default': {
        'HOST': os.getenv('REDIS_HOST', 'localhost'),
        'PORT': 6379,
        'DB': 0,
        'DEFAULT_TIMEOUT': 360,
    },
    'low': {
        'HOST': os.getenv('REDIS_HOST', 'localhost'),
        'PORT': 6379,
        'DB': 0,
        'DEFAULT_TIMEOUT': 360,
    }
}

...

运行配置

运行python manage.py rqworker default low次数（每次在其自己的 shell 中，或作为其自己的 Docker 容器，例如）与所需工作人员的数量一样多。命令中的队列顺序决定了它们的优先级。此时，所有工作人员都在监听两个队列。

在代码中

调用作业运行时，传入所需的队列：

对于高/普通优先级作业，您可以不带任何参数进行调用，作业将进入默认队列。对于低优先级，您必须在作业级别指定：

@job('low')
def my_low_priority_job():
  # some code

然后调用my_low_priority_job.delay()。

或者，在调用时确定优先级：

queue = django_rq.get_queue('low')
queue.enqueue(my_variable_priority_job)
于 2016-11-30T08:47:04.450 回答