匆匆那年

钟声敲在你我耳边

0%

python开发中的速度追求

发布： 分类： 本文字数： 9.7k 阅读时长 ≈ 9 分钟

python中执行速度的技术选择

前言

最近在开发一个扫描的项目，很多时候，发现一个函数的执行比较频繁且缓慢，我不知道应该用多线程、协程、还是多进程 或是消息队列等。

这说明目前开发水平太差，还是有很多需要学习的地方。所以就想着学习一些各个技术，然后简单的做个笔记啊，并总结出什么时候用哪种技术，可以让整个扫描更快。当然目前还是自嗨开发小项目阶段，没做过大项目工程，有见解不对的地方还望看到的人指正。

后面主要介绍多线程、协程、消息队列和异步。

多线程

python一提起多线程，我们搜到的资料基本就会说是鸡肋，是没用的，应该使用协程。以前我也没有好好的深究过这个问题，现在我们来看下python的多线程到底有没有用？

首先网上的资料都会说到GIL的问题导致python多线程很鸡肋，那什么是GIL锁？

GIL锁

GIL的存在主要是为了防止Python解释器中的数据结构被多个线程同时修改，导致数据结构出现不一致的情况。通过限制同一时刻只有一个线程能够执行Python字节码，GIL可以确保Python解释器中的数据结构不会被多个线程同时修改，从而保证线程安全。

测试

我们写一个多线程和普通执行的测试，都执行5000次计算，一个单线程执行，一个多线程执行：

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
35
36
37
# 打印随机数的函数
def print_random():
    for i in range(500):
        result = 9999999919+5678987656
        # response = requests.get("https://www.baidu.com")
        # print(response)
        

# 创建线程并启动
def create_threads():
    start_time = time.time()
    threads = []
    def ppdom():
        result = 9999999919+5678987656
        # response = requests.get("https://www.baidu.com")
        # print(response)
    for i in range(500):  # 创建500个线程
        thread = threading.Thread(target=ppdom)
        thread.start()
    for thread in threads:
        thread.join()
    end_time = time.time()
    execution_time = end_time - start_time
    print("多线程执行耗时：", execution_time)

def create_singal():
    start_time_signal = time.time()
    print_random()
    end_time_signal = time.time()
    print(start_time_signal)
    print(end_time_signal)
    execution_time1 = end_time_signal - start_time_signal
    print("单线程执行耗时：", execution_time1)
# 调用函数
if __name__ == '__main__':
    create_singal()
    create_threads()

执行多次发现，多线程比单线程的耗时更长(注意这里单线程是e-05次方)：

![image-20230904140830892](/Users/geez/Library/Application Support/typora-user-images/image-20230904140830892.png)

这真的很离谱，多线程好像还不如单线程呢，完全没必要用到多线程。难道真的就是网上说的啥时候都别用多线程就对了吗？

那再看看下面这个代码：

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
35
36
37
38
39
40
41
42
import threading
import random
import time
import requests



## 多线程和单线程的比较

# 打印随机数的函数
def print_random():
    for _ in range(5):
        # result = 99+100
        response = requests.get("https://www.baidu.com")
        print(response)
        

# 创建线程并启动
def create_threads():
    start_time = time.time()
    threads = []
    def ppdom():
        # result = 99+100
        response = requests.get("https://www.baidu.com")
        print(response)
    for _ in range(5):  # 创建5个线程
        thread = threading.Thread(target=ppdom)
        thread.start()
    end_time = time.time()
    execution_time = end_time - start_time
    print("多线程执行耗时：", execution_time)

def create_singal():
    start_time = time.time()
    print_random()
    end_time = time.time()
    execution_time = end_time - start_time
    print("单线程执行耗时：", execution_time)
# 调用函数
if __name__ == '__main__':
    create_threads()
    create_singal()

这里我们仅修改函数从原来的简单计算改为网络请求，运行结果如下：

![image-20230901172004526](/Users/geez/Library/Application Support/typora-user-images/image-20230901172004526.png)

同样执行五次请求，这时候我们可以发现多线程远远快于单线程，这是为什么呢？

因为耗时的IO应用（内存、网络、硬盘等等读取属于IO密集型运算）并不影响线程的执行，也就是cpu和IO是可以并行的。由于上面的代码中 request.get() 一个 I/O 密集型操作，即主要花费时间在等待网络 I/O 操作完成上。由于线程是独立的，一个线程在等待网络响应时，其他线程可以继续执行，从而最大程度上减少了阻塞等待的时间。所以我们也能看到速度飞快。

结论

由此我们可以得出结论，在GIL锁存在的情况下，如果程序是IO密集型的，那么多线程可以极大的加快速度。但如果程序不是 I/O 密集型时，多线程可能并不能提高性能，甚至还可能因为线程间的上下文切换导致额外的开销。另外由于GIL锁的存在，即使你的CPU多核，也很难发挥作用。所以多线程的运用与否主要考虑是否是IO密集型的操作。

协程

上面说到了多线程，现在再来看看协程是怎么回事。

