python中的协程

协程

概念

• 协程
  又称微线程(纤程), 是一种用户态的轻量级线程
• 子程序
  在所有的语言中都是层级调用的, 比如A中调用B, B在执行过程中调用C, C执行完返回B, B执行完返回A。这是通过栈实现的, 一个函数就是一个执行的子程序, 子程序的调用总是有一个入口、一次返回, 调用的顺序是明确的
• 理解协程
  普通理解: 线程是系统级别的, 它们是由操作系统调度。协程是程序级别, 由程序员根据需求自己调度。我们把一个线程中的一个个函数称为子程序, 那么一个子程序在执行过程中可以中断去执行别的子程序, 这就是协程。也就是说同一线程下的一段代码1执行执行着就中断, 然后去执行另一段代码2, 当再次回来执行代码1时, 接着从之前的中断位置继续向下执行
• 优点
  a、最大的优势就是协程极高的执行效率。因为子程序切换而不是线程切换, 而是由程序自身控制, 因此, 没有线程切换的开销, 和多线程相比, 线程数量越多, 协程的性能优势就越明显。
  b、不需要多线程的锁机制, 因为只有一个线程, 也不存在同时写变量冲突, 在协程中控制共享资源不加锁, 只需要判断状态就好了, 所以执行效率比多线程高很多。
• 缺点
  a、无法利用多核CPU, 协程的本质是单个线程, 他不能同时将多个CPU的多个核心使用上, 失去了标准线程使用多CPU的能力。
  b、进行阻塞操作(操作IO)会阻塞整个程序

同步和异步

同步和异步的概念

• 前言
  python由于GIL(全局锁)的存在, 不能发挥多核的优势, 其性能一直饱受诟病。然而在IO密集型的网络编程里, 异步处理比同步处理能提升成百上千倍的效率
  IO密集型就是磁盘的读取数据和输出数据非常大的时候就是属于IO密集型
  由于IO操作的运行时间远远大于CPU、内存运行时间, 所以任务的大部分时间都是在等待IO操作完成, IO的特点是CPU消耗小, 所以, IO任务越多, CPU效率越高, 当然不是越多越好, 有一个极限值。
• 同步
  指完成事物的逻辑,先执行第一个事务, 如果阻塞了, 会一直等待, 直到这个事务完成, 在执行第二个事务, 顺序执行
• 异步
  是和同步相对的, 异步是指在处理调用这个事务之后, 不会等待这个事务的处理结果, 直接处理第二个事务去了, 通过状态、通知、回调来通知调用者处理结果

同步与异步代码

• 同步

import time

def run(index):
    print("hello", index)
    time.sleep(2)
    print("world", index)

for i in range(1, 5):
    run(i)

• 异步

import time
import asyncio

async def run(i):
    print("hello", i)
    await asyncio.sleep(2)
    print("world", i)

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    tasks = []
    t1 = time.time()

    for url in range(1, 5):
        coroutine = run(url)
        task = asyncio.ensure_future(coroutine)
        tasks.append(task)
    loop.run_until_complete(asyncio.wait(tasks))
    t2 = time.time()
    print("总耗时: %.2f" % (t2-t1))

asyncio

概述

• asyncio模块
  是python3.4版本引入的标准库, 直接内置了对异步IO的操作
• 编程模式
  是一个消息循环, 我们从asyncio模块中直接获取一个EventLoop的引用, 然后把需要执行的协程扔到EventLoop中执行, 就实现了异步IO
• 说明
  能实现协程的不止asyncio, tornado和gevent都实现了类似功能
• 关键字的说明

关键字          说明
event_loop      消息循环, 程序开启一个无限循环, 把一些函数注册到事件循环上, 当满足事件发生的时候, 调用相应的协程函数
coroutine       协程对象, 指一个使用async关键字定义的函数, 它的调用不会立即执行函数, 而是会返回一个协程对象。协程对象需要注册到事件循环上, 由事件循环调用
task            任务, 一个协程对象就是一个原生可以挂起的函数, 任务则是对协程的进一步封装, 其中包含了任务的各种状态
async/await     python3.5用于定义协程的关键字, async定义一个协程, await用于挂起阻塞的异步调用接口

asyncio基本使用

• 定义一个协程

import asyncio
import time

async def run(x):
    print("waiting: %d"%x)
    await asyncio.sleep(x)
    print("结束run")

# 得到一个协程对象
coroutine = run(2)
asyncio.run(coroutine)

等同于

import asyncio

async def run(x):
    print("waiting: %d"%x)
    print("结束run")

loop = asyncio.get_event_loop()
loop.run_until_complete(run(2))

• 获取协程返回值

import asyncio

async def run():
    return 0;

def call_back(future):
    print("返回值为: ", future.result())

loop = asyncio.get_event_loop()
task = asyncio.ensure_future(run())
task.add_done_callback(call_back)
loop.run_until_complete(task)

Task概念

• Task, 是python中与事件循环进行交互的一种主要方式。
  创建Task, 意思就是把协程封装成Task实例, 并追踪协程的 运行 / 完成状态, 用于未来获取协程的结果。
