On SQS

Post by sorokod » Mon, May 27, 2019, 2:11 PM UTC

napsterbr wrote: post #76

sorokod wrote: post #73

Earlier quoted context omitted.

My question assumed a scenario where a consumer dequeues a batch, commits the deqieued change, and then crashes while processing the batch. Offcourse one could delay the commit until all processing is completed but then reasoning about the queue throughput becomes tricky.

That's the challenge of distributed systems :) it really boils down to how you want failures to be handled. If you ack before processing, and then you crash, those messages are lost (assuming you can't recover from the crash and you are not using something like a two-phase commit). If you ack after processing, you may fail after the messages have been processed but before you've been able to ack them. This leads to d…

Post by zug_zug » Mon, May 27, 2019, 2:21 PM UTC

andrewstuart wrote: post #15

I use Postgres SKIP LOCKED as a queue. I used to use SQS but Postgres gives me everything I want. I can also do priority queueing and sorting. I gave up on SQS when it couldn't be accessed from a VPC. AWS might have fixed that now. All the other queueing mechanisms I investigated were dramatically more complex and heavyweight than Postgres SKIP LOCKED.

Post by Feeble » Mon, May 27, 2019, 2:25 PM UTC

yilugurlu wrote: post #82

Has anyone ever measured the latency of the sending message to SQS? I was using with ELB in t2.medium instances, and my API (handle => send message to queue => return {status: true}) response times were around 150 - 300 ms and replaced SQS with RabbitMQ, and it went down to around 75-100 ms. Does anyone think that sending message to SQS is slow? Edit: With this update, I was able to process almost 3 x requests with t…

Post by andrewstuart » Mon, May 27, 2019, 2:29 PM UTC

andrewstuart wrote: post #15

I use Postgres SKIP LOCKED as a queue. I used to use SQS but Postgres gives me everything I want. I can also do priority queueing and sorting. I gave up on SQS when it couldn't be accessed from a VPC. AWS might have fixed that now. All the other queueing mechanisms I investigated were dramatically more complex and heavyweight than Postgres SKIP LOCKED.

    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