python 并发、异步和异步编程

最编程 2024-06-09 09:48:19

...

事件循环

可以理解为一个死循环，去检测并执行某些代码。

＃ 伪代码
任务列表 =〔任务1，任务2，任务了，……〕

while True :
	可执行的任务列表，已完成的任务列表 二去任务列表中检查所有的任务，将’可执行’和’已完成"的任务返回
	
	for 就绪任务 in 已准备就绪的任务列表：
		执行已就绪的任务
	
	for 已完成的任务 in 已完成的任务列表：
		在任务列表中移除 已完成的任务
	
	如果 任务列表 中的任务都已完成，则终止循环


# 去生成或获取一个事件循环
loop = asyncio.get_event_loop ()
# 将任务放到 ’任务列表‘
loop. run_until_complete(任务)

快速上手

协程函数，定义函数的时候 async def 函数名。
协程对象，执行协程函数（）得到的协程对象。

import asyncio

# 协程函数
async def func():
	pass

# 协程对象
result = func()

注意：执行协程函数创建协程对象，函数内部代码不会执行。如果想要运行协程函数内部代码，必须要将协程对象交给事件循环来处理。

import asyncio

# 协程函数
async def func():
	pass

# 协程对象
result = func()

# loop = asyncio.get_event_loop ()
# loop. run_until_complete(result )
asyncio.run( result )	# python3.7 后的写法，等同于上面两行代码。

await 关键字

从字面上来讲，a 表示 async ，wait 表示等待，表示等待执行。

await + 可等待的对象（协程对象，Future，Task对象->IO等待）

示列1：

import asyncio

async def func():
	print("开始")
	response = await asyncio.sleep(2)
	print("结束"，response)

asyncio.run( func() )

当执行asyncio.run( func() ) 会先打印 ”开始“ 当遇到函数中含有 await 关键字的代码，如果在事件循环中有其他未完成的任务则会切换到其他任务，直到IO阻塞结束，才会执行 "结束”。

import asyncio

async def others():
	print ("start")		# 2 执行协程函数内部代码
	await asyncio.sleep (2)
	print('end')	# 3 协程函数内部代码执行结束
	return '返回值'	
	
async def func() :
	print("执行协程函数内部代码")	# 1 进入主线程
	＃ 遇到I0操作挂起当前协程（任务），等IO操作完成之后再继续往下执行。当前协程挂起时，事件可以去执行其他协程（任务）。
	response = await others()
	print("Io请求结東，结果为：", response)	# 4 Io请求结東，结果为：'返回值'
	
asyncio.run( func() )

通过示列2 可以看出协程与协程之间可以嵌套调用。

示列3 ：

import asyncio


async def others():
    print("start")  
    await asyncio.sleep(2)
    print('end')  
    return '返回值'


async def func():
    print("执行协程函数内部代码")  

    response1 = await others()
    print("Io请求结東，结果1为：", response1)  

    response2 = await others()
    print("Io请求结東，结果2为：", response2)  

asyncio.run(func())

执行协程函数内部代码
start
end
Io请求结東，结果1为： 返回值
start
end
Io请求结東，结果2为： 返回值

Process finished with exit code 0

通过示列3 可以知道一个协程函数里面可以有多个await。并且在单个协程函数里面的多个await是串行的，直到上一个await 有返回值之后才会执行下一个await。

Task 对象

Tasks are used to schedule coroutines concurrently.
When a coroutine is wrapped into a Task with functions like asyncio.create task () the coroutine is automatically scheduled to run soon.

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

在协程中，task对象是一个抽象的概念，表示一个异步任务的执行状态。它通常与一个协程对象相关联，以便在协程执行时对其进行管理和控制。task对象有以下官方解释：

Task对象是一个Future对象的子类，它表示一个异步操作的执行状态。
Task对象具有额外的状态和方法，用于管理和控制协程的执行和取消操作。
Task对象可以从协程中创建，也可以在运行时动态创建。
Task对象可以被等待（await）以等待异步操作完成，并返回其结果或引发异常。
Task对象可以取消，以停止其关联的协程和异步操作的执行。
Task对象可以附加回调函数，在异步操作完成时自动运行。

示列1：

import asyncio


async def func():
    print(1)
    await asyncio.sleep(2)
    print(2)
    return "返回值，"


async def main():
    print("main开始")
    # 创建Task对象，将当前执行func函数任务添加到事件循环。
    task1 = asyncio.create_task(func())
    # 创建Task对象，将当前执行func函数任务添加到事件循环。
    task2 = asyncio.create_task(func())
    print("main结束")
    # 当执行某协程遇到IO操作时，会自动化切换执行其他任务。
    # 此处的await是等待相对应的协程全都执行完毕井获取结果
    ret1 = await task1
    ret2 = await task2
    print(ret1, ret2)


asyncio.run(main())

main开始
main结束
1
1
2
2
返回值， 返回值，

Process finished with exit code 0

示列2：

上面的示列1 是比较初级的写法，有助于理解协程的执行，在开发中经常用下面的写法