概念
Python 3 引入了协程（coroutine）的概念，通过使用关键字 async 和 await 实现。协程是一种轻量级的并发编程方式，支持在单线程中实现并发执行的效果。

下面是Python 3中协程的一些重要特点和概念：

  • 协程与生成器（generator）：协程使用async def定义，并通过await关键字来挂起执行和等待其他协程完成。协程本质上是一种特殊的生成器，利用生成器的yield关键字进行状态保存和恢复。
  • async 和 await关键字：async关键字用于定义协程函数，表示函数是一个可以被挂起和恢复的协程。await关键字用于挂起协程的执行，等待另一个协程完成或者等待I/O操作完成。
  • 非阻塞的等待：协程可以在等待I/O完成时挂起执行，允许其他协程继续执行。这种非阻塞的等待方式使得协程能够充分利用CPU和I/O资源，提高并发性能。
  • 事件循环（event loop）：Python 3的协程通常会在事件循环中执行。事件循环负责调度多个协程，使它们能够交替执行和等待，以实现并发效果。
  • 异步I/O支持：协程在处理I/O密集型任务时特别有效。常用的异步I/O库例如asyncio提供了高级的异步编程机制，用于处理协程之间的调度和协作。

测试
我们再次编写一个测试函数，分别用到了协程、多线程和单线程，分别执行50次的请求百度的运算：

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
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

import requests
import threading
import asyncio
import time

# 普通执行版本
def normal_execution():
    start_time = time.time()
    for _ in range(50):
        # result = 99+1000
        response = requests.get("https://www.baidu.com")
        # print(response)
    end_time = time.time()
    execution_time = end_time - start_time
    print("普通执行耗时：", execution_time)

# 多线程执行版本
def threaded_execution():
    start_time = time.time()
    threads = []

    def thread_function():
        # result = 99+1000
        response = requests.get("https://www.baidu.com")
        # print(response)

    for _ in range(50):
        thread = threading.Thread(target=thread_function)
        threads.append(thread)
        thread.start()

    for thread in threads:
        thread.join()

    end_time = time.time()
    execution_time = end_time - start_time
    print("多线程执行耗时：", execution_time)

async def fetch(session):
    url = 'https://www.baidu.com'
    async with session.get(url) as response:
        return await response.text()

async def request_baidu():
    async with aiohttp.ClientSession() as session:
        tasks = []
        for _ in range(50):
            task = asyncio.ensure_future(fetch(session))
            tasks.append(task)
        responses = await asyncio.gather(*tasks)



# 比较函数
def compare_execution_time():
    print("开始普通执行...")
    normal_execution()

    print("\n开始多线程执行...")
    threaded_execution()

    print("\n开始协程执行...")
    start_time = time.time()
    loop = asyncio.get_event_loop()
    results = loop.run_until_complete(request_baidu())
    end_time = time.time()
    execution_time = end_time - start_time
    print("协程执行耗时：", execution_time)

# 调用比较函数
if __name__ == '__main__':
    compare_execution_time()

执行结果如下：

![image-20230904141410552](/Users/geez/Library/Application Support/typora-user-images/image-20230904141410552.png)

可以看到协程速度飞快，

我们再换成IO非密集操作仅作算数计算，同样都计算500次：

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
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
import requests
import threading
import asyncio
import time

# 普通执行版本
def normal_execution():
    start_time = time.time()
    for _ in range(500):
        result = 99999+54673829
        # response = requests.get("https://www.baidu.com")
        # print(response)
    end_time = time.time()
    execution_time = end_time - start_time
    print("普通执行耗时：", execution_time)

# 多线程执行版本
def threaded_execution():
    start_time = time.time()
    threads = []

    def thread_function():
        result = 99999+54673829
        # response = requests.get("https://www.baidu.com")
        # print(response)

    for _ in range(500):
        thread = threading.Thread(target=thread_function)
        threads.append(thread)
        thread.start()

    for thread in threads:
        thread.join()

    end_time = time.time()
    execution_time = end_time - start_time
    print("多线程执行耗时：", execution_time)

async def fetch(session):
    result = 99999+54673829

async def request_baidu():
    async with aiohttp.ClientSession() as session:
        tasks = []
        for _ in range(500):
            task = asyncio.ensure_future(fetch(session))
            tasks.append(task)
        responses = await asyncio.gather(*tasks)



# 比较函数
def compare_execution_time():
    print("开始普通执行...")
    normal_execution()

    print("\n开始多线程执行...")
    threaded_execution()

    print("\n开始协程执行...")
    start_time = time.time()
    loop = asyncio.get_event_loop()
    results = loop.run_until_complete(request_baidu())
    end_time = time.time()
    execution_time = end_time - start_time
    print("协程执行耗时：", execution_time)

# 调用比较函数
if __name__ == '__main__':
    compare_execution_time()


执行结果：

![image-20230904141620788](/Users/geez/Library/Application Support/typora-user-images/image-20230904141620788.png)

