Django Celery的错误处理

from celery import shared_task
from celery.utils.log import get_task_logger

logger = get_task_logger(__name__)

@shared_task(bind=True, autoretry_for=(Exception,), retry_backoff=True)
def my_task(self, arg1, arg2):
    try:
        # some code here
    except Exception as e:
        logger.warning('Task failed, retrying ({0}/{1})...'.format(self.request.retries, self.max_retries))
        raise self.retry(exc=e, countdown=60)  # Retry after 60 seconds

from celery import shared_task
from celery.signals import task_failure
from celery.utils.log import get_task_logger

logger = get_task_logger(__name__)

@shared_task
def my_task(arg1, arg2):
    try:
        # some code here
    except Exception as e:
        logger.error('Task failed', exc_info=True)
        raise Exception('Task failed')

def my_failure_handler(sender, task_id, exception, args, kwargs, **kw):
    logger.error('Task {0} failed: {1!r}'.format(task_id, exception), exc_info=True)

task_failure.connect(my_failure_handler)

from celery.result import AsyncResult

result = my_task.delay(arg1, arg2)

if result.successful():
    print('Task completed successfully with result: {0}'.format(result.result))
else:
    print('Task failed with error: {0}'.format(result.traceback))

job_status = AsyncResult(result.task_id).state
print('Task status: {0}'.format(job_status))