芹菜工人的目录是从开始的

Question

我对芹菜工人需要一些帮助。我尤其无法理解芹菜工人的命令需要从哪里(哪个目录(发射，其背后的概念以及有关进口的一些概念。

所以说我有以下目录结构：

.
├── __init__.py
├── entry.py
├── state1
│   ├── __init__.py
│   ├── family1
│   │   ├── __init__.py
│   │   ├── task1.py
│   │   ├── task2.py
│   │   └── task3.py
│   └── family2
│       ├── __init__.py
│       └── task1.py
└── state2
    ├── __init__.py
    ├── family1
    │   ├── __init__.py
    │   ├── task1.py
    │   └── task2.py
    └── family2
        ├── __init__.py
        ├── task1.py
        └── task2.py

.根是当前的工作目录，命名project

taskn.py(task1.py，task2.py等(的每个任务都是单个任务。每个任务文件看起来像这样：

from celery import Celery
from celery.result import AsyncResult
from kombu import Queue
_name_ = "project_x"
celapp=Celery(backend='redis://localhost:6379/0', broker='amqp://a:b@localhost/a_vhost')
CELERY_CONFIG = {
    'CELERY_DEFAULT_QUEUE': 'default',
    'CELERY_QUEUES': (Queue('q1'), Queue('q2'),),
    'CELERY_TASK_SERIALIZER': 'pickle',
    'CELERY_ACCEPT_CONTENT': ['json','pickle']
}
celapp.conf.update(**CELERY_CONFIG)
@celapp.task()
def t1():
    print("starting task")
    time.sleep(5)
    print("Finished task")

以下是entry.py的内容：

import json
from flask_cors import CORS
from flask import Flask, Response, render_template
from flask import request, jsonify, redirect
from functools import wraps
<what would be the import statement to import all the tasks>
_name_ = "project_x"
app     = Flask(_name_)
@app.route("/api1", methods=['POST'])
def api1():
    req = request.jsonify
    if not req:
        return jsonify(success=False, msg="Missing request parameters", code="1")
    else:
        param1 = req.get('p1')
        param2 = req.get('p2')
        tId = startTask()
        return jsonify(success="True", msg="All Good", taskId=tId)

def startTask():
    tId = "abcd123"
    created_task = state1.family1.task1.subtask(queue='q1')
    created_task.delay()
    return tId

if __name__ == '__main__':
    app.run(debug=True, host="192.168.1.7", port="4444")

entry.py是烧瓶应用程序，从其中触发API1，然后根据我想启动特定任务的参数。

现在这是我的问题：

1. 导入entry.py文件中所有任务的导入语句
2. 我从哪里开始工人。我的意思是我应该从哪个目录启动Celery -A <directory name> worker -l info命令，为什么？
3. 在许多示例中，我看到任务和CeleryApp文件之间存在明显的隔离。有人可以建议什么是安排我的任务的更好方法，芹菜配置等。上面的两个问题如何与这个新提出的结构保持一致？

Answer 1

好吧，希望这可能会有所帮助。我会按照您的要求进行逆转。

在许多示例中，我看到任务之间存在明显的隔离 和CeleryApp文件。有人可以建议什么 更好的方法来安排我的任务，芹菜配置等，以及如何 以上两个问题与这个新提出的结构保持一致？

我在添加的片段中看到的第一个问题，即您拥有的每个taskn.py，都有他自己的celery实例。您需要在每个taskn.py之间共享此实例。我推荐什么是创建celery_app.py

my_app
├── __init__.py
├── entry.py
├── celery_app.py
│   ├── ...

在此文件中，您将创建芹菜实例

from celery import Celery
from celery.result import AsyncResult
from kombu import Queue
_name_ = "project_x"
celapp=Celery(backend='redis://localhost:6379/0', broker='amqp://a:b@localhost/a_vhost')
CELERY_CONFIG = {
    'CELERY_DEFAULT_QUEUE': 'default',
    'CELERY_QUEUES': (Queue('q1'), Queue('q2'),),
    'CELERY_TASK_SERIALIZER': 'pickle',
    'CELERY_ACCEPT_CONTENT': ['json','pickle']
}
celapp.conf.update(**CELERY_CONFIG)
celery_app.conf.imports = [
    'state1.family1.task1',
    'my_app.state1.family1.task2',  # Or Maybe
    ...
]

然后，在每个taskn.py中，您都可以导入此实例，每个任务都将在同一芹菜应用程序下注册

from my_app.celery_app import celapp
@celapp.task()
def t1():
    print("starting task")
    time.sleep(5)
    print("Finished task")

我从哪里开始工作。我的意思是我应该从哪个目录 启动芹菜-a工作-L信息命令，为什么？

然后，您应该轻松调用Celery -A my_app.celery_app worker -l info，因为您的芹菜实例将在模块MY_APP中，subsodule celery_app

将导入输入中的所有任务的导入语句是什么。

最后，您可以从entry.py进行import state1.family1.task1 import t1并致电t1.delay()或任何注册任务。

Answer 2

因此，从@patricio那里获取建议，似乎这确实是一个导入错误。我的新目录结构看起来如下：

.
├── __init__.py
├── celeryConfig
│   ├── __init__.py
│   └── celeryApp.py
├── entry.py
├── state1
│   ├── __init__.py
│   ├── family1
│   │   ├── __init__.py
│   │   ├── task1.py
│   │   ├── task2.py
│   │   └── task3.py
│   └── family2
│       ├── __init__.py
│       └── task1.py
└── state2
    ├── __init__.py
    ├── family1
    │   ├── __init__.py
    │   ├── task1.py
    │   └── task2.py
    └── family2
        ├── __init__.py
        ├── task1.py
        └── task2.py

celeryConfig/celeryApp.py的内容如下：

from celery import Celery
from celery.result import AsyncResult
from kombu import Queue
_name_ = "project_x"
celapp=Celery(backend='redis://localhost:6379/0', broker='amqp://a:b@localhost/a_vhost', include=['state1.family1.task1'])
CELERY_CONFIG = {
    'CELERY_DEFAULT_QUEUE': 'default',
    'CELERY_QUEUES': (Queue('q1'), Queue('q2'),),
    'CELERY_TASK_SERIALIZER': 'pickle',
    'CELERY_ACCEPT_CONTENT': ['json','pickle']
}
celapp.conf.update(**CELERY_CONFIG)

和taskn.py的内容类似于：

from celeryConfig.celeryApp import celapp
import time
@celapp.task()
def t1():
    print("starting task")
    time.sleep(5)
    print("Finished task")

虽然entry.py保持原样，但只有一个更改如下：

from state1.family1.task1 import t1

现在，当芹菜开始为： celery -A celeryConfig.celeryApp worker -l info从根目录中，project，一切都很好。作为上述命令的输出，我将消息作为

获取

.
.
.
[tasks]
  . state1.family1.task1.t1
.
.
.

表明芹菜已经正确开始，并且确实已经注册了任务。因此，现在，为了获得所有已注册的任务，我可以通过目录/目录阅读并动态地在celeryApp.py中创建include列表。(完成后将发布更多有关它的信息(

谢谢@patricio