前言
在最近的项目中，需要实现一个周期性执行的定时器。第一个想法是用celery，但是看了一下之后发现针对这个简单的需求，celery还是太大了，有点杀鸡焉用牛刀的感觉。正好之前在看coroutine的东西，于是就在这方面试试。但是也遇到了好多问题吧，本文就是针对这些问题的总结，希望对大家有所帮助。
先放两个链接，是我学习coroutine时候看到的，感觉讲的挺好的，有必要后面翻译一下。
AsyncIO for the Working Python Developer
I don't understand Python's Asyncio
我都是用的Python3.5中添加的async/await来定义coroutine，所以本文适用于3.5和3.6。
Q1、coroutine的运行
根据官网的文档，有两个方法可以运行coroutine，run_forever和run_until_complete。
run_until_complete
import asyncio
async def hello():
await asyncio.sleep(1)
print('Hello')
return 'World'
loop = asyncio.get_event_loop()
r = loop.run_until_complete(hello())
print(r)
loop.close()
# result
Hello
World
run_until_complete等待hello运行完，返回hello的结果。
run_forever
import asyncio
async def hello(future):
await asyncio.sleep(1)
print('Hello')
future.set_result('World')
def callback(future):
print(future.result())
loop.stop()
loop = asyncio.get_event_loop()
ft = asyncio.Future()
asyncio.ensure_future(hello(ft))
ft.add_done_callback(callback)
try:
loop.run_forever()
finally:
loop.close()
# result
Hello
World
如果使用run_forever，那么coroutine的运行需要asyncio.ensure_future。这时候可以创建一个Future用于保存hello的运行结果。
add_done_callback为Future添加hello方法的回调。
在回调里有
loop.stop()，用于停止loop.run_forever()。
Q2、 在同步函数中执行coroutine
我在开发过程中遇到这样一个问题，必须要将coroutine封在一个同步函数内，同时还要能拿到这个coroutine的执行结果。苦恼了半天没什么好方法，我又仔细地拜读了下官方文档，发现await后面跟着的不仅可以是coroutine，还可以是Future以及所有awaitable的对象。
Awaitable: An object that can be used in an await
expression. Can be a coroutine or an object with an __await__()
method. See also PEP 492.
import asyncio
async def async_hello(future):
print('async hello')
await asyncio.sleep(1)
future.set_result('Hello World')
def hello():
ft = asyncio.Future()
asyncio.ensure_future(async_hello(ft))
# 重点就是这里，返回一个Future
return ft
async def run():
print('Run ...')
r = await hello()
print(r)
loop = asyncio.get_event_loop()
loop.run_until_complete(run())
loop.close()
Q3、 开启Debug模式
官网文档
官网文档中讲的很清楚了，也比较简单，我就不赘述了。
Q4、Task exception was never retrieved
刚开始的时候，我总是遇到这个异常，很是头大，最终也还是在官网文档找到了解决方法。
我这里提供一个最终使用的方案。
import asyncio
async def task():
"""最终执行的函数"""
while True:
print('something')
await asyncio.sleep(1)
async def run(lp):
"""将task封装了一下，提供cancel和其他异常的处理"""
try:
await task()
except asyncio.CancelledError:
print('Cancel the future.')
except Exception as e:
print(e)
# 引发其他异常后，停止loop循环
lp.stop()
loop = asyncio.get_event_loop()
future = asyncio.ensure_future(run(loop))
try:
loop.run_forever()
except KeyboardInterrupt:
future.cancel()
loop.run_until_complete(future)
finally:
# 不管是什么异常，最终都要close掉loop循环
loop.close()
到这里，基础的coroutine执行已经差不多了，后面则是第三方库的使用总结。
Q4、aiomysql
import asyncio
import aiomysql
from aiomysql.cursors import DictCursor
async def mysql_test():
pool = await aiomysql.create_pool(
host='localhost', port=3306, user='root',
password='', db='mysql'
)
async with pool.acquire() as conn:
await conn.set_charset('utf8')
async with conn.cursor(DictCursor) as cur:
await cur.execute('SELECT Host,User FROM user')
print(cur.rowcount)
print(await cur.fetchone())
print(await cur.fetchall())
# 如果是插入或者修改则需要commit
# await conn.commit()
loop = asyncio.get_event_loop()
loop.run_until_complete(mysql_test())
loop.close()
如果你不想每个地方都要重新定义pool和conn，那么你可以这样来：
import asyncio
import aiomysql
from aiomysql.cursors import DictCursor
class MysqlWrapper:
def __init__(self, host, port, user, password, db):
self.host = host
self.port = port
self.user = user
self.password = password
self.db = db
self._pool = None
async def pool(self):
if not self._pool:
self._pool = await aiomysql.create_pool(
host=self.host, port=self.port, user=self.user,
password=self.password, db=self.db
)
return self._pool
async def close(self):
if not self._pool:
return
self._pool.close()
await self._pool.wait_closed()
mysql = MysqlWrapper(
host='localhost', port=3306, user='root',
password='', db='mysql'
)
def mysql_context(wrapper):
def _mysql_context(func):
async def __mysql_context(*args, **kwargs):
pool = await wrapper.pool()
async with pool.acquire() as conn:
await conn.set_charset('utf8')
r = await func(conn=conn, *args, **kwargs)
await conn.commit()
return r
return __mysql_context
return _mysql_context
@mysql_context(mysql)
async def mysql_test(conn=None):
async with conn.cursor(DictCursor) as cur:
await cur.execute('SELECT Host,User FROM user')
print(cur.rowcount)
print(await cur.fetchone())
print(await cur.fetchall())
async def close_pool():
await mysql.close()
print('Close mysql pool')
loop = asyncio.get_event_loop()
loop.run_until_complete(mysql_test())
loop.run_until_complete(close_pool())
loop.close()
Q5、aioredis
import asyncio
import aioredis
async def redis_test():
pool = await aioredis.create_pool(('localhost', 6379), db=0)
async with pool.get() as conn:
print(await conn.keys('*'))
pool.close()
await pool.wait_closed()
loop = asyncio.get_event_loop()
loop.run_until_complete(redis_test())
loop.close()
Q6、aiohttp
import asyncio
import aiohttp
async def http_test():
async with aiohttp.ClientSession() as session:
async with session.get('http://www.cip.cc/') as resp:
print(await resp.text())
loop = asyncio.get_event_loop()
loop.run_until_complete(http_test())
loop.close()
aioredis和aiohttp同样有封装，请看gist
