在 Celery 任务中使用多处理并发机制

Question

我正在尝试与只能接受单个 TCP 连接（内存限制）的设备进行交互，因此仅为每个工作线程启动一个连接不是一个选项，因为它在正常的客户端-服务器情况下，例如数据库连接。

我尝试使用线程间可全局访问的 Multiprocessing Manager dict，格式如下：

clients{(address, port): (connection_obj, multiprocessing.Manager.RLock)}

还有这样一个任务：

from celery import shared_task
from .celery import manager, clients

@shared_task
def send_command(controller, commandname, args):
    """Send a command to the controller."""
    # Create client connection if one does not exist.
    conn = None
    addr, port = controller
    if controller not in clients:
        conn = Client(addr, port)
        conn.connect()
        lock = manager.RLock()
        clients[controller] = (conn, lock,)
        print("New controller connection to %s:%s" % (addr, port,))
    else:
        conn, lock = clients[controller]

    try:
        f = getattr(conn, commandname) # See if connection.commandname() exists.
    except Exception:
        raise Exception("command: %s not known." % (commandname))

    with lock:
        res = f(*args)
        return res

但是任务会因序列化错误而失败，例如：

_pickle.PicklingError: Can't pickle <class '_thread.lock'>: attribute lookup lock on _thread failed

即使我没有使用不可序列化的值调用任务，并且任务也没有尝试返回不可序列化的值，Celery 似乎痴迷于尝试序列化这个全局对象？

我错过了什么？您将如何使 Celery 任务中使用的客户端设备连接成为线程安全且线程之间可访问的？示例代码？

Answer 1

如何使用 Redis 实现分布式锁管理器？ Redis python 客户端具有内置的锁定功能。另外，请参阅 redis.io 上的 this doc。即使您使用的是 RabbitMQ 或其他代理，Redis 也非常轻量级。

例如，作为装饰者：

from functools import wraps

def device_lock(block=True):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            return_value = None
            have_lock = False
            lock = redisconn.lock('locks.device', timeout=2, sleep=0.01)
            try:
                have_lock = lock.acquire(blocking=block)
                if have_lock:
                    return_value = func(*args, **kwargs)
            finally:
                if have_lock:
                    lock.release()
            return return_value
        return wrapper
    return decorator

@shared_task
@device_lock
def send_command(controller, commandname, args):
    """Send a command to the controller."""
    ...

您还可以使用 Celery 任务食谱中的 this approach：

from celery import task
from celery.utils.log import get_task_logger
from django.core.cache import cache
from hashlib import md5
from djangofeeds.models import Feed

logger = get_task_logger(__name__)

LOCK_EXPIRE = 60 * 5 # Lock expires in 5 minutes

@task(bind=True)
def import_feed(self, feed_url):
    # The cache key consists of the task name and the MD5 digest
    # of the feed URL.
    feed_url_hexdigest = md5(feed_url).hexdigest()
    lock_id = '{0}-lock-{1}'.format(self.name, feed_url_hexdigest)

    # cache.add fails if the key already exists
    acquire_lock = lambda: cache.add(lock_id, 'true', LOCK_EXPIRE)
    # memcache delete is very slow, but we have to use it to take
    # advantage of using add() for atomic locking
    release_lock = lambda: cache.delete(lock_id)

    logger.debug('Importing feed: %s', feed_url)
    if acquire_lock():
        try:
            feed = Feed.objects.import_feed(feed_url)
        finally:
            release_lock()
        return feed.url

    logger.debug(
        'Feed %s is already being imported by another worker', feed_url)

Answer 2

 ...
self._send_bytes(ForkingPickler.dumps(obj))
 File "/usr/lib64/python3.4/multiprocessing/reduction.py", line 50, in dumps
cls(buf, protocol).dump(obj)
_pickle.PicklingError: Can't pickle <class '_thread.lock'>: attribute lookup lock on _thread failed

浏览互联网后，我意识到我可能错过了追溯中的重要内容。在仔细查看回溯后，我意识到不是 Celery 试图腌制连接对象，而是 Multiprocessing.reduction。减少用于在一侧序列化并在另一侧重构。

我有一些替代方法来解决这个问题 - 但是它们都没有真正做到我最初想要的，即只是借用客户端库连接对象并使用它，而这对于 Multiprocessing 和 prefork 是不可能的。

Answer 3

您是否尝试过使用 gevent 或 eventlet celery worker 来代替进程和线程？在这种情况下，您将能够使用 global var 或 threading.local() 来共享连接对象。