【问题标题】:Setting environment variables with Redis Queue使用 Redis Queue 设置环境变量
【发布时间】:2021-02-12 23:47:23
【问题描述】:

我是 RQ 的新手,我正在尝试在我的烧瓶应用程序中实现它。我的一条路线的主要目标是更新数据库中的值。要设置我的工作人员,我使用以下内容:

from rq import Worker, Queue, Connection
import redis
import os

@app.before_first_request
def start_worker():
    def runworker():
        redis_url = os.environ.get("REDIS_URL") or 'redis://'
        conn = redis.from_url(redis_url)
        with Connection(conn):
            worker = Worker(list(map(Queue, listen)))
            worker.work()
    tp = ThreadPoolExecutor()
    tp.submit(runworker)


def get_redis_connection():
    redis_connection = getattr(g, '_redis_connection', None)
    if redis_connection is None:
        redis_url = os.environ.get('REDIS_URL') or 'redis://'
        redis_connection = redis.from_url(redis_url)
    return redis_connection


@app.before_request
def push_rq_connection():
    push_connection(get_redis_connection())


@app.teardown_request
def pop_rq_connection(exception=None):
    pop_connection()

然后更新路由将更新作业排队

@app.route('/update')
def update_db():
    q = Queue(connection=conn)
    job = q.q.enqueue('app.tasks.update_task', parameters)
    job_id = job.get_id()
    return {"job_id": job_id}, 201, {"Content-Type": 'application/json'}

最后,worker运行更新函数

def update_task(parameters):
    # script to update DB

我知道工人设置(几乎)有效,因为如果我将 update_task 功能切换为简单的东西,例如:

def update_task(seconds):
    for i in range(seconds):
        print(i)
        time.sleep(1)
    return "Hello world"

它有效。然而,对于真正的函数,我一直遇到我的环境变量没有被定义的问题,因为当我运行实际更新时,我收到异常说我的变量是 None 或类似的东西。

有人知道如何在 RQ 中处理环境变量吗?我应该在某个配置文件之类的地方再次声明它们吗?

【问题讨论】:

  • 答案取决于您在哪个操作系统上运行它。您能否将这些信息添加到问题中?
  • Ubuntu 19 或 20
  • 啊,好的。您可以像这样在 Bash 中设置环境变量:export REDIS_URL=redis://... 其中... 将替换为您的 Redis URL。
  • 是的,但我认为这不是问题所在。正在定义变量是因为烧瓶可以正常使用它们。 RQ 唤醒者是没有得到它们的那个。

标签: python flask environment-variables rq


【解决方案1】:

好的,最好使用multiprocessing

from multiprocessing import Process

def runworker():
    redis_url = os.environ.get("REDIS_URL") or "redis://"
    conn = redis.from_url(redis_url)
    listen = ['default']
    with Connection(conn):
        worker = Worker(list(map(Queue, listen)))
        worker.work()

pp_worker = Process(target=runworker)
pp_worker.start()


@app.teardow_appcontext
def teardown_rq(self):
    pp_worker.terminate()

aaand 就行了。只是如果有人想知道它。

【讨论】:

    猜你喜欢
    • 2013-04-14
    • 2013-03-04
    • 2015-10-21
    • 2018-08-28
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    相关资源
    最近更新 更多