python异步和协程,python协程和线程

　　本文主要详细介绍python的高级过程。本文中的示例代码非常详细，具有一定的参考价值。感兴趣的朋友可以参考一下，希望能帮到你。

　　00-1010协同学的定义及螺纹差协同学标准协同学的优缺点；用python实现协同学的方法：asyncawait关键字事件循环协同学和协同学对象awaitTask对象asyncio。未来物体期货。未来对象异步迭代器什么是异步迭代器？什么是异步可迭代对象？异步上下文管理器uvloop异步redis异步MySQL crawler摘要

目录

　　协程,也称为微线程，纤维。(Synergy是用户模式下的轻量级线程)

　　作用:在执行A功能时，可以随时中断它来执行B功能，然后中断B功能继续执行A功能(可以自动切换)。但是这个过程不是函数调用(没有call语句)，过程很像多线程。但是，在执行过程中只有一个线程。

　　通俗的理解:在一个线程中的某个函数中，你可以把一些临时变量和当前函数的其他信息保存在任何地方，然后切换到另一个函数进行执行。注意不是通过调用函数来完成的，切换的次数和什么时候切换到原来的函数是开发者自己决定的。

协程的定义

　　多任务处理时，线程切换远比从系统级保存和恢复CPU上下文简单。操作系统为了程序运行的高效率，每个线程都有自己的缓存和其他数据，而操作系统也会帮助你恢复这些数据。因此，线程的切换是非常消耗性能的。然而，协调切换只是操作CPU的上下文，因此系统可以抵抗每秒几百万次的切换。

协程和线程差异

　　并发只能在一个线程中实现。

　　不需要锁定用户程序中存储多个控制流的上下文堆栈来修改共享数据。当一个进程遇到IO操作时，会自动切换到其他进程。

协程的标准

　　有自己的上下文和栈，不需要线程上下文切换的开销，属于程序级切换，操作系统完全察觉不到。因此，在没有原子操作的锁定和同步开销的情况下，切换控制流更加轻量和方便。简化了编程模型，实现了单线程内的并发效果，充分利用了cpu，可扩展性高，成本低。

协程的优点

　　无法利用多核资源：协进程的本质是单线程，不能同时使用单个CPU的多个核心。协同进程需要与多个进程协作，以便在多个CPU上运行。阻塞操作(比如IO)会阻塞整个程序的计算操作。使用协进程来回切换没有任何意义，但是来回切换和保存状态会降低性能。

协程的缺点

　　第三方模块Greenlet用于实现协同协议代码(Gevent协同协议

　　程就是基于greenlet实现）

async&await关键字

　　协程实现的方式有多种，目前最流行的方式就是async&await，其他的方式了解即可，本文就介绍最流行的一种方式

　　使用协程需要了解2个，事件循环和定义协程函数

事件循环

　　事件循环是一种处理多并发量的有效方式，可以理解为死循环，循环过程中去检测并执行某些代码，我们来看下面的伪代码

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

　　上面的伪代码的意思：获取事件循环中，然后不断监听任务列表，有任务就执行，执行完成的任务就移除，直到任务列表中的所有任务都完成，终止循环

　　使用事件循环的好处：使得程序员不用控制任务的添加、删除和事件的控制

　　代码中的写法如下：

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

协程函数和协程对象

　　想要定义协程函数，格式：async def 函数名协程对象：执行协程函数()得到的协程对象

# 定义协程函数
　　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写法

await

　　await是一个只能在协程函数中使用的关键字，用于遇到IO操作时挂起 当前协程（任务），当前协程（任务）挂起过程中 事件循环可以去执行其他的协程（任务），当前协程IO处理完成时，可以再次切换回来执行await之后的代码。

　　举例：我们创建了2个任务，一个下载图片，一个下载视频，我们先执行下载图片任务，这时遇到io操作，正常情况下会等待图片下载完，但await可以先挂起下载图片任务，然后自动切换到下载视频任务

　　使用方法：await + 可等待对象（协程对象、Future对象、Task对象）

　　案例1

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

　　案列2

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()) # ①协程函数放入事件循环中运行

