python并发编程：协程asyncio、多线程threading、多进程multiprocessing_asyncio threading

CPU密集型计算与IO密集型计算

• CPU密集型计算：也叫计算密集型，是指IO在很短的时间内就可以完成，CPU需要大量的计算与处理，特点是CPU占用率高。如：压缩解压缩、加密解密、正则表达式
• IO密集型计算：指的是系统运作的大部分状况是CPU在等I/O（硬盘/内存）的读写操作，CPU占用率较低。如：网络爬虫、读写数据库

多线程、多进程与协程的对比

多线程

创建多线程的方法

1. 准备一个函数：

def my_func(a,b):
	do_something(a,b)
1
2

2. 创建一个线程：

import threading
t = threading.Thread(target = my_func,args = (100,200))
1
2

3. 启动线程：t.start()
4. 等待结束：t.join()

多线程实现的生产者-消费者爬虫

1. 多线程数据通信的queue.Queue

• 导入类库：import queue
• 创建Queue：q = queue.Queue()
• 添加元素：q.put(item)
• 获取元素：item = q.get()
• 查询元素的多少：q.qsize()
• 判断是否为空：q.empty()
• 判断是否已满：q.full()

2. 实现案例代码：

import threading
import requests
import queue

def do_crawl(url_queue,html_queue):
    while True:
        url = url_queue.get()
        html = requests.get(url)
        html_queue.put(html)
        print(threading.current_thread().name,f'crawl{url},url_queue.size={url_queue.qsize()}')

def do_parse(html_queue,fout):
    while True:
        html = html_queue.get()
        results = html.text
        for result in results:
            fout.write(str(result) + '\n')
        print(threading.current_thread().name,f'results.size={len(results)},html_queue.size={html_queue.qsize()}')

url_queue = queue.Queue()
html_queue = queue.Queue()
for url in url_list:
    url_queue.put(url)

#启动三个生产者线程
for idx in range(3): 
    t = threading.Thread(target=do_crawl,args=(url_queue,html_queue),name=f'crawl {idx}')
    t.start()

fout = open('test.txt','w')
#启动两个消费者线程
for idx in range(2):
    t = threading.Thread(target=do_parse,args=(html_queue,fout),name=f'parse {idx}')
    t.start()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

Lock解决线程安全问题

1. try-finally模式

import threading
lock = threading.Lock()
lock.acquire()
try:
	#do something
finally:
	lock.release()
1
2
3
4
5
6
7

2. with模式

import threading
lock = threading.Lock()
with lock:
	#do something
1
2
3
4

使用线程池ThreadPoolExecutor

1. 使用线程池的好处

• 提升性能：减去了大量新建、终止线程的开销，重用了线程资源
• 适用场景：处理突发性大量请求或需要大量线程完成任务，但实际任务处理时间较短
• 防御功能：能有效避免系统因为创建线程过多而导致系统负荷过大相应变慢等问题
• 代码优势：使用线程池的语法比自己新建线程执行线程更加简洁

2. ThreadPoolExecutor的使用方法

• 用法一：map函数，较简单，需要注意map的结果和入参是顺序对应的

from concurrent.futures import ThreadPoolExecutor,as_completed

with ThreadPoolExecutor() as pool:
    results = pool.map(crawl,urls)

    for result in results:
        print(result)
1
2
3
4
5
6
7

• 用法二：future模式，更强大，需要注意如果用as_completed顺序是不定的

from concurrent.futures import ThreadPoolExecutor,as_completed

with ThreadPoolExecutor() as pool:
    futures = [pool.submit(crawl,url) for url in urls]

    for future in futures:
        print(future.result())
    
    for future in as_completed(futures):
        print(future.result())
1
2
3
4
5
6
7
8
9
10

多进程

多进程multiprocessing对比多线程threading

多进程的使用方法和多线程比较类似，可以参考多线程的代码。

协程

协程简介

1. 协程的核心原理：用一个超级循环（其实就是while true），配合IO多路复用（IO时CPU可以做其他事情）。
2. 协程的简单使用：

import asyncio

#获取事件循环
loop = asyncio.get_event_loop()
#定义协程
async def myfunc(url):
	await get_url(url)
