4 ms·
Here are a couple of tips if you want to use postgres queues: - You probably want FOR NO KEY UPDATE instead of FOR UPDATE so you don't block inserts into table
by sa46 3y ago
Here are a couple of tips if you want to use postgres queues:
- You probably want FOR NO KEY UPDATE instead of FOR UPDATE so you don't block inserts into tables that have a foreign key relationship with the job table. [1]
- If you need to process messages in order, you don't want SKIP LOCKED. Also, make sure you have an ORDER BY clause.
My main use-case for queues is syncing resources in our database to QuickBooks. The overall structure looks like:
BEGIN; -- start a transaction
SELECT job.job_id, rm.data
FROM qbo.transmit_job job
JOIN resource_mutation rm USING (tenant_id, resource_mutation_id)
WHERE job.state = 'pending'
ORDER BY job.create_time
LIMIT 1 FOR NO KEY UPDATE OF job NOWAIT;
-- External API call to QuickBooks.
-- If successsful:
UPDATE qbo.transmit_job
SET state = 'transmitted'
WHERE job_id = $1;
COMMIT;
This code will serialize access to the transmit_job table. A more clever approach
would be to serialize access by tenant_id. I haven't figured out how to do that
yet (probably lock on a tenant ID first, then lock on the job ID).
Somewhat annoyingly, Postgres will log an error if another worker holds the row lock (since we're not using SKIP LOCKED). It won't block because of NOWAIT.
CrunchyData also has a good overview of Postgres queues: [2]
[1]: https://www.migops.com/blog/2021/10/05/select-for-update-and-its-behavior-with-foreign-keys-in-postgresql/ https://www.migops.com/blog/2021/10/05/select-for-update-and...
[2]: https://blog.crunchydata.com/blog/message-queuing-using-native-postgresql https://blog.crunchydata.com/blog/message-queuing-using-nati...
- dilyevsky 3y agoNot doing SKIP LOCKED will make it basically single threaded, no? I’m of the opinion that you should just use Temporal if you don’t need inter-job order guarantees
- AlfeG 3y agoAfree. Skip locked main utility is to provide work queue capability to PG. Even documentation of PG is referring to skip locked as a main driver for work queues
- sa46 3y agoI explicitly want single-threaded execution within a tenant. After writing the original post, I figured out how to parallelize across tenants. The trick is to limit the query to only look at the next pending job for each tenant, which then allows for using SKIP LOCKED to process tenants in parallel: SELECT job.job_id, rm.data FROM qbo.transmit_job job JOIN resource_mutation rm USING (tenant_id, resource_mutation_id) WHERE job.state = 'pending' AND job.job_id = ( SELECT qj2.job_id FROM qbo.transmit_job qj2 WHERE job.tenant_id = qj2.tenant_id AND qj2.state = 'pending' ORDER BY qj2.create_time LIMIT 1 ) ORDER BY job.create_time LIMIT 1 FOR NO KEY UPDATE OF job SKIP LOCKED > you should just use Temporal Do you mean the Airflow-esque DAG runner? I prefer Postgres because it's one less moving part, I like transactional guarantees, my volume is tiny, and I can tweak the queue logic like above with a simple predicate change.