• Task 核心作用: 在事件循环中添加多个并发任务;
  具体来说, 是通过asyncio.create_task() 创建Task, 让协程对象加入时事件循环中, 等待被调度执行。
  注意: Python 3.7 以后的版本支持 asyncio.create_task(), 在此前的写法为loop.create_task(), 开发过程中需要注意代码写法对不同版本python的兼容性。
• 需要指出的是, 协程封装成 Task 后不会立马启动, 当某个代码 await 这个 Task 的时候才会被执行。
  当多个Task被加入一个task_list的时候, 添加Task的过程中Task不会执行, 必须要用 await asyncio.wait() 或 await asyncio.gather() 将Task对象加入事件循环中异步执行。
• 一般开发中, 常用的写法是这样的:
  – 先创建 task_list 空列表
  – 然后用 asyncio.create_task() 创建 Task;
  – 再把 Task 对象加入 task_list;
  – 最后使用 await asyncio.wait 或 await asyncio.gather 将 Task 对象加入事件循环中异步执行。
  注意: 创建 Task 对象时, 除了可以使用 asyncio.create_task()之外, 还可以用最低层级的 loop.create_task() 或 asyncio.ensure_future(), 他们都可以用来创建 Task 对象。

aiohttp

安装与使用

pip install aiohttp

简单实例使用

aiohttp的自我介绍中就包含了客户端和服务器端, 所以我们分别来看下客户端和服务器端的简单实例代码。

客户端:

get请求

import aiohttp
import asyncio

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    async with aiohttp.ClientSession() as session:
        source = await fetch(session, 'http://httpbin.org/get')
        print(source)

asyncio.run(main())

post请求

import aiohttp
import asyncio

async def fetch(session, url):
    async with session.post(url, data="传递数据") as response:
        return await response.text()

async def main():
    async with aiohttp.ClientSession() as session:
        source = await fetch(session, 'http://httpbin.org/post')
        print(source)

asyncio.run(main())

代码解释:

在网络请求中, 一个请求就是一个会话, 然后aiohttp使用的是ClientSession来管理会话, 所以第一个重点, 看一下ClientSession:

class ClientSession:
    """First-class interface for making HTTP requests."""

在源码中, 这个类的注释是使用HTTP请求接口的第一个类。然后上面的代码就是实例化一个ClientSession突然后命名为session, 然后用session, 然后用session去发送请求。这里有一个坑, 那就是ClientSession.get()协程的必须参数只能是 str 类和 yarl.URL 的实例

非文本内容格式

对于网络请求, 有时候是去访问一张图片, 这种返回值是二进制的也是可以读取到的:

await response.read()

请求的自定义

ClientResponse(客户端响应) 对象:含有request_info(请求信息), 主要是url 和 headers 信息。raise_for_status 结构体上的信息会被复制给ClientResponseError实例。
(一) 自定义Headers
有时候做请求的时候需要自定义headers, 主要是为了让服务器认为我们是一个浏览器。然后就需要我们自己来定义一个headers:

headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.6261.95 Safari/537.36'
}
await session.post(url, headers = headers)

(二) 如果出现ssl验证失败的处理

import aiohttp
import asyncio
from aiohttp import TCPConnector

async def main():
    async with aiohttp.ClientSession(connector = TCPConnector(ssl=False)) as session:
        pass
asyncio.run(main())

(三) 自定义cookie
发送你自己的cookies给服务器, 你可以为ClientSession对象指定cookies对象

url = 'http://httpbin.org/cookies'
cookies = {'cookies_are': 'working'}
async with ClientSession(cookies=cookies) as session:
    async with session.get(url) as resp:
        assert await resp.json() == {
            "cookies" = {'cookies_are': 'working'}
        }

(四) 使用代理
aiohttp只支持http代理

proxy = 'http://127.0.0.1:10809'
async with aiohttp.ClientSession(headers=headers) as session:
    async with session.get(url=login_url, proxy=proxy) as response:
        resu = await response.text()

aiofiles文件读写

概述

平常使用的file操作模式为同步, 并且为线程阻塞。当程序 I/O 并发次数高的时候, CPU 被阻塞, 形成闲置。
线程开启文件读取异步模式
用线程 (Thread) 方式来解决。硬盘缓存可以被多个线程访问, 因此通过不同线程访问文件可以部分解决。但此方案涉及线程开启和关闭的开销, 而且不同线程间数据交换比较麻烦。

from threading import Thread
for file in list_file:
    tr = Thread(target=file.write, args=(data,))
    tr.start()

使用已编写好的第三方插件-aiofiles, 支持异步模式
使用aio插件来开启文件的非阻塞异步模式。

安装方法

pip install aiofiles

这个插件的使用和python原生open一致, 而且可以支持异步迭代

实例

打开文件

import asyncio
import aiofiles

async def main():
    async with aiofiles.open('first.m3u8', mode='r') as f:
        contents = await f.read()
        print(contents)

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

迭代

import asyncio
import aiofiles

async def main():
    async with aiofiles.open('filename') as f:
        async for line in f:
            print(line)

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