芹菜从蟒蛇检查任务

from celery import Celery 
from celery.worker.control import inspect_command 
app = Celery('tasks', broker='pyamqp://[email protected]//') 

@app.task 
def add(x, y): 
    return x + y 


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

我试图运行与芹菜的代码，但它给了我一个错误：芹菜从蟒蛇检查任务

File"c:\python27\lib\site-packages\kombu\utils\imports.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

来源

2017-01-11 Dev Jalla

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的代码。

来源

2017-01-11 14:05:22 Louis

怎么样让接下来的事情在寻求感谢那些帮助了很多 –

另外我想custome inspect_command运行此sysfile.py –

调用SYS_INFO（）：这是另一个PY文件，得到的结果整合SYS_INFO（）功能纳入他们自己的检查命令 –

芹菜从蟒蛇检查任务

回答

相关问题