　　上述的所有示例都只是创建了一个任务，即：事件循环的任务列表中只有一个任务，所以在IO等待时无法演示切换到其他任务效果。在程序想要创建多个任务对象，需要使用Task对象来实现。

Task对象

　　Tasks用于并发调度协程，通过asyncio.create_task(协程对象)的方式创建Task对象，这样可以让协程加入事件循环中等待被调度执行。除了使用asyncio.create_task()函数以外，还可以用低层级的loop.create_task()或ensure_future()函数。不建议手动实例化Task对象。

　　本质上是将协程对象封装成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("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())

　　案例await+任务列表（用的最多）

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

　　注意：asyncio.wait源码内部会对列表中的每个协程执行ensure_future从而封装为Task对象，所以在和wait配合使用时task_list的值为[func(),func()]也是可以的。

asyncio.Future对象

　　asyncio中的Future对象是一个相对更偏向底层的可对象，通常我们不会直接用到这个对象，而是直接使用Task对象来完成任务的并和状态的追踪。（ Task 是 Futrue的子类 ）

　　Future为我们提供了异步编程中的 最终结果 的处理（Task类也具备状态处理的功能）

　　案例1

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

　　结果就是程序一直在等待，无法结束

　　案例2

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

　　Future对象本身函数进行绑定，所以想要让事件循环获取Future的结果，则需要手动设置。而Task对象继承了Future对象，其实就对Future进行扩展，他可以实现在对应绑定的函数执行完成之后，自动执行set_result，从而实现自动结束。

　　虽然，平时使用的是Task对象，但对于结果的处理本质是基于Future对象来实现的。

futures.Future对象

　　在Python的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)

　　两个Future对象是不同的，他们是为不同的应用场景而设计，例如：concurrent.futures.Future不支持await语法 等。

　　在Python提供了一个将futures.Future对象包装成asyncio.Future对象的函数asynic.wrap_future。

　　接下里你肯定问：为什么python会提供这种功能？

　　其实，一般在程序开发中我们要么统一使用asycio的协程实现异步操作、要么都使用进程池和线程池实现异步操作。但如果 协程的异步和 进程池/线程池的异步 混搭时，那么就会用到此功能了。

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

　　应用场景：当项目以协程式的异步编程开发时，如果要使用一个第三方模块，而第三方模块不支持协程方式异步编程时，就需要用到这个功能，例如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 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) )

异步迭代器

什么是异步迭代器？

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

什么是异步可迭代对象？

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

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

　　异步迭代器其实没什么太大的作用，只是支持了async for语法而已。

异步上下文管理器

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

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

　　这个异步的上下文管理器还是比较有用的，平时在开发过程中 打开、处理、关闭 操作时，就可以用这种方式来处理。

uvloop

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

　　注意：知名的asgi uvicorn内部就是使用的uvloop的事件循环。

异步redis

　　当通过python去操作redis时，链接、设置值、获取值 这些都涉及网络IO请求，使用asycio异步的方式可以在IO等待时去做一些其他任务，从而提升性能。

　　安装Python异步操作redis模块

pip3 install aioredis

　　案例:连接多个redis做操作（遇到IO会切换其他任务，提供了性能）

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

异步MySQL

　　当通过python去操作MySQL时，连接、执行SQL、关闭都涉及网络IO请求，使用asycio异步的方式可以在IO等待时去做一些其他任务，从而提升性能。

　　安装Python异步操作redis模块

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

爬虫

　　在编写爬虫应用时，需要通过网络IO去请求目标数据，这种情况适合使用异步编程来提升性能，接下来我们使用支持异步编程的aiohttp模块来实现。

　　安装aiohttp模块

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

总结

　　本篇文章就到这里了，希望能够给你带来帮助，也希望您能够多多关注盛行IT软件开发工作室的更多内容!

郑重声明：本文由网友发布，不代表盛行IT的观点，版权归原作者所有，仅为传播更多信息之目的，如有侵权请联系，我们将第一时间修改或删除，多谢。