#创建task列表
tasks = [loop.create_task(myfunc(url)) for url in urls]
#执行爬虫事件列表
loop.run_until_complete(asyncio.wait(tasks))
1
2
3
4
5
6
7
8
9
10
11

控制并发

1. 信号量（Semaphore）是一个同步对象，用于保持在0至指定最大值之间的一个计数值

• 当线程完成一次对该semaphore对象的等待（wait）时，该计数值减1
• 当线程完成一次对该semaphore对象的释放（release）时，该计数值加1
• 当计数值为0，则线程等待该semaphore对象不再能成功直至该semaphore对象变成signaled状态
• semaphore对象的计数值大于0，为signaled状态；计数值等于0，为nonsignaled状态

2. 使用方式

• 用法1

sem = asyncio.Semaphore(10)
async with sem:
	#work with shared resourse
1
2
3

• 用法2

sem = asyncio.Semaphore(10)
await sem.acquire()
try:
	#work with shared resourse
finally:
	sem.release()
1
2
3
4
5
6

await关键字

根据官方文档说明，await 后面的对象必须是如下格式之一：

1. A native coroutine object returned from a native coroutine function，一个原生 coroutine 对象。
2. A generator-based coroutine object returned from a function decorated with types.coroutine()，一个由 types.coroutine() 修饰的生成器，这个生成器可以返回 coroutine 对象。
3. An object with an await__ method returning an iterator，一个包含 __await 方法的对象返回的一个迭代器。

简而言之，在await后面只能跟三种可等待的对象：协程对象、future对象、task对象

要实现异步处理，我们得先要有挂起的操作，当一个任务需要等待 IO 结果的时候，可以挂起当前任务，转而去执行其他任务，这样我们才能充分利用好资源。使用 await 可以将耗时等待的操作挂起，让出控制权。当协程执行的时候遇到 await，时间循环就会将本协程挂起，转而去执行别的协程，直到其他的协程挂起或执行完毕。

Task对象

tasks用于并发调度协程，它是对协程对象的进一步封装，包含了任务的各个状态。通过asyncio.creat_task(协程对象)的方式可以创建Task对象，这样可以让协程加入事件循环中等待被调度执行。除了使用asyncio.creat_task()以外，还可以用低层级的loop.creat_task()或asyncio.ensure_future()。
注意：asyncio.creat_task()在python3.7之后加入。在python3.7之前可以改用低层级的asyncio.ensure_future()或loop.create_task()（loop.create_task()是最低层级）。

例1：

import asyncio
import logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - |%(levelname)s| - %(message)s')

async def func():
    logging.info(1)
    await asyncio.sleep(3)
    logging.info(2)
    return '返回值'

async def main():
    print('main start!')
    task1 = asyncio.create_task(func())
    task2 = asyncio.create_task(func())
    print('main end!')

    r1 = await task1
    r2 = await task2

    print(r1,r2)

asyncio.run(main())
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

以上代码的输出为：

代码的分析：
程序内部本质上只有一个线程，在main函数开始之后，到运行到r1 = await task1之前，线程都没有遇到io等待，所以会一瞬间将main函数中的前四行代码执行完毕。
此时算上asyncio.run(main())中注册的main()协程对象，加上task1、task2，相当于在事件循环中有三个任务。当遇到await的时候，会切换到别的任务中执行，所以在main()遇到await即r1 = await task1时，会切换到task1中执行任务，切换到task1中执行任务之后又遇到了io等待，所以又立即切换到了task2中去执行任务。
此时在控制台打印出的效果就是一瞬间实行了task1与task2，此时没有了io等待，所以程序等待task1与task2执行完毕后又回到main()中继续执行任务。

例2：将task放在一个列表中

import asyncio
import logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - |%(levelname)s| - %(message)s')

async def func(x):
    logging.info(1)
    await asyncio.sleep(x)
    logging.info(2)
    return f'返回值{x}'

async def main():
    print('main start!')
    task_list = [
        asyncio.create_task(func(2),name = 'f1'),
        asyncio.create_task(func(5),name = 'f2')
    ]
    print('main end!')
    
    #写法1，使用asyncio.wait
    # done,pending = await asyncio.wait(task_list,timeout = None)
    # print(done)
    # 此处打印的是一个Task finished对象的集合
    
    #写法2，使用asyncio.gather，asyncio.gather依照协程执行的先后顺序来返回一个列表
    # t = await asyncio.gather(*task_list)
    # print(t)  
    # 此处打印：['返回值2', '返回值5']
