kapilt/mongoqueue

Simple python library for distributed queues in mongodb

mongoqueue

Properties

• Isolation

  Do not let different consumers process the same message.

• Reliablity

  Do not let a failed consumer disappear an item.

• Atomic

  Operations on the queue are atomic.

Usage

A queue can be instantiated with a mongo collection and a consumer identifier. The consumer identifier helps distinguish multiple queue consumers that are taking jobs from the queue:

>> from pymongo import Connection
>> from mongoqueue import MongoQueue
>> queue = MongoQueue(
...   Connection(TEST_DB).doctest_queue,
...   consumer_id="consumer-1",
...   timeout=300,
...   max_attempts=3)

The MongoQueue class timeout parameters specifies how long in a seconds a how long a job may be held by a consumer before its considered failed.

A job which timeouts or errors more than the max_attempts parameter is considered permanently failed, and will no longer be processed.

New jobs/items can be placed in the queue by passing a dictionary:

>> queue.put({"foobar": 1})

A job priority key and integer value can be specified in the dictionary which will cause the job to be processed before lower priority items:

>> queue.put({"foobar": 0}, priority=1})

An item can be fetched out by calling the next method on a queue. This returns a Job object:

>> job = queue.next()
>> job.payload
{"foobar": 1}

The job class exposes some control methods on the job, for marking progress, completion, errors, or releasing the job back into the queue.

• complete Marks a job as complete and removes it from the queue.

• error Optionally specified with a message, releases the job back to the

  queue, and increments its attempts, and stores the error message on the job.

• progress Optionally takes a progress count integer, notes progress on the job

  and resets the lock timeout.

• release Release a job back to the pool. The attempts counter is not modified.

As a convience the job supports the context manager protocol:

>> with job as data:
...   print data['payload']

{"foobar: 0}

If the context closure is exited without the job is marked complete, if there's an exception the error is stored on the job.

Inspired By

• [0] https://github.com/skiz/mongo_queue/blob/master/lib/mongo_queue.rb
• [1] http://blog.boxedice.com/2011/09/28/replacing-rabbitmq-with-mongodb/
• [2] http://blog.boxedice.com/2011/04/13/queueing-mongodb-using-mongodb/
• [3] https://github.com/lunaru/mongoqueue
• [4] http://www.captaincodeman.com/2011/05/28/simple-service-bus-message-queue-mongodb/

Running Tests

Unit tests can be run with

$ python setup.py nosetests

Changes

0.6.0 - Feb 4th, 2013 - Isolate passed in data from metadata in Job. 0.5.2 - Dec 9th, 2012 - Fix for regression in sort parameters from pymongo 2.4 0.5.1 - Dec 2nd, 2012 - Packaging fix for readme data file.

Credits

Kapil Thangavelu, author & maintainer Dustin Laurence, sort fix for pymongo 2.4 Jonathan Sackett, Job data isolation.

mongoqueue's Issues

Should add some indexs to speed up

It's very slow when pull tasks via next(), finally i find there is no any index.
After i added following indexs, it's fast now.
collection.ensure_index([('priority', ASCENDING)])
collection.ensure_index([('locked_by', ASCENDING), ('locked_at', ASCENDING), ('attempts', ASCENDING)])
collection.ensure_index([('_id', ASCENDING), ('locked_by', ASCENDING)])

Status of library

This library is in status coding, or is better use another, my project is in python 3 and this library not run the problem is that I use aws lambdas and I can't modify the code.

default timeout too long > 300 days

The default timeout is int: 300

See: https://github.com/kapilt/mongoqueue/blob/master/mongoqueue/mongoqueue.py#L34

It is used to find locked jobs which have expired here:
https://github.com/kapilt/mongoqueue/blob/master/mongoqueue/mongoqueue.py#L66

the use of one unnamed parameter to timedelta means it takes the argument as 'days'

The only way to get sub-day timeouts is to use day fraction.

See some examples here:

>>> from datetime import datetime, timedelta
>>> datetime.now()
datetime.datetime(2019, 6, 4, 17, 54, 52, 339957)

>>> datetime.now() - timedelta(300) # the default
datetime.datetime(2018, 8, 8, 17, 55, 4, 593967)

>>> datetime.now() - timedelta(1) # 1 day
datetime.datetime(2019, 6, 3, 17, 55, 12, 929884)

>>> datetime.now() - timedelta(0.04166667) # 1 hour as decimal days
datetime.datetime(2019, 6, 4, 16, 55, 39, 443801)

Please consider defaulting to a lower time-base like hours.

I had assumed it was minutes. (5hrs)

repair only affects one job

The use of find_and_modify means only one Job is recovered from stale lock.

The same query + update expressions work with the PyMongo function update_many which updates all documents that match.

it = q.collection.update_many({"locked_by": {"$ne": None}, "locked_at": { "$lt": datetime.now() - timedelta(0.125)}},{"$set": {"locked_by": None, "locked_at": None},"$inc": {"attempts": 1}})

'Collection' object is not callable

>>> queue = MongoQueue(Connection('localhost').doctest_queue, consumer_id='consumer-1', timeout=300, max_attempts=3)
>>> queue.put({"foobar": 1})
TypeError: 'Collection' object is not callable. If you meant to call the 'insert' method on a 'Database' object it is failing because no such method exists.

Attempts counter resets everytime on queue.next()

1. Defined
   queue = MongoQueue(
   ... Connection(TEST_DB).doctest_queue,
   ... consumer_id="consumer-1",
   ... timeout=300,
   ... max_attempts=3)

2. Then got job by ----- job = queue.next()

3. Job has error ---- job.error("Error") ------- here attempts counter incremented to 2

4. Then again got the same job by ----- job = queue.next() ------ but attempts counter got reset to 1

Question : how will the job be marked failed is attempts counter gets reset?

[pymongo] Connection() is Deprecated

Connection() is Deprecated --> Connection(TEST_DB).doctest_queue,
Now we should use MongoClient() instead of Connection()