时间戳

首页 > 解决方案 > 芹菜工人从哪个目录开始

python - 芹菜工人从哪个目录开始

问题描述

我需要一些关于芹菜工人的帮助。我尤其无法理解需要从哪里（哪个目录）触发 celery worker 命令，以及它背后的概念是什么以及与导入有关的一些事情。

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

.
├── __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 文件之间存在明显的隔离。有人可以建议安排我的任务和芹菜配置等的更好方法吗？上述两个问题如何与这个新提议的结构保持一致？

标签： pythoncelerycelery-taskceleryd

解决方案

好的，希望这可能会有所帮助。我会按照你的要求反过来回答。

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

我在您添加的片段中看到的第一个问题，taskn.py您拥有的每个片段都有他自己的celery. 您需要在每个taskn.py. 我推荐的是创建一个celery_app.py

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

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

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你可以导入这个实例，每个任务都将注册在同一个 celery 应用程序下

from my_app.celery_app import celapp

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

我从哪里开始工人。我的意思是我应该从哪个目录启动 Celery -A worker -l info 命令，为什么？

然后你应该很容易调用Celery -A my_app.celery_app worker -l info，因为你的 celery 实例将在模块 my_app，子模块 celery_app

导入entry.py中所有任务的导入语句是什么

最后，entry.py您可以执行import state1.family1.task1 import t1和调用t1.delay()或任何已注册的任务。

推荐阅读