来自 python 的 Celery Inspect 任务

Question

from celery import Celery
from celery.worker.control import inspect_command
app = Celery('tasks', broker='pyamqp://guest@localhost//')
@app.task
def add(x, y):
    return x + y

@inspect_command
def current_prefetch_count(state):
    return {'prefetch_count': state.consumer.qos.value}

我尝试使用Celery运行代码，但它给了我一个错误：

File"c:python27libsite-packageskombuutilsimports.py",line56,insymbo
_by_name
typeError: inspect_command() takes exactly 0 arguments (1 given)

另外，我想自定义inspect_command来运行此 sysfile.py

#!/usr/bin/env python
import platform
import celery
import os
import psutil
import json


def Speed_Test():
    Speed_list = (os.popen("speedtest-cli --share --simple").read()).split("n")
    result = [Speed_list[-2].split(":")[-1]]
    result.append(Speed_list[-3].split(":")[-1])
    return result
def Sys_Info():
    inner_Dict = {}
    inner_Dict["CPU Model"]=platform.processor()
    inner_Dict["No of CPU"]=psutil.cpu_count()
    inner_Dict["Disk info"]=psutil.disk_usage('/')
    inner_Dict["celery"] = (celery.__version__)
    inner_Dict["Upload"] = Speed_Test()[0]
    inner_Dict["Download"] = Speed_Test()[1]
    return json.dumps(inner_Dict)
x = Sys_Info()
print x

Answer 1

inspect_command装饰

器可以选择采用一系列kwargs。所以你必须称之为：

@inspect_command()
def current_prefetch_count(state):
    return {'prefetch_count': state.consumer.qos.value}

例如，如果您希望命令具有可以调用它的其他名称，则可以使用 alias ：

@inspect_command(alias='foo')
def current_prefetch_count(state):
    return {'prefetch_count': state.consumer.qos.value}

您可以通过阅读代码来发现所有这些 celery.worker.control .