The best way to use a DB table as a job queue (a.k.a batch queue or message queue)
Solution 1
Here's what I've used successfully in the past:
MsgQueue table schema
MsgId identity -- NOT NULL
MsgTypeCode varchar(20) -- NOT NULL
SourceCode varchar(20) -- process inserting the message -- NULLable
State char(1) -- 'N'ew if queued, 'A'(ctive) if processing, 'C'ompleted, default 'N' -- NOT NULL
CreateTime datetime -- default GETDATE() -- NOT NULL
Msg varchar(255) -- NULLable
Your message types are what you'd expect - messages that conform to a contract between the process(es) inserting and the process(es) reading, structured with XML or your other choice of representation (JSON would be handy in some cases, for instance).
Then 0-to-n processes can be inserting, and 0-to-n processes can be reading and processing the messages, Each reading process typically handles a single message type. Multiple instances of a process type can be running for load-balancing.
The reader pulls one message and changes the state to "A"ctive while it works on it. When it's done it changes the state to "C"omplete. It can delete the message or not depending on whether you want to keep the audit trail. Messages of State = 'N' are pulled in MsgType/Timestamp order, so there's an index on MsgType + State + CreateTime.
Variations:
State for "E"rror.
Column for Reader process code.
Timestamps for state transitions.
This has provided a nice, scalable, visible, simple mechanism for doing a number of things like you are describing. If you have a basic understanding of databases, it's pretty foolproof and extensible.
Code from comments:
CREATE PROCEDURE GetMessage @MsgType VARCHAR(8) )
AS
DECLARE @MsgId INT
BEGIN TRAN
SELECT TOP 1 @MsgId = MsgId
FROM MsgQueue
WHERE MessageType = @pMessageType AND State = 'N'
ORDER BY CreateTime
IF @MsgId IS NOT NULL
BEGIN
UPDATE MsgQueue
SET State = 'A'
WHERE MsgId = @MsgId
SELECT MsgId, Msg
FROM MsgQueue
WHERE MsgId = @MsgId
END
ELSE
BEGIN
SELECT MsgId = NULL, Msg = NULL
END
COMMIT TRAN
Solution 2
The best way to implement a job queue in a relational database system is to use SKIP LOCKED
.
SKIP LOCKED
is a lock acquisition option that applies to both read/share (FOR SHARE
) or write/exclusive (FOR UPDATE
) locks and is widely supported nowadays:
- Oracle 10g and later
- PostgreSQL 9.5 and later
- SQL Server 2005 and later
- MySQL 8.0 and later
Now, consider we have the following post
table:
The status
column is used as an Enum
, having the values of:
-
PENDING
(0), -
APPROVED
(1), -
SPAM
(2).
If we have multiple concurrent users trying to moderate the post
records, we need a way to coordinate their efforts to avoid having two moderators review the same post
row.
So, SKIP LOCKED
is exactly what we need. If two concurrent users, Alice and Bob, execute the following SELECT queries which lock the post records exclusively while also adding the SKIP LOCKED
option:
[Alice]:
SELECT
p.id AS id1_0_,1
p.body AS body2_0_,
p.status AS status3_0_,
p.title AS title4_0_
FROM
post p
WHERE
p.status = 0
ORDER BY
p.id
LIMIT 2
FOR UPDATE OF p SKIP LOCKED
[Bob]:
SELECT
p.id AS id1_0_,
p.body AS body2_0_,
p.status AS status3_0_,
p.title AS title4_0_
FROM
post p
WHERE
p.status = 0
ORDER BY
p.id
LIMIT 2
FOR UPDATE OF p SKIP LOCKED
We can see that Alice can select the first two entries while Bob selects the next 2 records. Without SKIP LOCKED
, Bob lock acquisition request would block until Alice releases the lock on the first 2 records.
Related videos on Youtube
Tjkoopa
For me, fun is taking a program, language, whatever and making it do things it's designer never envisioned it doing. It comes in handy when I'm asked to do something a little outside the norm as part of my job.
Updated on April 20, 2022Comments
-
Tjkoopa about 2 years
I have a databases table with ~50K rows in it, each row represents a job that need to be done. I have a program that extracts a job from the DB, does the job and puts the result back in the db. (this system is running right now)
Now I want to allow more than one processing task to do jobs but be sure that no task is done twice (as a performance concern not that this will cause other problems). Because the access is by way of a stored procedure, my current though is to replace said stored procedure with something that looks something like this
update tbl set owner = connection_id() where available and owner is null limit 1; select stuff from tbl where owner = connection_id();
BTW; worker's tasks might drop there connection between getting a job and submitting the results. Also, I don't expect the DB to even come close to being the bottle neck unless I mess that part up (~5 jobs per minute)
Are there any issues with this? Is there a better way to do this?
Note: the "Database as an IPC anti-pattern" is only slightly apropos here because
- I'm not doing IPC (there is no process generating the rows, they all already exist right now) and
- the primary gripe described for that anti-pattern is that it results in unneeded load on the DB as processes wait for messages (in my case, if there are no messages, everything can shutdown as everything is done)
-
dkretz over 15 yearsRight - bad = synchronous IPC with blocking on a dbms SELECT as a read. You're presumably doing this as a strategy for introducing asynchronicity.
-
dkretz over 15 yearsBTW, if you want to put the reader(s) on a timer, it's useful to have them check infrequently, but if they find work, they can drain the queue before sleeping again.
-
Tjkoopa over 15 yearsNote my edit: if they find no work, they will never find work. But if that were not true...
-
Meydjer Luzzoli over 15 yearsHow do you know it's an antipattern in this case, or that the software design is improper? You don't have any context on which to base this comment whatsoever.
-
dkretz over 15 yearsI'd called it a useful pattern for asynchronous IPC. You can configure it to operate like any garden-variety message queue, and they aren't in my experience branded "antipatterns".
-
dkretz over 15 yearsHere's a reference to the antipattern - tripatlas.com/Database_as_an_IPC The difference is that we're discussing using the database as a message queue, not as a mechanism for processes to interoperate.
-
Tjkoopa over 15 yearsThe part described as "The reader pulls one message and changes the state to "A"ctive while it works on it." is the part I'm interested in. How do you do that bit? (aside from that, it looks like mine is the same as yours with out the stuff that isn't needed for my case.)
-
Tjkoopa over 15 yearsI have the data in the DB, when I'm done I need the data in the DB. In my case I see no reason to add another component to the system. (BTW microsoft.com/windowsserver2003/technologies/msmq/default.mspx)
-
dkretz over 15 yearsRight, that requires multiple SQL statements between BEGIN TRAN and COMMIT TRAN. Immediately following - an SP for pulling the next message - hacked up a bit, I've omitted error trapping since it was written pre-TRY/CATCH.
-
dkretz over 15 years-- PART 1 CREATE PROCEDURE GetMessage @MsgType VARCHAR(8) ) AS DECLARE @MsgId INT BEGIN TRAN SELECT TOP 1 @MsgId = MsgId FROM MsgQueue WHERE MessageType = @pMessageType AND State = 'N' ORDER BY CreateTime
-
dkretz over 15 yearsPART 2 IF @MsgId IS NOT NULL BEGIN UPDATE MsgQueue SET State = 'A' WHERE MsgId = @MsgId SELECT MsgId, Msg FROM MsgQueue WHERE MsgId = @MsgId END ELSE BEGIN SELECT MsgId = NULL, Msg = NULL END COMMIT TRAN
-
Amitd over 14 yearswhat if i have to select more than(multiple) one row(s) at a time? can i update all at same time?
-
dkretz over 14 yearsAssuming you mark them with a common timestamp, or selection-batch id, you can update them all in a single statement, yes. Or use the "A" state described above, and update where state = 'A'.
-
jasonjwwilliams over 12 yearsUsing a database as a message queue is an anti-pattern. You're going to get lock contention up the ying-yang, and if you're using an MVCC system with multiple workers you're going to end up with nebulous state for any record. You should use a message queue broker like RabbitMQ.
-
sumit over 2 yearsThanks for the awesome answer Vlad! :)