异步可靠队列（基于redis）
arque的Python项目详细描述
阿尔克
异步可靠队列（基于redis）
灵感来自tomdewire的文章“Redis中的可靠排队（第1部分）”[1][2]和“torrelque”python模块[3]。在
特点：
- Asynchronous: based on asyncio and aioredis
- Reliable: at any moment task data stored in redis database
- Throttling: controls number of tasks in execution
- Delayed queue: defers task availability
- Dead letters: put task data in failed queue after number of predefined retry attempts
- Tested on Python 3.7 and redis server '>=3.0.6', '<=5.0.5'
- Used in containerized applications (managed by kubernetes) in high load environments
安装：
^{pr2}$用法：
importsignalimportrandomimportloggingimportasyncioimportaioredisimporttimefromfunctoolsimportwrapsfromarqueimportArquelogger=logging.getLogger(__name__)asyncdefshutdown(signal,loop):"""Cleanup tasks tied to the service's shutdown."""logging.info(f"Received exit signal {signal.name}...")tasks=[tfortinasyncio.all_tasks()iftisnotasyncio.current_task()][task.cancel()fortaskintasks]logging.info(f"Cancelling {len(tasks)}outstanding tasks")awaitasyncio.gather(*tasks)logging.info(f"Flushing metrics")loop.stop()defaioredis_pool(host='redis://localhost',encoding='utf8'):defwrapper(func):@wraps(func)asyncdefwrapped():redis=awaitaioredis.create_redis_pool(host,encoding=encoding)try:returnawaitfunc(redis=redis)finally:redis.close()awaitredis.wait_closed()returnwrappedreturnwrapper@aioredis_pool(host='redis://localhost',encoding='utf8')asyncdefproduce_task(redis=None):logger.info('Starting producing...')queue=Arque(redis=redis)whileTrue:for_inrange(1):task={'value':random.randint(0,99)}task_id=f"custom_{task['value']}_{time.time()}"logger.debug('Produced task %s',task)awaitqueue.enqueue(task,task_id=task_id,task_timeout=10,delay=1)awaitasyncio.sleep(1)asyncdefprocess(task_data):logger.debug('Consumed task %s',task_data)awaitasyncio.sleep(1)@aioredis_pool(host='redis://localhost',encoding='utf8')asyncdefconsume_task(redis=None):logger.info('Starting consuming...')queue=Arque(redis=redis,working_limit=3)whileTrue:task_id,task_data=awaitqueue.dequeue()iftask_id=='__not_found__':continueiftask_id=='__overloaded__':print(f'TASK ID: {task_id}')awaitasyncio.sleep(1)continueiftask_id=='__marked_as_failed___':print(f'FAILED ID: {task_id}')continuetry:awaitprocess(task_data)awaitqueue.release(task_id)exceptException:logger.exception('Job processing has failed')awaitqueue.requeue(task_id,delay=5)stats=awaitqueue.get_stats()logger.info(stats)@aioredis_pool(host='redis://localhost',encoding='utf8')asyncdefsweep_task(redis=None):logger.info('Starting sweeping...')queue=Arque(redis=redis,sweep_interval=5)awaitqueue.schedule_sweep()@aioredis_pool(host='redis://localhost',encoding='utf8')asyncdefstats_task(redis=None):logger.info('Starting stats...')queue=Arque(redis=redis)whileTrue:stats=awaitqueue.get_stats()logger.info(stats)awaitasyncio.sleep(5)asyncdefexample():tasks=[]for_inrange(5):tasks.append(consume_task())tasks.append(produce_task())tasks.append(sweep_task())tasks.append(stats_task())awaitasyncio.gather(*tasks)if__name__=='__main__':logging.basicConfig(level=logging.DEBUG,format='%(asctime)s%(message)s')loop=asyncio.get_event_loop()signals=(signal.SIGHUP,signal.SIGTERM,signal.SIGINT,signal.SIGUSR1)forsinsignals:loop.add_signal_handler(s,lambdas=s:asyncio.create_task(shutdown(s,loop)))try:loop.run_until_complete(example())finally:loop.close()logging.info("Successfully shutdown...")
参考
[1] Reliable Queueing in Redis (Part 1)
[2] DEWIRE Redis as a Reliable Work Queue.pdf
[3] torrelque
- 项目
标签：
欢迎加入QQ群-->： 979659372
