[python] python进阶（17）协程

• 修改共享数据不需加锁
  
• 用户程序里自己保存多个控制流的上下文栈
  
• 一个协程遇到 IO 操作自动切换到其它协程
   
  

• 由于自身带有上下文和栈，无需线程上下文切换的开销,属于程序级别的切换，操作系统完全感知不到，因而更加轻量级
  
• 无需原子操作的锁定及同步的开销
  
• 方便切换控制流，简化编程模型
  
• 单线程内就可以实现并发的效果，最大限度地利用 cpu，且可扩展性高，成本低
   
  

• 无法利用多核资源：协程的本质是个单线程,它不能同时将 单个 CPU 的多个核用上,协程需要和进程配合才能运行在多 CPU 上
  
• 进行阻塞（Blocking）操作（如 IO 时）会阻塞掉整个程序
  
• 计算型的操作，利用协程来回切换执行，没有任何意义，来回切换并保存状态 反倒会降低性能。
   
  

• greenlet，是一个第三方模块，用于实现协程代码（Gevent协程就是基于greenlet实现）
  
• yield，生成器，借助生成器的特点也可以实现协程代码。
  
• asyncio，在Python3.4中引入的模块用于编写协程代码。
  
• async & awiat，在Python3.5中引入的两个关键字，结合asyncio模块可以更方便的编写协程代码（推荐）。
   
  

任务列表 = [任务1， 任务2， 任务3............]
while True:
    可执行的任务列表，已完成的任务列表 = 去任务列表中检查所有的任务，将'可执行'和'已完成'的任务返回
    
    for 就绪任务 in 可执行的任务列表：
        执行已就绪的任务
    for 已完成的任务 in 已完成的任务列表：
        在任务列表中移除 已完成的任务
    
    如果任务列表中的任务都已完成，则终止循环

import asyncio 获取事件循环
loop = asyncio.get_event_loop() 将任务放到任务列表，监听事件循环
loop.run_until_complete(任务) 关闭事件
loop.close()

# 定义协程函数
async def func():
    pass
 创建协程对象
result = func()

import asyncio
async def func():
    print("你好")
result = func() 方式1
loop = asyncio.get_event_loop()
loop.run_until_complete(result) 方式2
asyncio.run(result)  # python3.7写法

import asyncio
async def func():
    print("执行协程函数内部代码")
    # 遇到IO操作挂起当前协程（任务），等IO操作完成之后再继续往下执行。
    # 当前协程挂起时，事件循环可以去执行其他协程（任务）。
    response = await asyncio.sleep(2)
    print("IO请求结束，结果为：", response)
result = func()
asyncio.run(result)

import asyncio
async def others():
    print("start")  # ④打印start
    await asyncio.sleep(2)  # ⑤等待耗时2秒，这过程中可以切换到其他协程中去
    print("end")  # ⑥打印end
    return '返回值'
async def func():
    print("执行协程函数内部代码")  # ②执行协程函数，打印print代码
    response = await others()  # ③等待协程函数others
    print(f"io请求结束，结果为{response}")  # ⑦等待others结束后打印print语句
if __name__ == '__main__':
    asyncio.run(func())  # ①协程函数放入事件循环中运行

import asyncio
async def func():
    print(1)
    await asyncio.sleep(2)
    print(2)
    return "返回值"
async def main():
    print("main开始")
    # 创建协程，将协程封装到一个Task对象中并立即添加到事件循环的任务列表中，等待事件循环去执行（默认是就绪状态）。
    task1 = asyncio.create_task(func())
    # 创建协程，将协程封装到一个Task对象中并立即添加到事件循环的任务列表中，等待事件循环去执行（默认是就绪状态）。
    task2 = asyncio.create_task(func())
    print("main结束")
    # 当执行某协程遇到IO操作时，会自动化切换执行其他任务。
    # 此处的await是等待相对应的协程全都执行完毕并获取结果
    ret1 = await task1
    ret2 = await task2
    print(ret1, ret2)
    
    
asyncio.run(main())

import asyncio
async def func():
    print(1)
    await asyncio.sleep(2)
    print(2)
    return "返回值"
async def main():
    print("main开始")
    # 创建协程，将协程封装到Task对象中并添加到事件循环的任务列表中，等待事件循环去执行（默认是就绪状态）。
    # 在调用
    task_list = [asyncio.create_task(func()), asyncio.create_task(func())]
    print("main结束")
    # 当执行某协程遇到IO操作时，会自动化切换执行其他任务。
    # 此处的await是等待所有协程执行完毕，并将所有协程的返回值保存到done
    # 如果设置了timeout值，则意味着此处最多等待的秒，完成的协程返回值写入到done中，未完成则写到pending中。
    done, pending = await asyncio.wait(task_list)
    print(done)
