Mehrere Worker-Threads, die an der gleichen Datenbank arbeiten - wie funktioniert es richtig?

Question

Ich habe eine Datenbank mit einer Liste von Zeilen, die bearbeitet werden müssen. Es sieht etwa so aus:

id  remaining  delivered locked 
============================================ 
1  10    24   f 
2  6    0   f 
3  0    14   f 

Ich verwende DataMapper Ruby, aber wirklich ich denke, das eine allgemeine Programmierung Frage, die nicht spezifisch für die genaue Umsetzung ist ich verwende ...

ich bin ein Bündel von worker-Threads erstellen, die so etwas wie dieser (pseudo-Rubin-Code) zu tun:

while true do 
    t = any_row_in_database_where_remaining_greater_than_zero_and_unlocked 
    t.lock # update database to set locked = true 
    t.do_some_stuff 
    t.delivered += 1 
    t.remaining -= 1 
    t.unlock 
end 

natürlich ist das Problem, konkurrieren diese Fäden miteinander und das ganze nicht wirklich fädelt sicher. Die erste Zeile in der while-Schleife kann die gleiche Zeile in mehreren Threads leicht herausziehen, bevor sie gesperrt werden können.

Ich muss sicherstellen, dass ein Thread nur in einer Zeile gleichzeitig arbeitet.

Was ist der beste Weg, dies zu tun?

Answer 1

Der Schlüsselschritt ist, wenn Sie eine entsperrte Zeile aus der Datenbank auswählen und als gesperrt markieren. Wenn Sie das sicher tun können, wird alles andere gut.

2 Möglichkeiten, die ich davon kenne, können diese sichere pessimistische und optimistische Sperrung machen. Sie beide verlassen sich auf Ihre Datenbank als ultimativer Garant, wenn es um Nebenläufigkeit geht.

pessimistisches Sperren

Pessimistische Sperrmittel eine Sperre im Voraus zu erwerben, wenn Sie die Zeilen auswählen, mit denen Sie arbeiten wollen, so dass niemand sonst kann sie lesen. So etwas wie

SELECT * from some_table WHERE ... FOR UPDATE 

funktioniert mit MySQL und Postgres (und möglicherweise andere) und andere Verbindung zur Datenbank verhindern, dass das Lesen der Zeilen an Sie zurückgegeben (wie granulare, die ist sperren am Motor hängt verwendet, Indizes usw. - Überprüfen Sie die Dokumentation Ihrer Datenbank. Es heißt pessimistisch, weil Sie annehmen, dass ein Nebenläufigkeitsproblem auftritt und die Sperre präventiv erfasst. Es bedeutet, dass Sie die Kosten für das Sperren selbst dann tragen, wenn dies nicht notwendig ist, und Ihren Nebenläufigkeit abhängig von der Granularität des Schlosses, das Sie haben, reduzieren können.

Optimistische Sperren

Optimistisches Sperre bezieht sich auf eine Technik, wo man nicht die Last einer pessimistischen Sperre will, weil die meiste Zeit wird es keine gleichzeitigen Aktualisierungen (wenn Sie die Zeile aktualisieren, um die gesperrten Flags Sobald Sie die Zeile gelesen haben, ist das Fenster relativ klein). AFAIK dies funktioniert nur beim Aktualisieren von jeweils einer Zeile

Fügen Sie zuerst eine Ganzzahlspalte lock_version in die Tabelle ein. Wenn Sie die Tabelle aktualisieren, erhöhen Sie lock_version um 1 neben den anderen Aktualisierungen, die Sie vornehmen. Angenommen, der aktuelle lock_version ist 3. Wenn Sie aktualisieren, ändern Sie die Update-Abfrage auf

update some_table set ... where id=12345 and lock_version = 3 

und überprüfen Sie die Anzahl der Zeilen aktualisiert (der db-Treiber gibt diese). Wenn das 1 Reihe aktualisiert, dann weißt du, dass alles in Ordnung war. Wenn diese Option 0 Zeilen aktualisiert, wurde entweder die gewünschte Zeile gelöscht oder ihre Sperrversion wurde geändert. Sie kehren daher zu Schritt 1 in Ihrem Prozess zurück und suchen nach einer neuen Zeile, an der Sie arbeiten möchten.

Ich bin kein Datamapper-Benutzer, also weiß ich nicht, ob es/plugins dafür Unterstützung für diese Ansätze bietet. Active Record unterstützt beide, so dass Sie dort nach Inspiration suchen können, wenn Data Mapper dies nicht tut.

Answer 2

würde ich ein Mutex:

# outside your threads 
worker_updater = Mutex.new 

# inside each thread's updater 
while true 
    worker_updater.synchronize do 
    # your code here 
    end 
    sleep 0.1 # Slow down there, mister! 
end 

Dies garantiert, dass nur ein Thread zu einem Zeitpunkt, den Code in den synchronize eingeben. Ermitteln Sie für eine optimale Leistung, welcher Teil des Codes threadsicher sein muss (die ersten beiden Zeilen?) Und wickeln Sie diesen Teil nur im Mutex um.