第十五章：异步编程(1)

标签：__ 异步 await 协程 编程 第十五章 print def asyncio

第十五章：异步编程 - asyncio

为什么我们要学习异步编程

• 异步非阻塞、asyncio
• tornado、fastapi、django3、aiothttp.....

以上框架都在逐步支持异步特性，目前python本身也在向异步方向发展，作为开发者还是有必要去学习异步开发的

学习步骤：

1. 什么是协程
2. asyncio模块
3. 实战案例

1. 协程

协程不是计算机提供，操作系统只是提供了线程和进程。程序员人为创造。

协程运行程序的大致状态：在一个线程中运行多个任务，任务与任务之间来回切换，并在同一时间内只能运行一个任务。

协程（coroutine）也可以被称之为微线程，是一种用户态内的上下文切换技术。简而言之，其实就是通过一个线程实现代码块相互切换执行。

def func1():
    print(1)
	print(2)

def func2():
    print(3)
    print(4)


func1()
func2()

实现协程有以下几种方式：

• greenlet - 早期模块
• yield 关键字
• asyncio 装饰器 - 早期python版本（python3.4）
• async、await 关键字（python3.5）

1.1 greenlet 实现协程

# pip install greenlet

from greenlet import greenlet


def func1():
    print(1)
    gr2.switch()
    print(2)
    gr2.switch()
    
def func2():
    print(3)
    gr1.switch()
    print(4)
    gr2.switch()
    
gr1 = greenlet(func1)
gr2 = greenlet(func2)
gr1.switch()

1.2 yield 关键字实现协程

def func1():
    yield 1
    yield from func2()
    yield 2
    
def func2():
    yield 3
    yield 4
    
f1 = func1()
for item in f1:
    print(item)

1.3 asyncio

需要在python3.4及以上版本使用

import asyncio


@asyncio.coroutine
def func1():
    print(1)
    yield from asyncio.sleep(2)
    print(2)


@asyncio.coroutine
def func2():
    print(3)
    yield from asyncio.sleep(2)
    print(4)


tasks = [
    asyncio.ensure_future(func1()),
    asyncio.ensure_future(func2())
]

loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))

注意：以上代码为遇到IO阻塞则自动切换任务。

1.4 async & await 关键字

需要在python3.5及以上版本使用

import asyncio


async def func1():
    print(1)
    await asyncio.sleep(2)
    print(2)


async def func2():
    print(3)
    await asyncio.sleep(2)
    print(4)


tasks = [
    asyncio.ensure_future(func1()),
    asyncio.ensure_future(func2())
]

loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))

2. 协程意义

在一个线程中如果遇到了IO等待的时间，那么线程会充分利用这个等待时间去执行其他的任务。

案例：使用 requests 模块完成图片下载

• 普通方式

  # pip install requests
  import requests
  
  
  def download_image(url):
      print('开始下载: ', url)
      response = requests.get(url)
      print('下载完成...')
  
      # 保存图片
      file_name = url.rsplit('/')[-1]
      with open(file_name, mode='wb') as f:
          f.write(response.content)
  
  
  if __name__ == '__main__':
      url_list = [
          'http://pic.bizhi360.com/bbpic/98/10798.jpg',
          'http://pic.bizhi360.com/bbpic/92/10792.jpg',
          'http://pic.bizhi360.com/bbpic/86/10386.jpg'
      ]
      for item in url_list:
          download_image(item)
  

• 协程方式

  import aiohttp
  import asyncio
  
  
  async def download_image(session, url):
      print('发送请求: ', url)
      async with session.get(url, verify_ssl=False) as response:
          content = await response.content.read()
          file_name = url.rsplit('/')[-1]
          with open(file_name, mode='wb') as f:
              f.write(content)
  
  
  async def main():
      async with aiohttp.ClientSession() as session:
          url_list = [
              'http://pic.bizhi360.com/bbpic/98/10798.jpg',
              'http://pic.bizhi360.com/bbpic/92/10792.jpg',
              'http://pic.bizhi360.com/bbpic/86/10386.jpg'
          ]
  
          tasks = [asyncio.create_task(download_image(session, url)) for url in url_list]
          await asyncio.wait(tasks)
  
  
  if __name__ == '__main__':
      asyncio.run(main())
  

3. 异步编程

3.1 事件循环

事件循环是异步编程中的重要的组成部分，大家现在可以暂时理解为一个死循环，可以检测并执行某些代码。

