4 ms·
Reposting from a while back in case it solved a problem for someone. I use Postgres SKIP LOCKED as a queue. Postgres gives me everything I want. I can also do
by andrewstuart 7y ago
Reposting from a while back in case it solved a problem for someone.
I use Postgres SKIP LOCKED as a queue. Postgres gives me everything I want. I can also do priority queueing and sorting.
All the other queueing mechanisms I investigated were dramatically more complex and heavyweight than Postgres SKIP LOCKED.
Here is a complete implementation - nothing needed but Postgres, Python and psycopg2 driver:
import psycopg2
import psycopg2.extras
import random
db_params = {
'database': 'jobs',
'user': 'jobsuser',
'password': 'superSecret',
'host': '127.0.0.1',
'port': '5432',
}
conn = psycopg2.connect(**db_params)
cur = conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
def do_some_work(job_data):
if random.choice([True, False]):
print('do_some_work FAILED')
raise Exception
else:
print('do_some_work SUCCESS')
def process_job():
sql = """DELETE FROM message_queue
WHERE id = (
SELECT id
FROM message_queue
WHERE status = 'new'
ORDER BY created ASC
FOR UPDATE SKIP LOCKED
LIMIT 1
)
RETURNING *;
"""
cur.execute(sql)
queue_item = cur.fetchone()
print('message_queue says to process job id: ', queue_item['target_id'])
sql = """SELECT * FROM jobs WHERE id =%s AND status='new_waiting' AND attempts <= 3 FOR UPDATE;"""
cur.execute(sql, (queue_item['target_id'],))
job_data = cur.fetchone()
if job_data:
try:
do_some_work(job_data)
sql = """UPDATE jobs SET status = 'complete' WHERE id =%s;"""
cur.execute(sql, (queue_item['target_id'],))
except Exception as e:
sql = """UPDATE jobs SET status = 'failed', attempts = attempts + 1 WHERE id =%s;"""
# if we want the job to run again, insert a new item to the message queue with this job id
cur.execute(sql, (queue_item['target_id'],))
else:
print('no job found, did not get job id: ', queue_item['target_id'])
conn.commit()
process_job()
cur.close()
conn.close()
- swuecho 7y agoThanks. Could you also post the sql ddl?
- pacala 7y agoWhat happens if the process dies while processing a job? Is said job going to remain locked and unserviced until the end of times?
- xyzzy_plugh 7y agoOnce the client session holding the lock disconnects, closing the session, the lock is released. You can also set client options like idle_in_transaction_session_timeout to force transactions to close after a certain amount of idle time, which also releases the lock.
- nurettin 7y agoso the process that fails releases the lock marking the job undone which causes another process to lock it?
- xyzzy_plugh 7y agoIt means another process could grab it, yes. In the parent post's model, the "scheduler" operates in a "at least once" model", as unless the transaction is committed, a premature failure would result in the message will be picked up again by a scheduler, which could result in some duplicate work. The parent post's model also tracks "attempts", but this only captures known or acknowledged failures -- unacknowledged failures (i.e. crashes) will not be recorded, so a task which explodes the process would be re-run ad infinitum. An alternative method could be to record an attempt in a separate transaction, so that the next scheduler execution can detect that the job/message was serviced before, even if the message itself appears fresh.
- wojcikstefan 7y agoVery relevant (and highly recommended) read: https://www.2ndquadrant.com/en/blog/what-is-select-skip-locked-for-in-postgresql-9-5/ https://www.2ndquadrant.com/en/blog/what-is-select-skip-lock...