import asyncio


async def func():
    print(1)
    await asyncio.sleep(2)
    print(2)
    return "返回值，"


async def main():
    print("main开始")
    
    task_list = [
		asyncio.create_task(func(), name='t1'),
		asyncio.create_task(func(), name='t2')
	]
	
    print("main结束")
   
    done,pending = await asyncio.wait(task_list, timeout=None)
    print(done, pending )


asyncio.run(main())

main开始
main结束
1
1
2
2
{<Task finished name='t1' coro=<func() done, defined at E:\Project\PyCharmProject\Test\test\ts.py:4> result='返回值，'>, <Task finished name='t2' coro=<func() done, defined at E:\Project\PyCharmProject\Test\test\ts.py:4> result='返回值，'>} set()

Process finished with exit code 0

示列3：
除了示列2，还有一种方式可以创建task对象，这种方式作为了解即可。

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)

Future 对象

A Future is a special low-level awaitable object that represents an eventual result of an
asynchronous operation.

Task继承Future, Task 对象内部await结果的处理基于Future对象，在Future对象内部有一个 _state 变量记录了 Task 对象的状态，通过_state的状态实现await 的功能。

示列1：

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

由于fut 这个Future对象没有任何结果，所以执行当前程序会一直夯住。这个示列主要展示 await 后面也可以跟 Future对象。

示列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函数，西数内部在25之后，会给fut赋值。
	# 即手动设置future任务的最终结果，那么fut就可以结東了。
	await loop.create_task(set_after (fut))
	# 等待 Future对象获取 最终结果，否则一直等下去
	data = await fut
	print (data)

asyncio.run( main() )

在示列2 中，Future对象由于设置了值，所以在执行到await fut 时不会一直等待。当我们使用task 对象时，底层的Future 会自动将协程函数返回的值 set_result ，达到和上面同样的效果。

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)
	return 123
	
# 创建线程池
pool = ThreadPoolexecutor (max_workers=5)

# 创建进程池
# pool = ProcesspoolExecutor(max_workers=5)

for i in range (10):
	fut = pool.submit (func, i)
	print (fut)

我们知道当我们使用 async 这种协程的编码所使用到的第三方模块必须支持协程，假如存在第三方模块不支持异步的情况就可以使用线程池、进程池中的concurrent.futures.Future对象实现异步。

import time
import asyncio
import concurrent.rutures

def func1 ():
	# 某个耗时操作
	time.sleep (2)
	return "耗时操作"
	
asvnc 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 rut
	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:
	# wilh 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):
	#发送网络请求，下载图片（遇到网络下载图片的工0请求，自动化切换到其他任务)
	print（"开始下载："，ur1)
	loop = asyncio.get_event_1oop()
	#requests模块默认不支持异步操作，所以就使用线程池来配合实现了。
	future = loop.run_in_executor(None, requests.get, url)
	response = await future
	print('下载完成'）
	＃ 图片保存到本地文件
	file_name = url.rsplit('_')[-1]
	with open (file_name, mode='w') as file_object:
		file_object.write(response.content)
	
if __name_ == '_ main_ ':
	url_list =[
	"https://www3.autoimg.cn/newsdfs/g26/M02/35/A9/120×90_0_autohomecar__ChsEel2AXQ6A00H_AAFOCMS
	8nzu621.jpg'
	"https://www2.autoimg.cn/newsdfs/g30/M01/3C/E2/120×90_0_autohomecar_chccsv2BBICAUntfAADjJFd
	6800429.jpg',
	'https://www.autoimg.cn/newsdfs/g26/MOB/3C/65/120×90_0_autohomecar__ChcCP12BFCmAI083AAGq7vK
	OsGY193.jpg'
	]
	tasks = [ download_image(url) for url in url_list]
	
	loop = asyncio.get_event_loop()
	loop.run_until_complete( asyncio.wait(tasks) )

上一篇： python教程：使用 async 和 await 协程进行并发编程

下一篇： Python3: 异步 IO - asyncio 与任务、流、子进程、队列、同步-2.网络 IO：asyncio 流

python 并发、异步和异步编程

事件循环

快速上手

await 关键字

Task 对象

Future 对象

concurrent.futures.Future对象

异步和非异步模块混合案例

Python和NCL在处理netCDF文件中的scale_factor和add_offset: 一次全面的气象编程指南

理解Python编程中的观察者和策略模式应用

搞定C#异步编程：理解async和await的妙用

C#并发与异步的深度解析：五、探讨排他锁的作用

如何实现74161的同步和异步级联？—— 数字电子技术探讨

简化版！理解Vue中的动态组件与Keep-Alive缓存、Webpack的代码分割及异步组件、$ref引用元素/组件、组件生命周期和v-model详解

使用 Swift 实现简单易懂的异步编程 async/await

Flutter中的异步编程：线程、任务队列、async/await与isolate深入解析

深入理解：setTimeout在同步、异步和回调中的执行顺序详解与经典闭包解析

打造和操控机器人：用Python编程基础入门1.1 详解什么是机器人