FIFO-Warteschlangentabelle für mehrere Worker in SQL Server

15

Ich habe versucht, die folgende Stackoverflow-Frage zu beantworten:

Nachdem ich eine etwas naive Antwort veröffentlicht hatte, stellte ich mir vor, ich würde mein Geld dahin stecken, wo mein Mund war, und tatsächlich das von mir vorgeschlagene Szenario testen , um sicherzugehen, dass ich den OP nicht auf wilde Gänsehaut schickte. Nun, es stellte sich heraus, dass es viel schwieriger ist als ich dachte (keine Überraschung für irgendjemanden, da bin ich mir sicher).

Folgendes habe ich versucht und darüber nachgedacht:

  • Zuerst habe ich ein TOP 1 UPDATE mit einem ORDER BY in einer abgeleiteten Tabelle mit versucht ROWLOCK, READPAST. Dies ergab Deadlocks und verarbeitete auch Artikel, die nicht in Ordnung waren. Es muss so nah wie möglich am FIFO sein, außer bei Fehlern, bei denen versucht werden muss, dieselbe Zeile mehr als einmal zu verarbeiten.

  • I dann die gewünschte nächste QueueID in eine Variable versucht Auswahl verschiedener Kombinationen der Verwendung von READPAST, UPDLOCK, HOLDLOCK, und ROWLOCKausschließlich durch diese Sitzung die Zeile für die Aktualisierung zu erhalten. Alle Variationen, die ich ausprobiert habe, litten unter den gleichen Problemen wie zuvor READPASTund beschwerten sich bei bestimmten Kombinationen mit :

    Sie können die READPAST-Sperre nur in den Isolationsstufen READ COMMITTED oder REPEATABLE READ angeben.

    Dies war verwirrend , weil es wurde READ COMMITTED. Ich bin schon früher darauf gestoßen und es ist frustrierend.

  • Seit ich diese Frage schreibe, gibt Remus Rusani eine neue Antwort auf die Frage. Ich lese seinen verlinkten Artikel und sehe, dass er destruktive Lesevorgänge verwendet, da er in seiner Antwort sagte, dass es "nicht realistisch möglich ist, Sperren für die Dauer der Webanrufe beizubehalten". Nachdem ich gelesen habe, was in seinem Artikel über Hotspots und Seiten steht, für die Sperren erforderlich sind, um Aktualisierungen oder Löschvorgänge durchzuführen, befürchte ich, dass sie nicht skalierbar sind und auch nicht funktionieren könnten, wenn ich die richtigen Sperren herausfinden könnte, um das zu tun, wonach ich suche nicht mit massiver Parallelität umgehen.

Im Moment bin ich mir nicht sicher, wohin ich gehen soll. Stimmt es, dass die Aufrechterhaltung von Sperren während der Verarbeitung der Zeile nicht erreicht werden kann (auch wenn keine hohe Geschwindigkeit oder massive Parallelität unterstützt wird)? Was vermisse ich?

In der Hoffnung, dass Menschen, die schlauer sind als ich und erfahrener als ich, helfen können, finden Sie unten das Testskript, das ich verwendet habe. Es wird wieder auf die TOP 1 UPDATE-Methode umgeschaltet, aber ich habe die andere Methode in auskommentiert, falls Sie das auch untersuchen möchten.

Fügen Sie diese in eine separate Sitzung ein, führen Sie Sitzung 1 aus und dann schnell alle anderen. In ca. 50 Sekunden ist der Test beendet. Sehen Sie sich die Nachrichten aus jeder Sitzung an, um zu sehen, welche Arbeit sie verrichtet hat (oder wie sie fehlgeschlagen ist). In der ersten Sitzung wird einmal pro Sekunde ein Rowset mit einem Snapshot angezeigt, in dem die vorhandenen Sperren und die in Bearbeitung befindlichen Warteschlangenelemente aufgeführt sind. Es funktioniert manchmal und manchmal überhaupt nicht.

Session 1

/* Session 1: Setup and control - Run this session first, then immediately run all other sessions */
IF Object_ID('dbo.Queue', 'U') IS NULL
   CREATE TABLE dbo.Queue (
      QueueID int identity(1,1) NOT NULL,
      StatusID int NOT NULL,
      QueuedDate datetime CONSTRAINT DF_Queue_QueuedDate DEFAULT (GetDate()),
      CONSTRAINT PK_Queue PRIMARY KEY CLUSTERED (QueuedDate, QueueID)
   );

