0

我做了类似于芹菜的设置

SCHEDULE_SETTINGS = {
    'x': {
        'task': 'app.tasks.x',
        'schedule': crontab(minute='*/1'),
        'kwargs': dict(xx='some')
    },
    'y': {
        'task': 'app.tasks.y',
        'schedule': crontab(minute='*/1'),
    },
}

功能

def x(xx):
    print('func x', xx)

def y():
    print('func y')

启动任务

from importlib import import_module
from app.tasks.config import huey


def tasks_bootloader():
    for name, settings in SCHEDULE_SETTINGS.items():
        task = settings['task']
        schedule = settings['schedule']
        kwargs = settings.get('kwargs')

        full_path = task.split('.')
        func_name = full_path[-1]
        module_path = '.'.join(full_path[:-1])
        module = import_module(module_path)

        func = getattr(module, func_name)

        def wrapper():
            if kwargs:
                func(**kwargs)
            else:
                func()

        huey.periodic_task(schedule, name=name)(wrapper)

结果：

[2020-12-03 23:42:07,572] INFO:huey.consumer.Scheduler:Scheduler:Enqueueing periodic task app.x: ae0628de-3df7-429b-a87f-35894db214ce.
INFO:huey.consumer.Scheduler:Enqueueing periodic task app.x: ae0628de-3df7-429b-a87f-35894db214ce.
[2020-12-03 23:42:07,574] INFO:huey.consumer.Scheduler:Scheduler:Enqueueing periodic task app.y: 549b6338-b352-494b-93ac-c106af05177d.
INFO:huey.consumer.Scheduler:Enqueueing periodic task app.y: 549b6338-b352-494b-93ac-c106af05177d.
[2020-12-03 23:42:07,582] INFO:huey:Worker-1:Executing app.x: ae0628de-3df7-429b-a87f-35894db214ce
INFO:huey:Executing app.x: ae0628de-3df7-429b-a87f-35894db214ce

func y

[2020-12-03 23:42:07,585] INFO:huey:Worker-1:app.x: ae0628de-3df7-429b-a87f-35894db214ce executed in 0.003s
INFO:huey:app.x: ae0628de-3df7-429b-a87f-35894db214ce executed in 0.003s
[2020-12-03 23:42:07,586] INFO:huey:Worker-1:Executing app.y: 549b6338-b352-494b-93ac-c106af05177d
INFO:huey:Executing app.y: 549b6338-b352-494b-93ac-c106af05177d

func y

一切正常，但只完成最后一项任务。

我需要这种启动方式。在生产中，在包装器中，任务将在线程中启动。

有人可以有任何想法吗？

4

1 回答 1

0

我做的。

我们每次都需要一个新的函数实例——__call__ 方法

通过参数传递在huey上实现了多次启动任务。设置类似于芹菜。

class LaunchingWrapper(object):
    def __init__(self, func, kwargs):
        self.func = func
        self.kwargs = kwargs

    def __call__(self):
        if self.kwargs:
            self.func(**self.kwargs)
        else:
            self.func()


def tasks_bootloader():
    for name, settings in SCHEDULE_SETTINGS.items():
        task = settings['task']
        schedule = settings['schedule']
        kwargs = settings.get('kwargs')

        full_path = task.split('.')
        func_name = full_path[-1]
        module_path = '.'.join(full_path[:-1])
        module = import_module(module_path)

        func = getattr(module, func_name)

        lw = LaunchingWrapper(func, kwargs)

        huey.periodic_task(schedule, name=name)(lw)
于 2020-12-04T15:46:32.833 回答