Huey如何调用任务

Question

我在这里有这个代码

# 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将引发未定义的错误(。

提前谢谢。

Answer 1

周期性任务由使用者调用(你正在运行一个任务，对吗？(我相信设计是这样的，你不应该向这些任务传递参数-你怎么会向使用者传递参数？我相信你能想出一个设计方案，但对我来说，这真的没有意义。来自文档：

因为周期性任务的调用独立于任何用户交互，他们不接受任何争论。

类似地，周期性任务的返回值也会被丢弃而不是放入结果存储中。这是因为没有应用程序获取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 ...>