<story><title>On SQS</title><url>https://www.tbray.org/ongoing/When/201x/2019/05/26/SQS</url></story><parent_chain><item><author>andrewstuart</author><text>I use Postgres SKIP LOCKED as a queue.&lt;p&gt;I used to use SQS but Postgres gives me everything I want.  I can also do priority queueing and sorting.&lt;p&gt;I gave up on SQS when it couldn&amp;#x27;t be accessed from a VPC.  AWS might have fixed that now.&lt;p&gt;All the other queueing mechanisms I investigated were dramatically more complex and heavyweight than Postgres SKIP LOCKED.</text></item></parent_chain><comment><author>andrewstuart</author><text>Here is a complete implementation:&lt;p&gt;&lt;pre&gt;&lt;code&gt;    import psycopg2
    import psycopg2.extras
    import random
    
    db_params = {
        &amp;#x27;database&amp;#x27;: &amp;#x27;jobs&amp;#x27;,
        &amp;#x27;user&amp;#x27;: &amp;#x27;jobsuser&amp;#x27;,
        &amp;#x27;password&amp;#x27;: &amp;#x27;superSecret&amp;#x27;,
        &amp;#x27;host&amp;#x27;: &amp;#x27;127.0.0.1&amp;#x27;,
        &amp;#x27;port&amp;#x27;: &amp;#x27;5432&amp;#x27;,
    }
    
    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(&amp;#x27;do_some_work FAILED&amp;#x27;)
            raise Exception
        else:
            print(&amp;#x27;do_some_work SUCCESS&amp;#x27;)
    
    def process_job():
    
        sql = &amp;quot;&amp;quot;&amp;quot;DELETE FROM message_queue 
    WHERE id = (
      SELECT id
      FROM message_queue
      WHERE status = &amp;#x27;new&amp;#x27;
      ORDER BY created ASC 
      FOR UPDATE SKIP LOCKED
      LIMIT 1
    )
    RETURNING *;
    &amp;quot;&amp;quot;&amp;quot;
        cur.execute(sql)
        queue_item = cur.fetchone()
        print(&amp;#x27;message_queue says to process job id: &amp;#x27;, queue_item[&amp;#x27;target_id&amp;#x27;])
        sql = &amp;quot;&amp;quot;&amp;quot;SELECT * FROM jobs WHERE id =%s AND status=&amp;#x27;new_waiting&amp;#x27; AND attempts &amp;lt;= 3 FOR UPDATE;&amp;quot;&amp;quot;&amp;quot;
        cur.execute(sql, (queue_item[&amp;#x27;target_id&amp;#x27;],))
        job_data = cur.fetchone()
        if job_data:
            try:
                do_some_work(job_data)
                sql = &amp;quot;&amp;quot;&amp;quot;UPDATE jobs SET status = &amp;#x27;complete&amp;#x27; WHERE id =%s;&amp;quot;&amp;quot;&amp;quot;
                cur.execute(sql, (queue_item[&amp;#x27;target_id&amp;#x27;],))
            except Exception as e:
                sql = &amp;quot;&amp;quot;&amp;quot;UPDATE jobs SET status = &amp;#x27;failed&amp;#x27;, attempts = attempts + 1 WHERE id =%s;&amp;quot;&amp;quot;&amp;quot;
                # 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[&amp;#x27;target_id&amp;#x27;],))
        else:
            print(&amp;#x27;no job found, did not get job id: &amp;#x27;, queue_item[&amp;#x27;target_id&amp;#x27;])
        conn.commit()
    
    process_job()
    cur.close()
    conn.close()&lt;/code&gt;&lt;/pre&gt;</text></comment>
<story><title>On SQS</title><url>https://www.tbray.org/ongoing/When/201x/2019/05/26/SQS</url></story><parent_chain><item><author>andrewstuart</author><text>I use Postgres SKIP LOCKED as a queue.&lt;p&gt;I used to use SQS but Postgres gives me everything I want.  I can also do priority queueing and sorting.&lt;p&gt;I gave up on SQS when it couldn&amp;#x27;t be accessed from a VPC.  AWS might have fixed that now.&lt;p&gt;All the other queueing mechanisms I investigated were dramatically more complex and heavyweight than Postgres SKIP LOCKED.</text></item></parent_chain><comment><author>sorokod</author><text>Using SKIP LOCKED - do you commit the change to the dequeued item (ack it) at the point where you exit the DB call. If so what happens if the instance that dequeued the messages  crashes?</text></comment>