摘要：是一個(gè)基于的分布式調(diào)度系統(tǒng)，文檔在這最近有個(gè)需求想要?jiǎng)討B(tài)的添加任務(wù)而不用重啟服務(wù)找了一圈沒找到什么好辦法也有可能是文檔沒看仔細(xì)，所以只能自己實(shí)現(xiàn)囉為動(dòng)態(tài)添加任務(wù)，首先我想到的是傳遞一個(gè)函數(shù)進(jìn)去，讓某個(gè)特定任務(wù)去執(zhí)行這個(gè)傳遞過去的函數(shù)，就像這

celery是一個(gè)基于Python的分布式調(diào)度系統(tǒng)，文檔在這 ,最近有個(gè)需求,想要?jiǎng)討B(tài)的添加任務(wù)而不用重啟celery服務(wù),找了一圈沒找到什么好辦法(也有可能是文檔沒看仔細(xì))，所以只能自己實(shí)現(xiàn)囉

為celery動(dòng)態(tài)添加任務(wù)，首先我想到的是傳遞一個(gè)函數(shù)進(jìn)去，讓某個(gè)特定任務(wù)去執(zhí)行這個(gè)傳遞過去的函數(shù)，就像這樣

@app.task
def execute(func, *args, **kwargs):
    return func(*args, **kwargs)

很可惜，會(huì)出現(xiàn)這樣的錯(cuò)誤

kombu.exceptions.EncodeError: Object of type "function" is not JSON serializable

換一種序列化方式

@app.task(serializer="pickle")
def execute(func, *args, **kwargs):
    return func(*args, **kwargs)

結(jié)果又出現(xiàn)一大串錯(cuò)誤信息

ERROR/MainProcess] Pool callback raised exception: ContentDisallowed("Refusing to deserialize untrusted content of type pickle (application/x-python-serialize)",)
Traceback (most recent call last):
  File "/home/jl/.virtualenvs/test/lib/python3.6/site-packages/kombu/utils/objects.py", line 42, in __get__
    return obj.__dict__[self.__name__]
KeyError: "chord"

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/home/jl/.virtualenvs/test/lib/python3.6/site-packages/kombu/utils/objects.py", line 42, in __get__
    return obj.__dict__[self.__name__]
KeyError: "_payload"

換一種思路

func = import_string(func)

不知道這樣是否可以，結(jié)果測試: No

哎，流年不利.

最后一直測試，一直測試，終于找到了一種辦法,直接上代碼

from importlib import import_module, reload

app.conf.CELERY_IMPORTS = ["task", "task.all_task"]

def import_string(import_name):
    import_name = str(import_name).replace(":", ".")
    modules = import_name.split(".")
    mod = import_module(modules[0])
    for comp in modules[1:]:
        if not hasattr(mod, comp):
            reload(mod)
        mod = getattr(mod, comp)
    return mod

@app.task
def execute(func, *args, **kwargs):
    func = import_string(func)
    return func(*args, **kwargs)

項(xiàng)目結(jié)構(gòu)是這樣的

├── celery_app.py  
├── config.py  
├── task  
│?? ├── all_task.py  
│?? ├── __init__.py

注意: 任務(wù)必須大于等于兩層目錄

以后每次添加任務(wù)都可以先添加到all_task.py里，調(diào)用時(shí)不用再重啟celery服務(wù)

# task/all_task.py

def ee(c, d):
    return c, d, "你好"

# example
from celery_app import execute

execute.delay("task.all_task.ee", 2, 444)

ok，另外發(fā)現(xiàn)celery也支持任務(wù)定時(shí)調(diào)用,就像這樣

execute.apply_async(args=["task.all_task.aa"], eta=datetime(2017, 7, 9, 8, 12, 0))

簡單實(shí)現(xiàn)一個(gè)任務(wù)重復(fù)調(diào)用的功能

@app.task
def interval(func, seconds, args=(), task_id=None):
    next_run_time = current_time() + timedelta(seconds=seconds)
    kwargs = dict(args=(func, seconds, args), eta=next_run_time)
    if task_id is not None:
        kwargs.update(task_id=task_id)
    interval.apply_async(**kwargs)
    func = import_string(func)
    return func(*args)

大概意思就是先計(jì)算下次運(yùn)行的時(shí)間,然后把任務(wù)添加到celery隊(duì)列里,這里有個(gè)task_id有些問題,因?yàn)榧僭O(shè)添加了每隔3s執(zhí)行一個(gè)任務(wù),
它的task_id默認(rèn)會(huì)使用uuid生成，如果想要再移除這個(gè)任務(wù)就不太方便，自定task_id可能會(huì)好一些，另外也許需要判斷task_id是否存在

AsyncResult(task_id).state

ok,再獻(xiàn)上一個(gè)好用的函數(shù)

from inspect import getmembers, isfunction

def get_tasks(module="task"):
    return [{
        "name": "task:{}".format(f[1].__name__),
        "doc": f[1].__doc__,
    } for f in getmembers(import_module(module), isfunction)]

就這樣.