python：优雅的退出程序或重启服务 在微服务中，使用任务队列有助于松耦合的设计，但有时，我们需要重启服务，但不能打断队列中正在进行的任务。 正确的做法是handle sigterm信号，具体代码如下： ``` import sys import argparse import logging import signal import asyncio class GracefulKiller: kill_now = False def __init__(self): signal.signal(signal.SIGINT, self.exit_gracefully) signal.signal(signal.SIGTERM, self.exit_gracefully) def exit_gracefully(self,signum, frame): self.kill_now = True async def loop_task(): killer = GracefulKiller() while 1: print("ha") # 任务主体 if killer.kill_now: break await asyncio.sleep(2) loop = asyncio.get_event_loop() try: loop.run_until_complete(loop_task()) finally: loop.run_until_complete(loop.shutdown_asyncgens()) # see: https://docs.python.org/3/library/asyncio-eventloop.html#asyncio.AbstractEventLoop.shutdown_asyncgens loop.close() ``` ## 利用异步 python3.5 之后，有了比较完善的asyncio库和协程机制。对IO密集型任务，可以使用异步调用在单线程中实现“并发”。极大的增加任务吞吐。 想要让IO 任务并发，只需要使用支持asyncio的库（比如aiohttp)，简单的loop.create_task就行。有时，需要限制后台任务的数量，在重启服务的时候，需要等待所有后台并发任务的完成。 此时消费者可以使用信号量进行控制。 ``` import argparse import logging import signal import asyncio class GracefulKiller: kill_now = False def __init__(self): signal.signal(signal.SIGINT, self.exit_gracefully) signal.signal(signal.SIGTERM, self.exit_gracefully) def exit_gracefully(self,signum, frame): self.kill_now = True def pop_queue(task_queue): try: task_args = task_queue.pop(0) print("dequeue args {}".format(task_args)) except: task_args = None return task_args async def wait_tasks_done(semaphore, value): while semaphore._value != value: print("wait all task done!") await asyncio.sleep(1) # 处理单个异步任务 async def run_task(task_args, semaphore): print("run_task {}".format(task_args)) await asyncio.sleep(3) # 模拟耗时 print("run_task {} done".format(task_args)) semaphore.release() async def loop_task(): killer = GracefulKiller() loop = asyncio.get_event_loop() task_queue = [x for x in range(0, 10)] # 模拟一个任务队列 task_limit = 2 # 同时允许任务数量 semaphore = asyncio.Semaphore(value=task_limit, loop=loop) while 1: task_args = pop_queue(task_queue) if task_args != None: await semaphore.acquire() loop.create_task(run_task(task_args, semaphore)) if killer.kill_now: await wait_tasks_done(semaphore, task_limit) break loop = asyncio.get_event_loop() try: loop.run_until_complete(loop_task()) finally: loop.run_until_complete(loop.shutdown_asyncgens()) loop.close() ``` 输出： ``` dequeue args 0 dequeue args 1 dequeue args 2 run_task 0 run_task 1 run_task 0 done run_task 1 done wait all task done! run_task 2 wait all task done! wait all task done! run_task 2 done ``` 上面的程序，无论何时重启，都将等待所有后台的任务完成。妈妈再也不用担心我重启服务被用户投诉了。 来自 大脸猫 写于 2017-11-18 00:47 -- 更新于2020-10-19 13:06 -- 2 条评论
