python - Python Celery 可以从进程内线程开始吗？

Question

我想用我的芹菜代码做一个测试用例。但通常 celery 需要从一个新进程开始$ celery -A CELERY_MODULE worker，这意味着我不能直接运行我的测试用例代码？

我正在为 Celery 配置内存存储，以使测试用例中的额外 I/O 无效。那是配置不能在不同的进程中共享任务队列。

Answer 1

这是我天真的工具。来自 的 celery 条目celery.bin.celeryd.WorkCommand，它解析 args 并执行工作。
在这种情况下使用solo来取消 MultiProcess 的使用。当然，您需要先安装lib。

您可以在 celery 测试用例开始之前使用它。

#!/usr/bin/env python
#vim: encoding=utf-8

import time
import unittest
from threading import Thread

from celery import Celery, states
from celery.bin.celeryd import WorkerCommand

class CELERY_CONFIG(object):
    BROKER_URL = "memory://"
    CELERY_CACHE_BACKEND = "memory"
    CELERY_RESULT_BACKEND = "cache"
    CELERYD_POOL = "solo"

class CeleryTestCase(unittest.TestCase):
    def test_inprocess(self):
        app = Celery(__name__)
        app.config_from_object(CELERY_CONFIG)

        @app.task
        def dumpy_task(dct):
            return 321

        worker = WorkerCommand(app)
        #worker.execute_from_commandline(["-P solo"])
        t = Thread(target=worker.execute_from_commandline, args=(["-c 1"],))
        t.daemon = True
        t.start()
        ar = dumpy_task.apply_async(({"a": 123},))
        while ar.status != states.SUCCESS:
            time.sleep(.01)
        self.assertEqual(states.SUCCESS, ar.status)
        self.assertEqual(ar.result, 321)
        t.join(0)