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)])