celery 配置
实例化方式
import celery
import time
backend='redis://127.0.0.1:6379/1'
broker='redis://127.0.0.1:6379/2'
app=celery.Celery('test',backend=backend,broker=broker)
配置文件方式
app = celery.Celery('test')
app.config_from_object('django.conf:settings')
或者
app.config_from_object('django.conf:settings', namespace='CELERY')
配置追加方式
加载配置myapp.config
app.conf.update(
CELERY_RESULT_BACKEND='rpc://',
)
获取当前celery实例
from celery import current_app
#: The Django-Celery app instance.
app = current_app._get_current_object()
在当前实例下查询task
from celery.result import AsyncResult
from celery import current_app
#: The Django-Celery app instance.
app = current_app._get_current_object()
# 查询指定任务ID的结果
async_result = AsyncResult('e73bf051-fee7-4172-9d65-5c4588c6aa8b', app=app)
# 上一条命令也可以这样写
async_result = app.AsyncResult('e73bf051-fee7-4172-9d65-5c4588c6aa8b')
# 打印任务状态和结果
print(f"Task status: {async_result.status}")
print(f"Task result: {async_result.result}")
在当前实例下查询实例相关信息
# 获取当前 Celery 实例的名称
app_name = app.main
# 获取后端（结果存储）的 URL
backend_url = app.conf.result_backend
# 获取 broker（消息代理）的 URL
broker_url = app.conf.broker_url
# 打印这些信息
print(f"App Name: {app_name}")
print(f"Backend URL: {backend_url}")
print(f"Broker URL: {broker_url}")