2

我这里有这段代码

# it's not every five mins but let's overlook that for now
@periodic_task(crontab(minute='*/1'))
def every_five_mins():
    # ...

但我找不到 Huey 在哪里调用该函数。我用过 Huey 的唯一其他地方是，settings.py但我仍然只包括

HUEY = {
    'huey_class': 'huey.RedisHuey',
    'name': DATABASES['default']['NAME'], 
    'results': True,  
    'store_none': False, 
    'immediate': False, 
    'utc': True,
    'blocking': True,
    'connection': {
        'host': 'localhost',
        'port': 6379,
        'db': 0,
        'connection_pool': None,
        'read_timeout': 1, 
        'url': None, 
    },
    'consumer': {
        'workers': 1,
        'worker_type': 'thread',
        'initial_delay': 0.1,
        'backoff': 1.15, 
        'max_delay': 10.0, 
        'scheduler_interval': 1, 
        'periodic': True,
        'check_worker_health': True,  
        'health_check_interval': 1, 
    },
}

谁能告诉我任务是如何执行的？我想知道这一点，因为我想将参数传递给every_five_mins()，例如，every_five_mins(argument1=argument1)但我不能在不知道函数在哪里调用的情况下这样做（否则 argument1 会引发未定义的错误）。

提前致谢。

4

1 回答 1

0

定期任务由消费者调用（你正在运行一个，对吗？），我相信设计是这样的，你不打算将参数传递给这些——你甚至如何将参数传递给消费者？我相信你可以想出一个设计，但对我来说这没有任何意义。从文档：

因为周期性任务的调用独立于任何用户交互，所以它们不接受任何参数。

类似地，周期性任务的返回值被丢弃，而不是被放入结果存储中。这是因为应用程序没有明显的方法来获取 Result 句柄来访问给定周期性任务执行的结果。

根据您的需要，您可以通过调用返回一些参数以在任务中使用的函数来实现您想要的：

def get_argument():
    arg = None
    # do stuff
    return arg

# it's not every five mins but let's overlook that for now
@periodic_task(crontab(minute='*/1'))
def every_five_mins():
    argument1 = get_argument()
    # ...

或者，您也可以动态定义周期性任务。复制文档中的示例（注意，这不适用于流程工作人员——请参阅文档链接）：

def dynamic_ptask(message):
    print('dynamically-created periodic task: "%s"' % message)

@huey.task()
def schedule_message(message, cron_minutes, cron_hours='*'):
    # Create a new function that represents the application
    # of the "dynamic_ptask" with the provided message.
    def wrapper():
        dynamic_ptask(message)

    # The schedule that was specified for this task.
    schedule = crontab(cron_minutes, cron_hours)

    # Need to provide a unique name for the task. There are any number of
    # ways you can do this -- based on the arguments, etc. -- but for our
    # example we'll just use the time at which it was declared.
    task_name = 'dynamic_ptask_%s' % int(time.time())

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

假设消费者正在运行，我们现在可以设置任意数量的“动态 ptask”函数实例：

>>> from demo import schedule_message
>>> schedule_message('I run every 5 minutes', '*/5')
<Result: task ...>
>>> schedule_message('I run between 0-15 and 30-45', '0-15,30-45')
<Result: task ...>
于 2021-06-18T20:34:59.317 回答