如何为芹菜后端连接添加重试？

Question

我正在使用 celery 5.0.1 并使用 CELERY_BACKEND_URL 作为 redis://:password@redisinstance1:6379/0。它工作正常，但是当有一个 Redis 实例连接松动时，它会错误地中断任务。

Exception: Error while reading from socket: (104, 'Connection reset by peer')
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/dist-packages/redis/connection.py", line 198, in _read_from_socket
    data = recv(self._sock, socket_read_size)
  File "/usr/local/lib/python3.7/dist-packages/redis/_compat.py", line 72, in recv
    return sock.recv(*args, **kwargs)
ConnectionResetError: [Errno 104] Connection reset by peer

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/local/lib/python3.7/dist-packages/celery/app/trace.py", line 477, in trace_task
    uuid, retval, task_request, publish_result,
  File "/usr/local/lib/python3.7/dist-packages/celery/backends/base.py", line 154, in mark_as_done
    self.store_result(task_id, result, state, request=request)
  File "/usr/local/lib/python3.7/dist-packages/celery/backends/base.py", line 439, in store_result
    request=request, **kwargs)
  File "/usr/local/lib/python3.7/dist-packages/celery/backends/base.py", line 855, in _store_result
    current_meta = self._get_task_meta_for(task_id)
  File "/usr/local/lib/python3.7/dist-packages/celery/backends/base.py", line 873, in _get_task_meta_for
    meta = self.get(self.get_key_for_task(task_id))
  File "/usr/local/lib/python3.7/dist-packages/celery/backends/redis.py", line 346, in get
    return self.client.get(key)
  File "/usr/local/lib/python3.7/dist-packages/redis/client.py", line 1606, in get
    return self.execute_command('GET', name)
  File "/usr/local/lib/python3.7/dist-packages/redis/client.py", line 901, in execute_command
    return self.parse_response(conn, command_name, **options)
  File "/usr/local/lib/python3.7/dist-packages/redis/client.py", line 915, in parse_response
    response = connection.read_response()
  File "/usr/local/lib/python3.7/dist-packages/redis/connection.py", line 739, in read_response
    response = self._parser.read_response()
  File "/usr/local/lib/python3.7/dist-packages/redis/connection.py", line 324, in read_response
    raw = self._buffer.readline()
  File "/usr/local/lib/python3.7/dist-packages/redis/connection.py", line 256, in readline
    self._read_from_socket()
  File "/usr/local/lib/python3.7/dist-packages/redis/connection.py", line 223, in _read_from_socket
    (ex.args,))
redis.exceptions.ConnectionError: Error while reading from socket: (104, 'Connection reset by peer')


Celery worker: None
Celery task id: 244b56af-7c96-56cf-a01a-9256cfd98ade
Celery retry attempt: 0
Task args: []
Task kwargs: {'address': 'ipadd', 'uid': 'uid', 'hexID': 'hexID', 'taskID': '244b56af-7c96-56cf-a01a-9256cfd98ade'}

当我运行第二个任务时，它工作正常，短时间内连接出现一些故障。

我可以设置一些东西吗，当 celery 尝试将结果更新到 Redis 时，如果 returns 出错，它会在 2-5 秒后重试？

我知道如何在任务中设置重试，但这并不会导致任务失败。我的任务工作正常，它 returns 数据，但 celery 在更新到后端时失去连接。

Answer 1

要处理 connection timeouts，您可以在 Celery 配置中包含以下内容：

app.conf.broker_transport_options = {
    'retry_policy': {
       'timeout': 5.0
    }
}
app.conf.result_backend_transport_options = {
    'retry_policy': {
       'timeout': 5.0
    }
}

您可能要考虑在您的配置中加入其他一些 Redis backend settings，例如 redis_retry_on_timeout。

如何为芹菜后端连接添加重试？

How to add retry for celery backend connection?

python

redis

celery