Python-RLocks über mehrere unabhängige Prozesse hinweg verwenden

Question

Ich arbeite an einem Django-Projekt, das Sellerie verwendet, um einige langfristige Aufgaben zu planen. Sowohl Django als auch Sellery laufen in völlig unabhängigen Prozessen und benötigen eine Möglichkeit, den Zugriff auf die Datenbank zu koordinieren. Ich würde gerne Pythons multiprocessing.RLock-Klasse (oder gleichwertig) verwenden, da ich die Sperre wiedereinwechseln muss.

Meine Frage ist, wie stelle ich den Zugriff auf den RLock für die separaten Prozesse zur Verfügung?

Die zwei besten Lösungen, die ich gefunden habe (posix_ipc module und fcntl), sind auf Unix-basierte Systeme beschränkt, und wir möchten uns darauf beschränken, uns darauf zu beschränken.

Gibt es eine plattformübergreifende Möglichkeit, die Sperren zwischen Prozessen zu teilen, ohne einen gemeinsamen Vorfahrenprozess zu haben?

Answer 1

Ich endete mit RabbitMQ als eine Möglichkeit, verteilte Sperren zu erstellen. Details dazu finden Sie auf RabbitMQ's Blog: https://www.rabbitmq.com/blog/2014/02/19/distributed-semaphores-with-rabbitmq/.

Kurz gesagt, Sie erstellen eine RabbitMQ-Warteschlange für die Sperre und senden eine einzelne Nachricht an sie. Um die Sperre zu erhalten, führen Sie eine basic_get (nicht blockierend) oder basic_consume (blockierend) in der Warteschlange aus. Dadurch wird die Nachricht aus der Warteschlange entfernt und verhindert, dass andere Threads die Sperre erhalten. Sobald Ihre Arbeit beendet ist, wird RabbitMQ durch das Senden einer negativen Bestätigung aufgefordert, die Nachricht erneut zu senden, damit der nächste Thread fortgesetzt werden kann.

Leider sind Reentrant-Sperren nicht zulässig.

Der oben erwähnte Link gibt Java-Code für die Vorgehensweise. Herauszufinden, wie man das in Python/Pika übersetzt, war nervig genug, dass ich dachte, ich sollte hier einen Beispielcode posten.

Um die Sperre zu produzieren:

import pika 

with pika.BlockingConnection(pika.ConnectionParameters('localhost')) as connection: 
    channel = connection.channel() 
    channel.queue_declare(queue="LockQueue") 
    channel.basic_publish(exchange='', routing_key='LockQueue', body='Lock') 
    channel.close() 

das Schloss Erwerb:

import pika 
import time 

def callback(ch, method, properties, body): 
    print("Got lock") 

    for i in range(5, 0, -1): 
     print("Tick {}".format(i)) 
     time.sleep(1) 

    print("Releasing lock") 
    ch.basic_nack(delivery_tag=method.delivery_tag) 
    ch.close() # Close the channel to continue on with normal processing. Without this, `callback` will continue to request the lock. 

with pika.BlockingConnection(pika.ConnectionParameters('localhost')) as connection: 
    channel = connection.channel() 

    channel.queue_declare(queue='LockQueue') 
    channel.basic_qos(prefetch_count=1) 
    channel.basic_consume(callback, queue='LockQueue') 

    print("Waiting for lock") 
    channel.start_consuming() 
    print("Task completed")