# 伪代码
任务列表 = [任务1, 任务2, 任务3...]

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

import asyncio


# 生成或者获取一个事件循环
loop = asyncio.get_event_loop()
# 将任务放到事件循环[任务列表]中检测任务状态是否可执行
loop.run_until_complete(任务)

3.2 基本应用

• 协程函数：在定义函数时使用async def 函数名()修饰的函数被称之为协程函数
• 协程对象：执行 协程函数() 得到的就是协程对象

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

注意：在获取协程对象时 协程函数的内部代码并不会被执行。

想要执行协程函数，需要使用协程函数、协程对象、事件循环搭配使用。

事件循环会执行协程函数中的内部代码。

import asyncio

async def func():
    print('这是一个协程任务')
    
result = func()
loop = asyncio.get_event_loop()
loop.run_until_complete(result)

在python3.7之后无需自己创建事件循环, 可以直接使用asyncio.run(任务)来执行协程函数中的内部代码

import asyncio

async def func():
    print('这是一个协程任务')
    
result = func()
asyncio.run(result)

3.3 await 关键字

• await 定义：等待某些IO任务

• 基本使用方式：await + 可等待的对象（协程对象、Future对象、Task对象）

  示例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('执行协程函数内部代码')
      response = await others()
      print(response)
  
  asyncio.run(func())
  

  示例3：

  import asyncio
  
  
  async def others():
      print('start')
      await asyncio.sleep(2)
      print('end')
      return '这是执行完协程函数后所得到的结果'
    
    
  async def func():
      print('执行协程函数内部代码')
      # 注意：在当前代码中使用await进行io等待后, 会堵塞代码。当response_1执行完毕后解堵塞继续往下执行
      response_1 = await others()
      print(response_1)
      
      # response_2同理, 等待others任务执行完毕后才能继续往下执行
      response_2 = await others()
      print(response_2)
  
  asyncio.run(func())
  

  await就是等待对象执行完毕并得到返回值之后才会继续向下执行。

3.4 Task 对象

Tasks用于并发调度协程，通过asyncio.create_task(协程对象)的方式创建Task对象，这样可以让协程加入事件循环中等待被调度执行。

除了使用asyncio.create_task()函数外，还可以使用低层级的loop.create_task()或ensure_future()函数。但是不建议手动实例化Task对象

注意：asyncio.create_task()函数在 Python 3.7 中被加入。在 Python 3.7 之前可以改用低层级的asyncio.ensure_future()函数

实例1：

import asyncio

async def func():
    print(1)
    await asyncio.sleep(2)
    print(2)
    return '这是一个返回值'
  
async def main():
    print('开始任务...')
    # 创建task对象，将当前的func任务添加到事件循环中
    task_1 = asyncio.create_task(func())
    task_2 = asyncio.create_task(func())
    print('任务结束...')
    
    # 当执行某些协程遇到IO操作时，会自动切换执行其他任务
    # 此处的await是等待相对应的协程全部执行完毕并获取结果
    result_1 = await task_1
    result_2 = await task_2
    print(result_1, result_2)

asyncio.run(main())    

示例2：

import asyncio


async def func():
    print(1)
    await asyncio.sleep(2)
    print(2)
    return '这是一个返回值'


async def main():
    print('开始任务...')
    # 创建task对象，将当前的func任务添加到事件循环中
    task_list = [
        # 在创建task对象时, 可以在task内部给协程对象取别名: name=别名名称
        asyncio.create_task(func(), name='n1'),
        asyncio.create_task(func(), name='n2')
    ]
    print('任务结束...')
    # await 关键字之后只能连接可以等待的对象 列表不能在await之后
    # await task_list
    done, pending = await asyncio.wait(task_list, timeout=None)
    '''
    done: 接收协程函数所返回的值
    pending: 如果设置超时时间, 则接收在超时时间内没有执行完毕的函数的状态

    一般情况下不会使用pending参数
    '''
    print(done)  # 返回的类型为一个集合
    
    for item in done:  # 使用迭代的方式取出集合中的协程对象
        print(item.result())


asyncio.run(main())

示例3：

import asyncio


async def func():
    print(1)
    await asyncio.sleep(2)
    print(2)
    return '这是一个返回值'

task_list = [
    func(),
    func()
]
 
done, pending = asyncio.run(asyncio.wait(task_list))
print(done)