IF Object_ID('dbo.QueueHistory', 'U') IS NULL
   CREATE TABLE dbo.QueueHistory (
      HistoryDate datetime NOT NULL,
      QueueID int NOT NULL
   );

IF Object_ID('dbo.LockHistory', 'U') IS NULL
   CREATE TABLE dbo.LockHistory (
      HistoryDate datetime NOT NULL,
      ResourceType varchar(100),
      RequestMode varchar(100),
      RequestStatus varchar(100),
      ResourceDescription varchar(200),
      ResourceAssociatedEntityID varchar(200)
   );

IF Object_ID('dbo.StartTime', 'U') IS NULL
   CREATE TABLE dbo.StartTime (
      StartTime datetime NOT NULL
   );

SET NOCOUNT ON;

IF (SELECT Count(*) FROM dbo.Queue) < 10000 BEGIN
   TRUNCATE TABLE dbo.Queue;

   WITH A (N) AS (SELECT 1 UNION ALL SELECT 1 UNION ALL SELECT 1 UNION ALL SELECT 1),
   B (N) AS (SELECT 1 FROM A Z, A I, A P),
   C (N) AS (SELECT Row_Number() OVER (ORDER BY (SELECT 1)) FROM B O, B W)
   INSERT dbo.Queue (StatusID, QueuedDate)
   SELECT 1, DateAdd(millisecond, C.N * 3, GetDate() - '00:05:00')
   FROM C
   WHERE C.N <= 10000;
END;

TRUNCATE TABLE dbo.StartTime;
INSERT dbo.StartTime SELECT GetDate() + '00:00:15'; -- or however long it takes you to go run the other sessions
GO
TRUNCATE TABLE dbo.QueueHistory;
SET NOCOUNT ON;

DECLARE
   @Time varchar(8),
   @Now datetime;
SELECT @Time = Convert(varchar(8), StartTime, 114)
FROM dbo.StartTime;
WAITFOR TIME @Time;

DECLARE @i int,
@QueueID int;
SET @i = 1;
WHILE @i <= 33 BEGIN
   SET @Now  = GetDate();
   INSERT dbo.QueueHistory
   SELECT
      @Now,
      QueueID
   FROM
      dbo.Queue Q WITH (NOLOCK)
   WHERE
      Q.StatusID <> 1;

   INSERT dbo.LockHistory
   SELECT
      @Now,
      L.resource_type,
      L.request_mode,
      L.request_status,
      L.resource_description,
      L.resource_associated_entity_id
   FROM
      sys.dm_tran_current_transaction T
      INNER JOIN sys.dm_tran_locks L
         ON L.request_owner_id = T.transaction_id;
   WAITFOR DELAY '00:00:01';
   SET @i = @i + 1;
END;

WITH Cols AS (
   SELECT *, Row_Number() OVER (PARTITION BY HistoryDate ORDER BY QueueID) Col
   FROM dbo.QueueHistory
), P AS (
   SELECT *
   FROM
      Cols
      PIVOT (Max(QueueID) FOR Col IN ([1], [2], [3], [4], [5], [6], [7], [8])) P
)
SELECT L.*, P.[1], P.[2], P.[3], P.[4], P.[5], P.[6], P.[7], P.[8]
FROM
   dbo.LockHistory L
   FULL JOIN P
      ON L.HistoryDate = P.HistoryDate

/* Clean up afterward
DROP TABLE dbo.StartTime;
DROP TABLE dbo.LockHistory;
DROP TABLE dbo.QueueHistory;
DROP TABLE dbo.Queue;
*/

Sitzung 2

/* Session 2: Simulate an application instance holding a row locked for a long period, and eventually abandoning it. */
SET TRANSACTION ISOLATION LEVEL READ COMMITTED;
SET NOCOUNT ON;
SET XACT_ABORT ON;

DECLARE
   @QueueID int,
   @Time varchar(8);
SELECT @Time = Convert(varchar(8), StartTime + '0:00:01', 114)
FROM dbo.StartTime;
WAITFOR TIME @Time;
BEGIN TRAN;

