docker - RuntimeError : concurrent poll() invocation using celery

在从其他容器接收restAPI的Docker容器上运行Celery时，出现RuntimeError:并发poll()调用。

有人遇到过类似的错误吗？

我附加了追溯。

Traceback (most recent call last):
  File "/usr/lib/python3.5/threading.py", line 914, in _bootstrap_inner
    self.run()
  File "/usr/lib/python3.5/threading.py", line 862, in run
    self._target(*self._args, **self._kwargs)
  File "/opt/www/api/api/training_call.py", line 187, in start_process
    result_state.get(on_message=self._on_raw_message, propagate=False)
  File "/usr/local/lib/python3.5/dist-packages/celery/result.py", line 226, in get
    on_message=on_message,
  File "/usr/local/lib/python3.5/dist-packages/celery/backends/asynchronous.py", line 188, in wait_for_pending
    for _ in self._wait_for_pending(result, **kwargs):
  File "/usr/local/lib/python3.5/dist-packages/celery/backends/asynchronous.py", line 255, in _wait_for_pending
    on_interval=on_interval):
  File "/usr/local/lib/python3.5/dist-packages/celery/backends/asynchronous.py", line 56, in drain_events_until
    yield self.wait_for(p, wait, timeout=1)
  File "/usr/local/lib/python3.5/dist-packages/celery/backends/asynchronous.py", line 65, in wait_for
    wait(timeout=timeout)
  File "/usr/local/lib/python3.5/dist-packages/celery/backends/redis.py", line 127, in drain_events
    message = self._pubsub.get_message(timeout=timeout)
  File "/usr/local/lib/python3.5/dist-packages/redis/client.py", line 3135, in get_message
    response = self.parse_response(block=False, timeout=timeout)
  File "/usr/local/lib/python3.5/dist-packages/redis/client.py", line 3034, in parse_response
    if not block and not connection.can_read(timeout=timeout):
  File "/usr/local/lib/python3.5/dist-packages/redis/connection.py", line 628, in can_read
    return self._parser.can_read() or self._selector.can_read(timeout)
  File "/usr/local/lib/python3.5/dist-packages/redis/selector.py", line 28, in can_read
    return self.check_can_read(timeout)
  File "/usr/local/lib/python3.5/dist-packages/redis/selector.py", line 156, in check_can_read
    events = self.read_poller.poll(timeout)
RuntimeError: concurrent poll() invocation

最佳答案

代理连接不是线程安全的，因此您需要在应用程序代码中处理线程安全。
@Laizer提到了ticket，在python核心库中引入了此错误

一种方法是将所有阻塞的调用包装到共享Lock中，直到任务完成为止:

import celery
import threading

@celery.shared_task
def debug_task(self):
    print('Hello, world')

def boom(nb_tasks):
    """ not thread safe - raises RuntimeError during concurrent executions """
    tasks = celery.group([debug_task.s() for _ in range(nb_tasks)])
    pool = tasks.apply_async()
    pool.join() # raised from here

CELERY_POLL_LOCK = threading.Lock()

def safe(nb_tasks):
    tasks = celery.group([debug_task.s() for _ in range(nb_tasks)])
    pool = tasks.apply_async()
    with CELERY_POLL_LOCK:  # prevents concurrent calls to poll()
        pool.join()

def main(nb_threads, nb_tasks_per_thread):
    for func in (safe, boom):
        threads = [threading.Thread(target=func, args=(nb_tasks_per_thread, )) for _ in range(nb_threads)]

        for a_thread in threads:
            a_thread.start()

        for a_thread in threads:
            a_thread.join()

main(10, 100)

这是一种幼稚的方法，适合我，因为我并不期望太多的并发性，而且所有任务都相对较快(〜10s)。
如果您使用不同的“配置文件”，则可能需要进行更复杂的操作(例如，一个轮询任务会定期轮询所有未决的组/任务)。

关于docker - RuntimeError : concurrent poll() invocation using celery，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/55578621/