注意点：

• 如果需要将task列表放在函数外，需要注意在执行协程函数列表之前需要首先创建事件循环。

• 并且在列表中不能直接添加Task对象。Task对象需要基于事件循环执行。

3.5 asyncio.Future 对象

Task 继承 Future, Task对象内部中的await结果的处理基于Future对象来的

在Future对象中会保存当前执行的这个协程任务的状态，如果当前任务状态为finished, 则await不再等待。

示例1：

import asyncio


async def main():
    # 获取当前事件循环
    loop = asyncio.get_running_loop()
    # 创建一个任务[Future对象] 当前没有任何任务
    fut = loop.create_future()
    # 等待任务的最终结果，没有结果则一直等待
    await fut


asyncio.run(main())

示例2：

import asyncio


async def set_after(fut):
    await asyncio.sleep(2)
    fut.set_result('这是一个测试结果')


async def main():
    # 获取事件循环
    loop = asyncio.get_running_loop()

    # 创建一个任务, 并且当前任务没有绑定任何行为, 则这个任务永远不知道什么时候结束
    fut = loop.create_future()

    # 手动设置future任务的最终结果
    await loop.create_task(set_after(fut))

    # 等待Future对象获取最终的结果, 否则就一直等
    data = await fut
    print(data)


asyncio.run(main())

3.6 concurrent.futures.Future 对象

使用线程池、进程池实现异步操作时会使用到的对象。

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)

一般情况下, 代码编写需要统一编程风格，简而言之，就是如果使用的是线程/进程，则整个程序都统一使用线程/进程。

只有一种情况可能会进行交叉编程。一个项目中的所有IO请求为协程异步请求，假设MySQL数据库版本过低导致无法使用协程进行并发存储，这种情况会使用线程/进程完成并发存储任务。

import time
import asyncio
import concurrent.futures


def func_1():
    time.sleep(2)
    return '测试'


async def main():
    loop = asyncio.get_running_loop()

    # 在协程函数中运行普通函数 在执行函数时，协程内部会自动创建一个线程池来运行任务
    # run_in_executor()方法第一个参数为None时则默认创建一个线程池
    fut = loop.run_in_executor(None, func_1)
    result = await fut
    print('当前方式会自动创建一个线程池去执行普通函数: ', result)

    # 在协程函数中运行基于线程池的任务, 效果与以上代码一致
    with concurrent.futures.ThreadPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, func_1)
        print('在线程池中得到的执行结果: ', result)

    # 在协程函数中运行基于进程池的任务
    with concurrent.futures.ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, func_1)
        print('在进程池中得到的执行结果: ', result)


if __name__ == "__main__":
    asyncio.run(main())

案例：asyncio + 不支持异步的模块（requests）

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 f:
        f.write(response.content)


if __name__ == '__main__':
    url_list = [
        'http://pic.bizhi360.com/bbpic/98/10798.jpg',
        'http://pic.bizhi360.com/bbpic/92/10792.jpg',
        'http://pic.bizhi360.com/bbpic/86/10386.jpg'
    ]
    
    tasks = [download_image(url) for url in url_list]
    # loop = asyncio.get_event_loop()
    # loop.run_until_complete(asyncio.wait(tasks))
    asyncio.run(asyncio.wait(tasks))

3.7 异步迭代器

什么是异步迭代器？

实现了__aiter__() 和 __anext__() 方法的对象。__aiter__() 必须返回一个awaitable对象。async for会处理异步迭代器的 __anext__() 方法所返回的可等待对象，直到引发一个StopAsyncIteration异常。

什么是异步可迭代对象？

可在async for语句中被使用的对象。必须通过它的__aiter__()方法返回一个asynchronous iterator。

import asyncio


# 自定义异步迭代器
class Reader:
    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():
    obj = Reader()
    # 异步for循环必须在协程函数内执行，协程函数名称随意取名
    async for item in obj:
        print(item)


asyncio.run(func())

3.8 异步上下文管理器

此种现象通过定义__aenter__()和__axeit__()方法来对async with语句中的环境进行控制。

import asyncio


class AsyncContextManager:
    def __init__(self, conn=None):
        self.conn = conn

    async def do_something(self):
        # 异步操作数据库
        return 'crud'

    async def __aenter__(self):
        # 异步连接数据库
        self.conn = await asyncio.sleep(1)
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        # 异步关闭数据库连接
        await asyncio.sleep(1)


