rpqueue 0.31.1

vis_超时的自动重试

包含在rpqueue 0.30.0或更高版本中，您可以提供任务（现在是数据队列） 可见性超时，即（按照amazon sqs样式的语义）一个时间，用于 在自动重新输入之前，任务必须正确执行 进入队列。：

@task(attempts=20, vis_timeout=5, use_dead=False)
def usually_eventually_succeed(**kwargs):
    # (4/5)**20  is ~ 0.0115, so call chain fails about 1% of the time
    if not random.randrange(5):
        return "done!"

    time.sleep(6) # fail silently

死信任务队列

如果你想知道哪些任务失败了，失败的呼叫可以自动 进入死信队列。：

@rpqueue.task(attempts=5, vis_timeout=5, use_dead=True)
def fails_to_dead(**kwargs):
    # (4/5)**5  is 0.32768, so call chain fails about 33% of the time
    if not random.randrange(5):
        return "done!"

    time.sleep(6) # fail silently

task_deadletter = rpqueue.Data(rpqueue.DEADLETTER_QUEUE, is_tasks=True)
dead_tasks = task_deadletter.get_data(items=5)

有关详细信息，请参见help(rpqueue.Data)。

crontab

使用类似crontab的语法支持cron_任务需要python crontab 模块：http://pypi.python.org/pypi/crontab/，允许：

@cron_task('0 5 tue * *')
def function2():
    # Will be executed every Tuesday at 5AM.
    pass

数据队列

将数据放入队列，而不是任务。我是说，应该是从 开始，但它现在在这里。

便捷功能：

• 1-100个每次读取0个数据项，由您自行决定
• vis_timeout
• attempts
• use_dead
• 如果您想继续处理数据，请刷新数据（我们不识别读卡器，因此如果您希望保证独占性，则应使用显式锁）

几个例子：

# 0/1 queue
dq = rpqueue.Data('best_effort')
dq.put_data([item1, item2, item3, ...])
items = dq.get_data(2) # {<uuid>: <item>, ...}

# Up to 5 deliveries, with 5 second delay before re-insertion
dq5 = rpqueue.Data('retry_processing', attempts=5, vis_timeout=5)
dq5.put_data([item1, item2, item3, ...])
items = dq5.get_data(2) # {<uuid>: <item>, ...}
items2 = dq5.get_data(2, vis_timeout=20) # override timeout on read
refreshed = set(dq5.refresh_data(items, vis_timeout=7)) # refresh our lock
items = {k:v for k,v in items if k in refreshed}
dq5.done_data(items)
dq5.done_data(items2)

# Up to 1 try with a 5 second delay before insertion into deadletter queue
dqd = rpqueue.Data('retry_processing', attempts=1, vis_timeout=5, use_dead=True)
dqd.put_data([item1, item2, item3, ...])
items = dqd.get_data(2) # {<uuid>: <item>, ...}
items2 = dqd.get_data(2, vis_timeout=20) # override timeout on read
refreshed = set(dqd.refresh_data(items, vis_timeout=7)) # refresh our lock
items = {k:v for k,v in items if k in refreshed}
dqd.done_data(items)
time.sleep(20)
# items2 are now "dead"
dead = rpqueue.Data(rpqueue.DEADLETTER_QUEUE)
dead_items = dead.get_data(2) # these have a different format, see docs!

更接近实际情况的较长示例：

aggregate_queue = rpqueue.Data("aggregate_stats", vis_timeout=30, use_dead=False)

@rpqueue.periodic_task(60)
def aggregate():
    # If vis_timeout is not provided, will use the queue default.
    # If vis_timeout is <= 0, will act as a 0/1 queue, and later "done data"
    # calling is unnecessary.
    data = aggregate_queue.get_data(items=100, vis_timeout=5)
    # data is a dictionary: {<uuid>: <item>, <uuid>: <item>, ...}
    # do something with data
    done_with = []
    for id, value in data.items():
        # do something with value
        done_with.append(id)

    aggregate_queue.refresh_data(data) # still working!

    # You can pass any iterator that naturally iterates over the uuids you
    # want to be "done" with.
    aggregate_queue.done_data(done_with)
    # also okay:
    # aggregate_queue.done_data(data)
    # aggregate_queue.done_data(tuple(data))
    # aggregate_queue.done_data(list(data))