FIFO-Warteschlangentabelle für mehrere Worker in SQL Server

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

• Welchen SQL Server 2005/2008-Sperransatz sollte ich verwenden, um einzelne Tabellenzeilen in mehreren Serveranwendungsinstanzen zu verarbeiten?

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

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.

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