如何在Python-RQ中创建多个工作线程

Question

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

我们使用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()

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

我们如何在不阻塞其他作业的情况下并发运行作业?我们应该动态地创建队列吗?是否有一种方法可以利用工人来实现这一点?

Answer 1

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

Answer 2

运行多个worker不仅是可能的，而且是理想的。我使用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

Answer 3

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

样本settings.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容器中)与所需的worker的数量相同。命令中队列的顺序决定了它们的优先级。此时，所有工作线程都在监听两个队列。

代码中

当调用一个作业运行时，传入所需的队列:

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

@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)