async def func():
    # 上下文管理器处理也需要在协程函数中运行
    async with AsyncContextManager() as f:
        result = await f.do_something()
        print(result)


asyncio.run(func())

3.9 uvloop

是asyncio的事件循环的替代方案。

uvloop事件循环的执行效率比asyncio默认的事件循环的效率高。

# pip install uvloop

import asyncio
import uvloop

# 设置事件循环为uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

# 编写的asyncio代码与之前一致

# 内部事件循环会自动切到uvloop
asyncio.run(...)

4. 实战案例

4.1 异步操作redis

在使用python代码操作redis时，像连接、读取/写入、断开都是IO操作。

pip install aioredis==1.3.1

案例1：

import asyncio
import aioredis


async def execute(address):
    print('开始执行: ', address)
    # 网络IO 创建redis连接
    redis = await aioredis.create_redis(address)
    # 网络IO 在redis中设置哈希值
    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('结束...')


asyncio.run(execute('redis://127.0.0.1:6379/0'))

案例2：

import asyncio
import aioredis


async def execute(address, password):
    print('开始执行: ', address)
    # 网络IO 创建redis连接
    redis = await aioredis.create_redis_pool(address, password=password)
    # 网络IO 在redis中设置哈希值
    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('结束...')

task_list = [
    execute('redis://localhost:6379/0', None),
    execute('redis://localhost:6379/1', None)
]


asyncio.run(asyncio.wait(task_list))

4.2 异步MySQL

pip install aiomysql

案例1：

import asyncio
import aiomysql


async def execute():
    # 网络IO操作 连接mysql
    conn = await aiomysql.connect(host='127.0.0.1', port=3306, user='root', password='root', db='mysql')

    # 网络IO操作 创建游标
    cursor = await conn.cursor()

    # 网络IO操作 执行sql
    await cursor.execute('select host,user from user')

    # 网络IO操作 获取sql结果
    result = await cursor.fetchall()
    print(result)

    # 网络IO操作
    await cursor.close()
    conn.close()


asyncio.run(execute())

案例2：

import asyncio
import aiomysql


async def execute(host, password):
    print('开始连接:', host)
    # 网络IO操作 连接mysql
    conn = await aiomysql.connect(host=host, port=3306, user='root', password=password, db='mysql')

    # 网络IO操作 创建游标
    cursor = await conn.cursor()

    # 网络IO操作 执行sql
    await cursor.execute('select host,user from user')

    # 网络IO操作 获取sql结果
    result = await cursor.fetchall()
    print(result)

    # 网络IO操作
    await cursor.close()
    conn.close()
    print('结束:', host)


task_list = [
    execute('localhost', 'root'),
    execute('localhost', 'root')
]


asyncio.run(asyncio.wait(task_list))

4.3 FastAPI 框架

pip install uvicorn
pip install fastapi

示例：

import uvicorn
import asyncio
import aioredis
from fastapi import FastAPI

app = FastAPI()

# 创建redis连接池
REDIS_POOL = aioredis.ConnectionsPool('redis://localhost:6379', password=None, minsize=1, maxsize=10)


@app.get('/')
def index():
    # 普通视图函数
    return {'message': 'hello world'}


@app.get('/red')
async def red():
    # 异步视图
    print('请求来了...')
    await asyncio.sleep(3)

    # 获取连接池中的一个链接
    conn = await REDIS_POOL.acquire()
    redis = aioredis.Redis(conn)

    # 设置值
    await redis.hmset_dict('car_fastApi', key1=1, key2=2, key3=3)

    # 读取值
    result = await redis.hgetall('car_fastApi', encoding='utf-8')
    print(result)

    # 将单个连接归还给连接池
    REDIS_POOL.release(conn)

    return result


if __name__ == '__main__':
  	# fastapi_test为当前这个脚本文件的名称
    uvicorn.run("fastapi_test:app", host='127.0.0.1', port=5000, log_level='info')

4.4 爬虫

import asyncio
import aiohttp


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',
        ]

        tasks = [asyncio.create_task(fetch(session, url)) for url in url_list]
        await asyncio.wait(tasks)


if __name__ == '__main__':
    asyncio.run(main())

标签：__,异步,await,协程,编程,第十五章,print,def,asyncio
From： https://www.cnblogs.com/shiruiyang/p/16825628.html