Python – Processing SQS queues with boto

amazon-sqsamazon-web-servicesbotopython

I have a python script using the boto library on ec2 instance which is part of an autoscaling group. The script processes messages from a SQS queue:

import boto
from boto.sqs.message import Message

conn = boto.connect_sqs()
q = conn.create_queue('queue-name')

while (qin.count() > 0):
    m = q.get_messages()
    #do something with the message

Does using the while statement make sense? Does count() update in real time as:

other instances take messages off the queue (or am I going to double up)
new messages are added to the queue (or will I miss them?)

How do I make this script constantly listen for new additions to the queue, even while the queue is empty?

In this question Processing items in SQS queue with a php script it was mentioned that 'sqs ruby client library has a method "poll" which continuously polls the queue and on receiving a message in the queue passes it on to a block'. Is there an equivalent in Python?

It has also been suggested that the SNS could be used to notify the scripts of the message queue status but I do not see how you could configure a responsive system with SNS as the metric alarms are not fine grained enough.

Best Answer

You shouldn't rely on the count for a queue as it's only meant to provide an approximate count and is not guaranteed to be accurate.

If you want to just keep polling forever, just do this:

while 1:
    messages = q.get_messages()
    # do something with messages
    time.sleep(N)

I have added the call to time.sleep to introduce a delay in the loop. The value of N should be at least one second and could be considerably more, depending on how quickly you expect new messages to appear in your queue. If you don't put some sort of delay in the loop you will probably start getting throttled by the service.

To avoid a message getting read multiple times, you should try to adjust the visibility timeout of the queue to be a value greater than the time it takes you to process a message and then make sure you delete the message when processing has completed.

Related Solutions

Python – How to get all messages in Amazon SQS queue using boto library in Python

I've been working with AWS SQS queues to provide instant notifications, so I need to be processing all of the messages in real time. The following code will help you to efficiently dequeue (all) messages and handle any errors when removing.

Note: to remove messages off the queue you need to delete them. I'm using the updated boto3 AWS python SDK, json library, and the following default values:

import boto3
import json

region_name = 'us-east-1'
queue_name = 'example-queue-12345'
max_queue_messages = 10
message_bodies = []
aws_access_key_id = '<YOUR AWS ACCESS KEY ID>'
aws_secret_access_key = '<YOUR AWS SECRET ACCESS KEY>'
sqs = boto3.resource('sqs', region_name=region_name,
        aws_access_key_id=aws_access_key_id,
        aws_secret_access_key=aws_secret_access_key)
queue = sqs.get_queue_by_name(QueueName=queue_name)
while True:
    messages_to_delete = []
    for message in queue.receive_messages(
            MaxNumberOfMessages=max_queue_messages):
        # process message body
        body = json.loads(message.body)
        message_bodies.append(body)
        # add message to delete
        messages_to_delete.append({
            'Id': message.message_id,
            'ReceiptHandle': message.receipt_handle
        })

    # if you don't receive any notifications the
    # messages_to_delete list will be empty
    if len(messages_to_delete) == 0:
        break
    # delete messages to remove them from SQS queue
    # handle any errors
    else:
        delete_response = queue.delete_messages(
                Entries=messages_to_delete)

Python – How to get messages receive count in Amazon SQS using boto library in Python

There are at least a couple of ways of doing this.

When you read a message in boto, you receive a Message object or some subclass thereof. The Message object has an "attributes" field that is a dict containing all message attributes known by SQS. One of the things SQS tracks is the approximate # of times the message has been read. So, you could use this value to determine whether the message should be deleted or not but you would have to be comfortable with the "approximate" nature of the value.

Alternatively, you could record message ID's in some sort of database and increment a count field in the database each time you read the message. This could be done in a simple Python dict if the messages are always being read within a single process or it could be done in something like SimpleDB if you need to record readings across processes.

Hope that helps.

Here's some example code:

>>> import boto.sqs
>>> c = boto.sqs.connect_to_region()
>>> q = c.lookup('myqueue')
>>> messages = c.receive_message(q, num_messages=1, attributes='All')
>>> messages[0].attributes
{u'ApproximateFirstReceiveTimestamp': u'1365474374620',
 u'ApproximateReceiveCount': u'2',
 u'SenderId': u'419278470775',
 u'SentTimestamp': u'1365474360357'}
>>>

Best Answer

Related Solutions

Python – How to get all messages in Amazon SQS queue using boto library in Python

Python – How to get messages receive count in Amazon SQS using boto library in Python

Related Topic