--SET @QueueID = (
--   SELECT TOP 1 QueueID
--   FROM dbo.Queue WITH (READPAST, UPDLOCK)
--   WHERE StatusID = 1 -- ready
--   ORDER BY QueuedDate, QueueID
--);

--UPDATE dbo.Queue
--SET StatusID = 2 -- in process
----OUTPUT Inserted.*
--WHERE QueueID = @QueueID;

SET @QueueID = NULL;
UPDATE Q
SET Q.StatusID = 1, @QueueID = Q.QueueID
FROM (
   SELECT TOP 1 *
   FROM dbo.Queue WITH (ROWLOCK, READPAST)
   WHERE StatusID = 1
   ORDER BY QueuedDate, QueueID
) Q

PRINT @QueueID;

WAITFOR DELAY '00:00:20'; -- Release it partway through the test

ROLLBACK TRAN; -- Simulate client disconnecting

Sitzung 3

/* Session 3: Run a near-continuous series of "failed" queue processing. */
SET TRANSACTION ISOLATION LEVEL READ COMMITTED;
SET XACT_ABORT ON;
SET NOCOUNT ON;
DECLARE
   @QueueID int,
   @EndDate datetime,
   @NextDate datetime,
   @Time varchar(8);

SELECT
   @EndDate = StartTime + '0:00:33',
   @Time = Convert(varchar(8), StartTime, 114)
FROM dbo.StartTime;

WAITFOR TIME @Time;

WHILE GetDate() < @EndDate BEGIN
   BEGIN TRAN;

   --SET @QueueID = (
   --   SELECT TOP 1 QueueID
   --   FROM dbo.Queue WITH (READPAST, UPDLOCK)
   --   WHERE StatusID = 1 -- ready
   --   ORDER BY QueuedDate, QueueID
   --);

   --UPDATE dbo.Queue
   --SET StatusID = 2 -- in process
   ----OUTPUT Inserted.*
   --WHERE QueueID = @QueueID;

   SET @QueueID = NULL;
   UPDATE Q
   SET Q.StatusID = 1, @QueueID = Q.QueueID
   FROM (
      SELECT TOP 1 *
      FROM dbo.Queue WITH (ROWLOCK, READPAST)
      WHERE StatusID = 1
      ORDER BY QueuedDate, QueueID
   ) Q

   PRINT @QueueID;

   SET @NextDate = GetDate() + '00:00:00.015';
   WHILE GetDate() < @NextDate SET NOCOUNT ON;
   ROLLBACK TRAN;
END

Sitzung 4 und höher - so viele, wie Sie möchten

/* Session 4: "Process" the queue normally, one every second for 30 seconds. */
SET TRANSACTION ISOLATION LEVEL READ COMMITTED;
SET XACT_ABORT ON;
SET NOCOUNT ON;

DECLARE @Time varchar(8);
SELECT @Time = Convert(varchar(8), StartTime, 114)
FROM dbo.StartTime;
WAITFOR TIME @Time;

DECLARE @i int,
@QueueID int;
SET @i = 1;
WHILE @i <= 30 BEGIN
   BEGIN TRAN;

   --SET @QueueID = (
   --   SELECT TOP 1 QueueID
   --   FROM dbo.Queue WITH (READPAST, UPDLOCK)
   --   WHERE StatusID = 1 -- ready
   --   ORDER BY QueuedDate, QueueID
   --);

   --UPDATE dbo.Queue
   --SET StatusID = 2 -- in process
   --WHERE QueueID = @QueueID;

   SET @QueueID = NULL;
   UPDATE Q
   SET Q.StatusID = 1, @QueueID = Q.QueueID
   FROM (
      SELECT TOP 1 *
      FROM dbo.Queue WITH (ROWLOCK, READPAST)
      WHERE StatusID = 1
      ORDER BY QueuedDate, QueueID
   ) Q

   PRINT @QueueID;
   WAITFOR DELAY '00:00:01'
   SET @i = @i + 1;
   DELETE dbo.Queue
   WHERE QueueID = @QueueID;   
   COMMIT TRAN;
