Использование механизмов многопроцессорного параллелизма в задачах Celery

Как насчет реализации диспетчера распределенных блокировок с помощью Redis? Клиент Python Redis имеет встроенную функцию блокировки. Также см. этот документ на redis.io. Даже если вы используете 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:

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)

 ...
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. Редукция используется для сериализации с одной стороны и восстановления с другой.

У меня есть несколько альтернативных подходов к решению этой проблемы, однако ни один из них на самом деле не делает то, что я изначально хотел, а именно просто заимствовать объект подключения клиентской библиотеки и использовать его, что просто невозможно с многопроцессорностью и предварительным ответвлением.

Вы пытались использовать gevent или eventlet celery worker вместо процессов и потоков? В этом случае вы сможете использовать глобальную переменную или threading.local() для совместного использования объекта соединения.