python – 无法导入该任务时运行Celery任务

我有两个服务器：一个运行 django应用程序,另一个运行rabbitmq队列和芹菜工作者.运行队列/工作服务器的服务器上的tasks.py包含如下任务：

@task(queue="reports")
def test_task():
    time.sleep(120)

我的目标是从django视图执行此任务.由于任务的代码在与我想调用任务的django视图不同的服务器上,我正在尝试使用以下代码将任务从django发送到工作机器.

send_task("tasks.test_task",task_id=task_id,args=[],kwargs={},publisher=publisher,queue=queue)

我发现这个方法here,但到目前为止测试它还没有用.

我正在celery工作服务器上的celery worker日志文件中使用tail -F进行测试,然后在浏览器中导航到包含send_task的视图的url.我正在寻找在尾部输出中显示为“已接收”的任务,但事实并非如此.

芹菜工作者的日志级别是DEBUG,日志文件显示任务是使用正确的名称注册的,而django应用程序的settings.py包含rabbitmq服务器的正确IP和凭据.在尝试不同的方法时,我偶尔会在celery日志文件中看到一条错误消息,当我将传递给send_task的字符串更改为不是有效任务的字符串(即send_task(‘asdf’))时.这导致日志文件中出现UnregisteredError.但是,这种情况有时只会发生,到目前为止,在测试设置和调用的不同组合时,我还没有找到可靠地复制行为的方法.

此外,这是django项目上settings.py的相关部分(删除了实际值)：

CELERY_RESULT_BACKEND = 'amqp'
broKER_HOST = 'the.correct.IP.address'
broKER_USER = 'the_correct_user'
broKER_PASSWORD = 'the_correct_pass'
broKER_VHOST = 'the_correct_vhost'
broKER_PORT = 5672

我google了一下,在send_task上找不到太多东西.关于我可能做错的任何想法？

解决方法

已解决,结果是发送者关键字arg我传递给send_task无效并抛出错误.没有看到错误,因为我是AJAX请求页面而不是直接导航到它.关于这种情况的其他一切都是正确的.我还删除了不必要的关键字args和传递给send_task的args.

send_task("tasks.test_task",queue=queue)