可以看到协程速度虽然比多线程快，但还不如单线程，这就又回到了前面那样，在计算密集型计算中协程和多线程都是弟弟水平。但是协程度又确实比多线程快，那是不是以后需要用到多线程的地方直接用协程就行？

当然不是，如果注意观察可以看到上述代码中请求百度的库，别的代码使用的是requests，而协程使用的是aiohttp，这是因为requests不支持异步，也就没办法支持协程。另外协程的编程复杂度也高于多线程。所以很多时候在协程库不支持或是协程复杂度实现不能接受的时候，可以考虑使用多线程。

结论：

CPU密集型操作协程速度大于多线程但不如单线程，IO密集型操作协程远远大于多线程和单线程。

多进程

上面的比较又引出一个新的问题，CPU密集型的运算中 难道就只能使用单线程吗？

当然不是，这里我们引入多进程的概念：

  • Python中的多进程是指同时运行多个独立进程的能力。每个进程都有自己独立的内存空间和资源，可以并行地执行任务。这种并行和独立的特性使得多进程在处理计算密集型任务和提高CPU利用率方面非常有效。Python提供了multiprocessing模块来支持多进程编程

测试CPU密集型计算:

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
35
36
37
38
39
40
41

import time
import multiprocessing

def cpu_intensive_calculation():
    total = 0
    for i in range(20000000):
        total += i
    return total

def single_threaded():
    start_time = time.time()

    for i in range(8):
        cpu_intensive_calculation()

    end_time = time.time()
    elapsed_time = end_time - start_time
    print(f"单线程执行耗时: {elapsed_time} seconds")
def multi_process1():
    start_time = time.time()

    num_processes = multiprocessing.cpu_count()  # 获取CPU核心数量 8个
    print(num_processes)
    num_tasks = 5000

    processes = []
    for _ in range(num_processes):
        process = multiprocessing.Process(target=cpu_intensive_calculation)
        processes.append(process)
        process.start()

    for process in processes:
        process.join()
    end_time = time.time()
    elapsed_time = end_time - start_time
    print(f"多进程执行耗时: {elapsed_time} seconds")

if __name__ == '__main__':
    single_threaded()
    multi_process1()

执行结果如下

![image-20230904143720227](/Users/geez/Library/Application Support/typora-user-images/image-20230904143720227.png)

可以看到 多进程在CPU密集型计算中会更有优势一点。

消息队列

其实协程中就有异步的概念，消息队列也有点异步的意思，严格来说消息队列和线程、进程不是一类技术，它是为了解决另一个问题：解耦、异步、削峰。

解耦：比如打卡系统和考勤系统是关联的，那么如果中间加入消息队列，打卡系统就只需要把打卡的数据放入消息队列由考勤系统去取就行。不必关心这两者系统的关联。

异步：多人打卡时可能每个打卡到都需要记录到考勤系统，但不可能每个人都是打完卡记录完以后才能下一个人打开，消息队列可以异步处理。简单说就是 打卡时打卡，考勤记录是考勤记录，中间由消息队列提供缓存和消息传递。

削峰：并发访问高峰期，可以缓存到中间件中减少对其他任务的压力。比如发送邮件是耗时应用，每天定期发送的邮件可能有几封，但是一家公司可能有上千人，这时候发送邮件就可以进入消息队列，然后再慢慢发送，防止流量过载。

消息队列的使用demo：

1、安装celery和redis

1
pip install celery

2、创建使用celery的应用

1
2
3
4
5
6
7
8
9
10
from celery import Celery

# 创建 Celery 应用
app = Celery('myapp', broker='redis://localhost:6379/0')

# 定义任务
@app.task
def add(x, y):
    return x + y

3、启动celry服务端，也就是启动消息队列：

1
celery -A tasks worker --loglevel=info

4、其他函数调用消息队列

1
2
3
4
from tasks import add

result = add.delay(4, 6)
print(result.get())

什么时候使用？耗时应用需要解耦、异步、削峰处理时使用此技术。

比如一个爬虫，它需要爬去、解析内容、记录到数据库。

那每个任务都可以写成多线程，然后进入消息队列，下一个任务直接从消息队列中取并继续压入消息队列：

怎么选择

如果是CPU密集型任务，那当然直接多进程。

如果是IO密集型任务，就有两种选择多线程和协程，如果协程库支持不完善，代码复杂度超出能力。那就选择多线程，否则选择协程，毕竟协程是新的技术，应该有更好的速度表现。

当需要 解耦、异步、削峰 时我们考虑消息队列并结合上述的技术进行优化。

![image-20230905153811141](/Users/geez/Library/Application Support/typora-user-images/image-20230905153811141.png)

备注

需要注意的是进程、线程、协程不是并列关系，他们是包含关系。进程包含线程包含协程。也就是说一个进程可以有多个线程一个线程可以有多个进程。

由于本身不是程序员，对代码理解有限，如果文中有错误的地方欢迎评论区指出。


采用署名-非商业性使用-相同方式共享 4.0（CC BY-NC-SA 4.0）许可协议
「分享也是一种学习」