asyncio.run(main())

async def main():
    # 获取当前事件循环
    loop = asyncio.get_running_loop()
    # 创建一个任务（Future对象），这个任务什么都不干。
    fut = loop.create_future()
    # 等待任务最终结果（Future对象），没有结果则会一直等下去。
    await fut
asyncio.run(main())

import asyncio
async def set_after(fut):
    await asyncio.sleep(2)
    fut.set_result("666")
async def main():
    # 获取当前事件循环
    loop = asyncio.get_running_loop()
    # 创建一个任务（Future对象），没绑定任何行为，则这个任务永远不知道什么时候结束。
    fut = loop.create_future()
    # 创建一个任务（Task对象），绑定了set_after函数，函数内部在2s之后，会给fut赋值。
    # 即手动设置future任务的最终结果，那么fut就可以结束了。
    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)

import time
import asyncio
import concurrent.futures
def func1():
    # 某个耗时操作
    time.sleep(2)
    return "OK"
async def main():
    loop = asyncio.get_running_loop()
    # 方式1. Run in the default loop's executor ( 默认ThreadPoolExecutor )
    # 第一步：内部会先调用 ThreadPoolExecutor 的 submit 方法去线程池中申请一个线程去执行func1函数，并返回一个concurrent.futures.Future对象
    # 第二步：调用asyncio.wrap_future将concurrent.futures.Future对象包装为asycio.Future对象。
    # 因为concurrent.futures.Future对象不支持await语法，所以需要包装为 asycio.Future对象 才能使用。
    fut = loop.run_in_executor(None, func1)
    result = await fut
    print('default thread pool', result)
    # 方式2. Run in a custom thread pool:
    # with concurrent.futures.ThreadPoolExecutor() as pool:
    #     result = await loop.run_in_executor(
    #         pool, func1)
    #     print('custom thread pool', result)
    # 方式3. Run in a custom process pool:
    # with concurrent.futures.ProcessPoolExecutor() as pool:
    #     result = await loop.run_in_executor(
    #         pool, func1)
    #     print('custom process pool', result)
asyncio.run(main())

import asyncio
import requests
async def download_image(url):
    # 发送网络请求，下载图片（遇到网络下载图片的IO请求，自动化切换到其他任务）
    print("开始下载:", url)
    loop = asyncio.get_event_loop()
    # requests模块默认不支持异步操作，所以就使用线程池来配合实现了。
    future = loop.run_in_executor(None, requests.get, url)
    response = await future
    print('下载完成')
    # 图片保存到本地文件
    file_name = url.rsplit('_')[-1]
    with open(file_name, mode='wb') as file_object:
        file_object.write(response.content)
if __name__ == '__main__':
    url_list = [
        'https://www3.autoimg.cn/newsdfs/g26/M02/35/A9/120x90_0_autohomecar__ChsEe12AXQ6AOOH_AAFocMs8nzU621.jpg',
        'https://www2.autoimg.cn/newsdfs/g30/M01/3C/E2/120x90_0_autohomecar__ChcCSV2BBICAUntfAADjJFd6800429.jpg',
        'https://www3.autoimg.cn/newsdfs/g26/M0B/3C/65/120x90_0_autohomecar__ChcCP12BFCmAIO83AAGq7vK0sGY193.jpg'
    ]
    tasks = [download_image(url) for url in url_list]
    loop = asyncio.get_event_loop()
    loop.run_until_complete( asyncio.wait(tasks) )

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 == 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())

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())

pip3 install uvloop

import asyncio
import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
 编写asyncio的代码，与之前写的代码一致。 内部的事件循环自动化会变为uvloop
asyncio.run(...)

pip3 install aioredis

import asyncio
import aioredis
async def execute(address, password):
    print("开始执行", address)
    # 网络IO操作：先去连接 77.95.4.197:6379，遇到IO则自动切换任务，去连接77.95.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://77.95.4.197:6379', "123456"),
    execute('redis://77.95.4.198:6379', "123456")
]
asyncio.run(asyncio.wait(task_list))

pip3 install aiomysql

import asyncio
import aiomysql
async def execute(host, password):
    print("开始", host)
    # 网络IO操作：先去连接 77.95.40.197，遇到IO则自动切换任务，去连接77.95.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('77.95.40.197', "123456"),
    execute('77.95.40.198', "123456")
]
asyncio.run(asyncio.wait(task_list))

pip3 install aiohttp

import aiohttp
import asyncio
async def fetch(session, url):
    print(f"发送请求：{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 = ["http://www.baidu.com", "http://www.taobao.com", "http://www.jd.com"]
        tasks = [asyncio.create_task(fetch(session, url)) for url in url_list]
        await asyncio.wait(tasks)
if __name__ == '__main__':
    asyncio.run(main())