为什么 celery 重试，但我的工作没有失败？

Question

我有一个 celery 工作来运行 MySQL 数据库，但是，它总是得到Lock Wait Timeout。在深入研究数据库查询后，我意识到 celery 在 1800 秒后触发了另一项工作，并遇到了我的数据库问题。我不知道为什么——我的工作还没有失败！

@celery.task(bind=True, acks_late=True)
def etl_pipeline(dev=dev, test=test):

我可以告诉 MySQL 再次得到相同的查询，可能是 Celery 触发了相同的工作。为什么我在这里重试，默认重试是 180 秒（3 分钟）。

这是官方文档：

default_retry_delay = 180

重试任务之前的默认时间（以秒为单位）。默认 3 分钟。

但我的情况是 1800 秒。

另外，我的经纪人收到了一些其他警告，我不确定这是否相关：

AMQP 结果后端计划在 4.0 版中弃用并在 v5.0 版中移除。请使用 RPC 后端或持久化后端。

配置 RabbitMq

RABBITMQ_SERVER = 'amqp://{}:{}@{}'.format(
    os.getenv('RABBITMQ_USER'),
    os.getenv('RABBITMQ_PASS'),
    os.getenv('RABBITMQ_HOST')
)
broker_url = '{}/{}'.format(
    RABBITMQ_SERVER,
    os.getenv('RABBITMQ_VHOST'),
)
backend = 'amqp'

我该如何解决这个问题？谢谢！

芹菜：4.2.0

I am using job = chain(single_job), but i only have one single_job job() starting the job.

mysql> show processlist;
+-------+------+---------------+------------------+---------+------+-----------+
| Id    | User | Host          | db               | Command | Time | State     |
+-------+------+---------------+------------------+---------+------+-----------+
| 97189 | clp  | 172.11.17.202 | bain_ai_database | Query   |    0 | init      |
| 97488 | clp  | 172.11.11.252 | bain_ai_database | Query   | 1505 | executing |
| 97489 | clp  | 172.11.11.252 | bain_ai_database | Sleep   | 1851 |           |
| 97543 | clp  | 172.21.6.242  | bain_ai_database | Query   |   51 | updating  |
| 97544 | clp  | 172.21.6.242  | bain_ai_database | Sleep   |   51 |           |
+-------+------+---------------+------------------+---------+------+-----------+

Answer 1

根据您执行 sql 查询的方式，我会尝试以下方法。 (1) 既然你有bind=True，那么任务应该是你函数的第一个参数。 celery 中的约定是调用第一个参数self。 (2) 您想尝试捕获正在发生的数据库级异常并忽略它。

from celery.utils.log import get_task_logger

log = get_task_logger(__name__)


@celery.task(bind=True, acks_late=True)
def etl_pipeline(self, dev=dev, test=test):
    try:
        # try querying the database here using sqlalchemy or mysqlconnect??
    except Exception as ex:
        # for now, log the exception and type so that you can drill down into what is happening
        log.info('[etl_pipeline]  exception of type %s.%s: %s', ex.__class__.__module__, ex.__class__.__name__, ex)
        raise       

您将从日志记录中获得的调试应该帮助您确定您在客户端遇到的错误。