END
ErikE
quelle
2
Die im verlinkten Artikel beschriebenen Warteschlangen können auf Hunderttausende oder weniger Vorgänge pro Sekunde skaliert werden. Die Streitfragen im Zusammenhang mit Hot Spots sind nur in größerem Maßstab relevant. Es sind Schadensbegrenzungsstrategien bekannt, mit denen ein höherer Durchsatz auf High-End-Systemen erzielt werden kann, der Zehntausende pro Sekunde beträgt . Diese Schadensbegrenzungen müssen jedoch sorgfältig evaluiert und unter der Aufsicht von SQLCAT eingesetzt werden .
Remus Rusanu
Ein interessanter Fehler ist, dass mit READPAST, UPDLOCK, ROWLOCKmeinem Skript zum Erfassen von Daten in der QueueHistory-Tabelle nichts getan wird. Ich frage mich, ob das daran liegt, dass die StatusID nicht festgeschrieben ist. Es ist WITH (NOLOCK)so theoretisch sollte funktionieren ... und es hat funktioniert, bevor! Ich bin mir nicht sicher, warum es jetzt nicht funktioniert, aber es ist wahrscheinlich eine weitere Lernerfahrung.
ErikE
Können Sie Ihren Code auf das kleinste Beispiel reduzieren, das Deadlocks und andere Probleme aufweist, die Sie lösen möchten?
Nick Chammas
@ Nick Ich werde versuchen, den Code zu reduzieren. Zu Ihren anderen Kommentaren gibt es eine Identitätsspalte, die Teil des Clustered-Index ist und nach dem Datum sortiert ist. Ich bin durchaus bereit, einen "destruktiven Lesevorgang" (DELETE with OUTPUT) durchzuführen, aber eine der gestellten Anforderungen bestand darin, dass die Zeile bei einem Fehler in einer Anwendungsinstanz automatisch zur Verarbeitung zurückkehren sollte. Meine Frage hier ist also, ob das möglich ist.
ErikE
Probieren Sie die destruktive Lesemethode aus und platzieren Sie aus der Warteschlange genommene Elemente in einer separaten Tabelle, in der sie gegebenenfalls erneut in die Warteschlange gestellt werden. Wenn dies das Problem behebt, können Sie in den reibungslosen Ablauf dieses Re-Enqueue-Prozesses investieren.
Nick Chammas

Antworten:

10

Sie benötigen genau 3 Sperrhinweise

  • READPAST
  • UPDLOCK
  • DOLLE

Ich habe dies bereits auf SO beantwortet: /programming/939831/sql-server-process-queue-race-condition/940001#940001

Wie Remus sagt, ist die Verwendung von Service Broker besser, aber diese Tipps funktionieren

Ihr Fehler bezüglich der Isolationsstufe bedeutet normalerweise, dass Replikation oder NOLOCK beteiligt sind.

gbn
quelle
Die Verwendung der oben angegebenen Hinweise in meinem Skript führt zu Deadlocks und Prozessen außerhalb der angegebenen Reihenfolge. ( UPDATE SET ... FROM (SELECT TOP 1 ... FROM ... ORDER BY ...)) Bedeutet dies, dass mein UPDATE-Muster mit gehaltener Sperre nicht funktioniert? Auch in dem Moment, in dem Sie sich READPASTmit HOLDLOCKIhnen verbinden, wird der Fehler angezeigt. Auf diesem Server findet keine Replikation statt, und die Isolationsstufe lautet READ COMMITTED.
ErikE
2
@ErikE - Genauso wichtig wie die Abfrage der Tabelle ist die Struktur der Tabelle. Die Tabelle, die Sie als Warteschlange verwenden, muss in der Reihenfolge der Warteschlangen zusammengefasst werden, sodass das nächste Element, das aus der Warteschlange entfernt werden soll, eindeutig ist . Das ist kritisch. Wenn Sie den obigen Code überfliegen, werden keine definierten Clustered-Indizes angezeigt.
Nick Chammas
@Nick das macht absolut herausragenden Sinn und ich weiß nicht, warum ich nicht daran gedacht habe. Ich habe die richtige PK-Einschränkung hinzugefügt (und mein Skript oben aktualisiert) und trotzdem Deadlocks erhalten. Die Elemente wurden jetzt jedoch in der richtigen Reihenfolge verarbeitet, mit Ausnahme der Wiederholungsverarbeitung für die festgefahrenen Elemente.
ErikE
@ErikE - 1. Ihre Warteschlange sollte nur Elemente enthalten, die sich in der Warteschlange befinden. Ausreihen und Element sollte bedeuten, dass es aus der Warteschlangentabelle gelöscht wird. Ich sehe, dass Sie stattdessen das aktualisieren StatusID, um einen Artikel zu entfernen. Ist das korrekt? 2. Ihre Reihenfolge muss eindeutig sein. Wenn Sie Artikel nach in die Warteschlange stellen GETDATE(), ist es sehr wahrscheinlich, dass mehrere Artikel gleichzeitig in die Warteschlange gestellt werden können. Dies führt zu Deadlocks. Ich schlage vor, ein IDENTITYzu dem Clustered-Index hinzuzufügen , um eine eindeutige Reihenfolge zu gewährleisten.
Nick Chammas
1