asyncio.run(main())
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

例3：一种错误的写法

import asyncio
import logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - |%(levelname)s| - %(message)s')

async def func(x):
    logging.info(1)
    await asyncio.sleep(x)
    logging.info(2)
    return f'返回值{x}'

task_list = [
    asyncio.create_task(func(2),name = 'f1'),
    asyncio.create_task(func(5),name = 'f2')
]
done, pending = asyncio.run(asyncio.wait(task_list))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

这段代码会报错：no running event loop
原因在于，使用asyncio.create_task的时候，事件循环还没有创建，所以会报错。
此时我们需要改写task_list成这样：

task_list = [
    func(2),
    func(5)
]
1
2
3
4

在task_list里面放协程对象，而不是放task对象，改成这样就可以正常运行了。原因在于在asyncio.run的内部其实会帮你创建一个事件循环，然后这个事件循环中，你传进去的协程对象，它会在内部帮你创建成task对象，就解决了事件循环还没有创建而报错的问题。
建议还是按照例2的方式，将asyncio.creak_task放在一个main()的协程对象里就行了。

Future对象

Future：代表将来执行或没有执行的任务的结果，Future 相较于Task属于更底层的概念。Future，又称 未来对象、期程对象，其本质上是一个容器，用于接受异步执行的结果。
Furture 对象内部封装了一个 _state，这个 _state 维护着四种状态：Pending、Running、Done，Cancelled，如果变成 Done 完成，就不再等待，而是往后执行，这四种状态的存在其实类似与进程的运行态、就绪态、阻塞态，事件循环凭借着四种状态对 Future\协程对象 进行调度。
我们前面讲的 Task 是继承自 Future，Task对象内部await结果的处理是基于Future对象来的。

例1

import asyncio

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

    # 创建一个任务(Future对象,这个任务什么都不干)
    fut = loop.create_future()

    # 等待任务最终结果(Future对象),没有结果则会一直等下去
    await fut

asyncio.run( main() )
1
2
3
4
5
6
7
8
9
10
11
12
13
14

上面代码会一直运行，不会停止。

例2

import asyncio
import logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - |%(levelname)s| - %(message)s')

async def set_after(fut):
    logging.info('enter set_after')
    await asyncio.sleep(3)
    fut.set_result('666')

async def main():
    logging.info('enter 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))        #此处分析一下是否加await的区别

    logging.info('get data')
    #等待Future对象获取最终结果,否则一直等下去
    data = await fut
    logging.info(f'fut的返回值为{data}')

asyncio.run(main())
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

上述代码中，在loop.create_task(set_after(fut))前加了await后的输出结果为:

分析如下：
首先进入main()协程对象，打印enter_main，然后

creak_task与ensure_future的异同

asyncio.wait与asynio.gather的异同

asynio.gather()中的参数可以是task或协程对象,并且得到其列表形式的返回值。其中如果传入的是协程对象，那么首先gather会先把这些协程对象转换为task，

event loop

event loop相当于一个大脑，在下面有若干的task，task是没有办法去控制event loop去执行某一个task的，它只能告诉event loop说我在等这个task，最终是由event loop去决定下面要运行哪一个task。而event loop一旦开始运行task，就必须要task显式地把控制权交还给event loop，交还控制权的方式有await和函数运行完毕。所以尽管我们说协程的方式是并发的，但实际上同一时刻只有一段代码在运行，它只是想办法利用了代码中间等待的时间，所以如果我们的代码里面没有等待的话，协程的帮助并不大。

参考文章

Python：协程中Task和Future的理解及使用
Python爬虫进阶、异步协程
Python异步编程视频讲解
【python】asyncio的理解与入门，搞不明白协程？看这个视频就够了。
【2021最新版】Python 并发编程实战，用多线程、多进程、多协程加速程序运行
Python中协程异步IO（asyncio）详解
深入理解asyncio(二)
深入理解asyncio(一)