- 可以理解为一个死循环，去检测并执行某些代码
任务列表 = [任务1, 任务2, 任务3,....] while True: 可执行任务列表, 已完成任务列表 = 检查任务列表中的所有任务, 将可执行和已完成的任务返回 for 就绪任务 in 可执行任务列表: 执行已就绪任务 for 已完成任务 in 已完成人物列表: 任务列表.remove(已完成任务) if not 任务列表: break
import asyncio # 生成或获取一个事件循环 loop = asyncio.get_event_loop() # 将任务注册到任务列表中 loop.run_until_complete(任务)2.快速上手 1.定义
- 定义函数时使用@asyncio.coroutine装饰器或使用async def xxx
- 协程对象：执行 协程函数() 得到的对象
async def func(): pass # result此时就是一个协程对象，函数内部代码不会执行 result = func()2.协程对象的执行需要借助事件循环来处理
import asyncio
async def func():
print("我是测试事件循环的")
result = func()
loop = asyncio.get_event_loop()
loop.run_until_complete(result)
# python3.7之后可以使用asyncio.run(result), 代替loop对象
3.await对象
-
await + 可等待的对象（协程对象、Future、Task对象------>IO等待）
-
示例1
import asyncio
async def func():
print("测试IO等待")
response = await asyncio.sleep(2)
print("结束", response)
asyncio.run(func())
- 示例2
import asyncio
async def others():
print("start")
await asyncio.sleep(2)
print("end")
return "执行结束"
async def func():
print("测试IO等待")
# 遇到IO操做，挂起当前协程（任务），等待IO操做完成之后再继续往下执行，当前协程挂起时，事件循环可以去执行其他协程（任务）
response = await others()
print("IO请求结束，结果为: ", response)
asyncio.run(func())
- 示例3：一个协程中可以由多个await
import asyncio
async def others():
print("start")
await asyncio.sleep(2)
print("end")
return "执行结束"
async def func():
print("测试IO等待")
# 遇到IO操做，挂起当前协程（任务），等待IO操做完成之后再继续往下执行，当前协程挂起时，事件循环可以去执行其他协程（任务）
response1 = await others()
print("IO请求结束，结果为: ", response1)
response2 = await others()
print("IO请求结束，结果为: ", response2)
asyncio.run(func())
- 上述写法，同步串行（当我们第一个任务结果需要传递给第二个任务时比较有用）
Task are used to shedule coroutines concurrently.
When a coroutines is wrapped into a Task with functions like asyncio.create_task() the coroutines is automatically sheduled to run soon。
-
在事件循环中添加多个任务
-
Tasks用于并发调度协程，通过asyncio.create_task(协程对象)，的方式创建Task对象，这样可以让协程加入事件循环中，等待被调度执行。除了使用asyncio.create_task()函数外，还可以使用低层及的loop.create_task()或asyncio.ensure_future()函数。不建议手动实例化Task对象。
-
示例1
import asyncio from functools import partial async def func(): print(1) await asyncio.sleep(2) print(2) return "finished" def callback(name, future): print("{}回调结果".format(name), future.result()) async def main(): print("main 开始") # 创建task对象，将当前执行func协程对象添加到事件循环中 task1 = asyncio.create_task(func()) task1.add_done_callback(partial(callback, 'task1')) task2 = asyncio.create_task(func()) task2.add_done_callback(partial(callback, 'task2')) print("main 结束") # 当执行某协程遇到IO操作时，会自动化切换执行其他任务 # 此处的await是等待相对应的协程全部执行完毕并获取结果 ret1 = await task1 ret2 = await task2 print("任务执行结果：", ret1, ret2) asyncio.run(main()) -
示例2
import asyncio async def func(): print(1) await asyncio.sleep(2) print(2) return "返回值" async def func2(): print("======test======") await asyncio.sleep(5) print('======test======') return "测试结果" async def main(): print("main 开始") task_list = [ asyncio.create_task(func()), asyncio.create_task(func2()) ] print("main 结束") done, pending = await asyncio.wait(task_list, timeout=3) # done集合，任务执行结果 # pending: 如果指定了timeout超时等待时间，超时时间内未完成的任务信息将会保存在pending中 print(done) print(pending) """ {result='返回值'>} { wait_for= ()]>>} """ asyncio.run(main()) asyncio.wait对应还可以使用asyncio.gather
asyncio.gather可以实现任务分组import asyncio async def func(): print(1) await asyncio.sleep(2) print(2) return "返回值" async def func2(): print("======test======") await asyncio.sleep(5) print('======test======') return "测试结果" async def main(): print("main 开始") task_list1 = [ asyncio.create_task(func()), asyncio.create_task(func2()) ] task_list2 = [ asyncio.create_task(func()), asyncio.create_task(func2()) ] print("main 结束") group1 = asyncio.gather(*task_list1) group2 = asyncio.gather(*task_list2) result = await asyncio.gather(group1, group2) # [['返回值', '测试结果'], ['返回值', '测试结果']] print(result) asyncio.run(main()) -
示例3：在创建事件循环前创建task错误示例
import asyncio async def func(): print(1) await asyncio.sleep(2) print(2) return "返回值" task_list = [ # 此时事件循环还未创建，所以直接将协程对象注册进事件循环时会出问题 asyncio.create_task(func()), asyncio.create_task(func()) ] # 此处才创建事件循环对象 asyncio.run(asyncio.wait(task_list)) -
示例4：在创建事件循环前创建task
import asyncio async def func(): print(1) await asyncio.sleep(2) print(2) return "返回值" task_list = [ func(), func() ] asyncio.run(asyncio.wait(task_list))
A Future is a specil low-level awaitable object that represents an eventual result of an asynchronous operation
-
Task继承Future，Task对象内部await结果的处理基于Future对象来的
-
示例1
import asyncio async def main(): # 获取当前事件循环 loop = asyncio.get_running_loop() # 创建一个任务（Future对象），这个任务什么都不干 fut = loop.create_future() # 等待任务最终结果（Future对象），没有结果则会一直等待下去 await fut asyncio.run(main()) -
示例2
import asyncio async def main(): # 获取当前事件循环 loop = asyncio.get_running_loop() # 创建一个任务（Future对象），这个任务什么都不干 fut = loop.create_future() # 等待任务最终结果（Future对象），没有结果则会一直等待下去 await fut asyncio.run(main()) -
示例3
import asyncio async def set_after(fut): await asyncio.sleep(2) fut.set_result("插入任务结果") async def main(): # 获取当前事件循环 loop = asyncio.get_running_loop() # 创建一个任务（Future对象），这个任务什么都不干 fut = loop.create_future() # 创建一个任务（Task对象），绑定了set_after函数，函数内部在2秒之后，会给赋值 await loop.create_task(set_after(fut)) # 等待任务最终结果（Future对象），没有结果则会一直等待下去 data = await fut print(data) asyncio.run(main())
-
使用线程池、进程池实现异步操做时用到的对象
import time from concurrent.futures import Future from concurrent.futures.thread import ThreadPoolExecutor from concurrent.futures.process import ProcessPoolExecutor def func(value): time.sleep(1) print(value) # 创建线程池 pool = ThreadPoolExecutor(max_workers=5) # 创建进程池 # pool = ProcessPoolExecutor(max_workers=5) for i in range(10): fut = pool.submit(func, i) print(fut) -
某些情况下使用到的第三方包还未支持asyncio，但项目中调用的地方是asyncio异步形式，此时就需要将该第三方包转为异步形式（两种Future的转换）
- 第一步：将该对象转换为concurrent.futures.Future对象
- 第二步：将concurrent.futures.Future对象转换为asyncio.Future对象
import time import asyncio from concurrent import futures def func1(): time.sleep(2) return "已完成" async def main(): loop = asyncio.get_running_loop() # 默认executor为：ThreadPoolExecutor # 第一步：内部调用ThreadPoolExecutor.submit方法去线程池申请一个线程去执行func1函数，并返回一个concurrent.futures.Future对象 # 第二步：调用asyncio.wrap_future装饰器，将concurrent.futures.Future包装为asyncio.Future对象 # 因为concurrent.futures.Future对象不支持await语法，所以需要包装为asyncio.Future对象才能使用 fut = loop.run_in_executor(None, func1) result = await fut print("default thread pool: ", result) # 如果想要使用进程池 # with futures.ProcessPoolExecutor() as pool: # result = await loop.run_in_executor(pool, func1) # print("default process pool: ", result) if __name__ == '__main__': asyncio.run(main())
-
异步迭代器：实现了 __aiter__() 和 __anext__() 方法的对象。__anext__ 必须返回一个 awaitable 对象。async for 会处理异步迭代器的 __anext__() 方法所返回的可等待对象，直到其引发一个 StopAsyncIteration 异常。由 PEP 492 引入。
-
异步可迭代对象：可在 async for 语句中被使用的对象。必须通过它的 __aiter__() 方法返回一个 asynchronous iterator。由 PEP 492 引入。
import asyncio class Reader(object): """ 自定义异步迭代器（同时也是异步可迭代对象） """ def __init__(self): self.count = 0 async def readline(self): # await asyncio.sleep(1) self.count += 1 if self.count == 100: return None return self.count def __aiter__(self): return self async def __anext__(self): val = await self.readline() if val is None: raise StopAsyncIteration return val async def func(): # 创建异步可迭代对象 async_iter = Reader() # async for 必须要放在async def函数内，否则语法错误。 async for item in async_iter: print(item) asyncio.run(func())
- 此种对象通过定义 __aenter__() 和 __aexit__() 方法来对 async with 语句中的环境进行控制。由 PEP 492 引入。
import asyncio
class AsyncContextManager:
def __init__(self):
self.conn = None
async def do_something(self):
# 异步操作数据库
return 666
async def __aenter__(self):
# 异步链接数据库
self.conn = await asyncio.sleep(1)
return self
async def __aexit__(self, exc_type, exc, tb):
# 异步关闭数据库链接
await asyncio.sleep(1)
async def func():
async with AsyncContextManager() as f:
result = await f.do_something()
print(result)
asyncio.run(func())
9.uvloop
-
Python标准库中提供了asyncio模块，用于支持基于协程的异步编程。
-
uvloop是 asyncio 中的事件循环的替代方案，替换后可以使得asyncio性能提高。事实上，uvloop要比nodejs、gevent等其他python异步框架至少要快2倍，性能可以比肩Go语言
-
安装uvloop
pip3 install uvloop
-
uvloop替换asyncio的事件循环
import asyncio import uvloop asyncio.set_event_loop_policy(uvloop.EventLoopPolicy()) # 编写asyncio的代码，与之前写的代码一致。 # 内部的事件循环自动化会变为uvloop asyncio.run(...)
-
当通过python去操作redis时，链接、设置值、获取值 这些都涉及网络IO请求，使用asycio异步的方式可以在IO等待时去做一些其他任务，从而提升性能。
pip3 install aioredis
import asyncio import aioredis async def execute(address, password): print("开始执行", address) # 网络IO操作：创建redis连接 redis = await aioredis.create_redis(address, password=password) # 网络IO操作：在redis中设置哈希值car，内部在设三个键值对，即： redis = { car:{key1:1,key2:2,key3:3}} await redis.hmset_dict('car', key1=1, key2=2, key3=3) # 网络IO操作：去redis中获取值 result = await redis.hgetall('car', encoding='utf-8') print(result) redis.close() # 网络IO操作：关闭redis连接 await redis.wait_closed() print("结束", address) asyncio.run(execute('redis://47.93.4.198:6379', "root!2345"))import asyncio import aioredis async def execute(address, password): print("开始执行", address) # 网络IO操作：先去连接 47.93.4.197:6379，遇到IO则自动切换任务，去连接47.93.4.198:6379 redis = await aioredis.create_redis_pool(address, password=password) # 网络IO操作：遇到IO会自动切换任务 await redis.hmset_dict('car', key1=1, key2=2, key3=3) # 网络IO操作：遇到IO会自动切换任务 result = await redis.hgetall('car', encoding='utf-8') print(result) redis.close() # 网络IO操作：遇到IO会自动切换任务 await redis.wait_closed() print("结束", address) task_list = [ execute('redis://10.221.120.207:6379', "root!2345"), execute('redis://10.221.120.208:6379', "root!2345") ] asyncio.run(asyncio.wait(task_list))
-
当通过python去操作MySQL时，连接、执行SQL、关闭都涉及网络IO请求，使用asycio异步的方式可以在IO等待时去做一些其他任务，从而提升性能
pip3 install aiomysql
import asyncio import aiomysql async def execute(): # 网络IO操作：连接MySQL conn = await aiomysql.connect(host='127.0.0.1', port=3306, user='root', password='123', db='mysql', ) # 网络IO操作：创建CURSOR cur = await conn.cursor() # 网络IO操作：执行SQL await cur.execute("SELECt Host,User FROM user") # 网络IO操作：获取SQL结果 result = await cur.fetchall() print(result) # 网络IO操作：关闭链接 await cur.close() conn.close() asyncio.run(execute())import asyncio import aiomysql async def execute(host, password): print("开始", host) # 网络IO操作：先去连接 47.93.40.197，遇到IO则自动切换任务，去连接47.93.40.198:6379 conn = await aiomysql.connect(host=host, port=3306, user='root', password=password, db='mysql') # 网络IO操作：遇到IO会自动切换任务 cur = await conn.cursor() # 网络IO操作：遇到IO会自动切换任务 await cur.execute("SELECt Host,User FROM user") # 网络IO操作：遇到IO会自动切换任务 result = await cur.fetchall() print(result) # 网络IO操作：遇到IO会自动切换任务 await cur.close() conn.close() print("结束", host) task_list = [ execute('10.221.120.207', "root!2345"), execute('10.221.120.208', "root!2345") ] asyncio.run(asyncio.wait(task_list))
-
aiohttp模块
pip3 install aiohttp
import aiohttp import asyncio async def fetch(session, url): print("发送请求：", url) async with session.get(url, verify_ssl=False) as response: text = await response.text() print("得到结果：", url, len(text)) async def main(): async with aiohttp.ClientSession() as session: url_list = [ 'https://python.org', 'https://www.baidu.com', 'https://academy.laiye.com/' ] tasks = [asyncio.create_task(fetch(session, url)) for url in url_list] await asyncio.wait(tasks) if __name__ == '__main__': asyncio.run(main())
import random
import asyncio
async def get_html(sleep_time):
print("waiting...")
await asyncio.sleep(sleep_time)
print("done after {}s".format(sleep_time))
if __name__ == '__main__':
tasks = [get_html(random.randint(1, 5)) for i in range(3)]
loop = asyncio.get_event_loop()
try:
loop.run_until_complete(asyncio.wait(tasks))
except KeyboardInterrupt as e:
all_tasks = asyncio.Task.all_tasks()
for task in all_tasks:
print("cancel task")
resp = task.cancel()
print(resp)
loop.stop() # 此方法仅仅修改状态
loop.run_forever()
finally:
loop.close() # 此方法会清空队列，关闭执进程池和线程池....
14.task子任务调度
import asyncio
async def compute(x, y):
print("Compute %s + %s..." % (x, y))
await asyncio.sleep(2)
return x + y
async def print_sum(x, y):
result = await compute(x, y)
print("%s + %s = %s" % (x, y, result))
if __name__ == '__main__':
loop = asyncio.get_event_loop()
loop.run_until_complete(print_sum(1, 2))
loop.close()