SQL Server eignet sich hervorragend zum Speichern relationaler Daten. Eine Job-Warteschlange ist nicht so toll. Siehe diesen Artikel, der für MySQL geschrieben wurde, aber auch hier gelten kann. https://blog.engineyard.com/2011/5-subtle-ways-youre-using-mysql-as-a-queue-and-why-itll-bite-you

Eric Humphrey - Lotsahelp
quelle
Danke, Eric. In meiner ursprünglichen Antwort auf die Frage habe ich vorgeschlagen, SQL Server Service Broker zu verwenden, da ich sicher bin, dass die Methode "Tabelle als Warteschlange" nicht wirklich das ist, wofür die Datenbank erstellt wurde. Aber ich denke, das ist keine gute Empfehlung mehr, da SB wirklich nur für Nachrichten ist. Die ACID-Eigenschaften von Daten in der Datenbank machen es zu einem sehr attraktiven Container, den Sie (ab) verwenden können. Können Sie ein alternatives, kostengünstiges Produkt vorschlagen, das als allgemeine Warteschlange gut funktioniert? Und kann usw. usw. gesichert werden?
ErikE
8
Der Artikel ist eines bekannten Fehlers bei der Verarbeitung von Warteschlangen schuldig: Kombinieren Sie Status und Ereignisse in einer einzigen Tabelle (wenn Sie sich die Artikelkommentare ansehen, werden Sie feststellen, dass ich dies vor einiger Zeit beanstandet habe). Das typische Symptom für dieses Problem ist das Feld "Verarbeitet / Verarbeitet". Den Zustandes mit den Ereignissen der Kombination (dh. Die Zustandstabelle macht die ‚Warteschlange‘) führt die ‚Warteschlange‘ zu großen Größen in wachsenden (da die Zustandstabelle ist die Warteschlange). Das Trennen von Ereignissen in eine echte Warteschlange führt zu einer Warteschlange, die "leer wird" und sich viel besser verhält .
Remus Rusanu
Schlägt der Artikel nicht genau Folgendes vor: Die Warteschlangentabelle enthält NUR Elemente, die für die Arbeit bereit sind.
ErikE
2
@ErikE: Sie beziehen sich auf diesen Absatz, richtig? Es ist auch sehr einfach, das One-Big-Table-Syndrom zu vermeiden. Erstellen Sie einfach eine separate Tabelle für neue E-Mails. Wenn Sie mit der Verarbeitung fertig sind, fügen Sie sie in den Langzeitspeicher ein und löschen Sie sie dann aus der Warteschlangentabelle. Die Tabelle der neuen E-Mails bleibt in der Regel sehr klein und die Bearbeitung erfolgt schnell . Mein Streit damit ist, dass als Workaround für das Problem der "großen Warteschlangen" angegeben wird. Diese Empfehlung hätte bei der Eröffnung des Artikels stehen sollen, ist ein grundsätzliches Thema.
Remus Rusanu
Wenn Sie anfangen, in einer klaren Trennung von Staat und Ereignis zu denken, dann gehen Sie einen viel einfacheren Weg. Sogar die obige Empfehlung würde sich dahingehend ändern , dass neue E-Mails in die emailsTabelle und in die new_emailsWarteschlangenew_emailsemails eingefügt werden . Die Verarbeitung fragt die Warteschlange ab und aktualisiert den Status in der Tabelle . Dies vermeidet auch das Problem des "fetten" Zustands, der sich in Warteschlangen bewegt. Wenn wir über verteilte Verarbeitung und echte Warteschlangen mit Kommunikation (z. B. SSB) sprechen würden, dann würden die Dinge komplizierter, da der gemeinsame Status in verteilten Systemen problematisch ist.
Remus Rusanu