python计算密集型效率对比

一： 需求：

在进行大变量赋值计算的时候， 我发现之前人的代码， 使用了多线程。但是根据我的经验， 计算密集型， 效率一般遵循这样的规律：多进程 > 顺序运行 > 协程 > 多线程。 因此我感觉之前的写法效率不会高。

二：验证

• 计算1～1000数字相加， 并打印结果。

1.1: 顺序计算

# -*- coding: utf-8 -*-
import time
import gevent
from gevent import monkey
from concurrent.futures import ThreadPoolExecutor

monkey.patch_all()
def a_and_b(ab):
    """两数相加"""
    a, b = ab
    return a + b

def normal_test():
    """顺序计算"""
    start_time = time.time()
    results = [a_and_b((i, i - 1)) for i in range(1, 1000)]

    for result in results:
        print(result)
    end_time = time.time()
    print("normal cost time is {}".format(end_time - start_time))


if __name__ == '__main__':
    normal_test()
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

• normal cost time is 0.002415895462036133

1.2: 协程计算

# -*- coding: utf-8 -*-
import time
import gevent
from gevent import monkey
from concurrent.futures import ThreadPoolExecutor

monkey.patch_all()

def a_and_b(ab):
    """两数相加"""
    a, b = ab
    return a + b
    
def coroutine_test():
    """协程测试"""
    start_time = time.time()
    tasks = [gevent.spawn(a_and_b, (i, i - 1)) for i in range(1, 1000)]
    gevent.joinall(tasks)

    for task in tasks:
        print(task.value)
    end_time = time.time()
    print("coroutine cost time is {}".format(end_time - start_time))
    
if __name__ == '__main__':
    coroutine_test()
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

• coroutine cost time is 0.010415792465209961

1.3: 多线程计算

# -*- coding: utf-8 -*-
import time
import gevent
from gevent import monkey
from concurrent.futures import ThreadPoolExecutor

monkey.patch_all()
def a_and_b(ab):
    """两数相加"""
    a, b = ab
    return a + b
    
def threading_test():
    """线程测试"""
    futures = []
    start_time = time.time()
    with ThreadPoolExecutor(max_workers=10) as executor:
        for i in range(1, 1000):
            futures.append(executor.submit(a_and_b, (i, i - 1)))

    for future in futures:
        print(future.result())

    end_time = time.time()
    print("threading cost time is {}".format(end_time - start_time))

if __name__ == '__main__':
    threading_test()
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

• threading cost time is 0.07938933372497559

三：结论

• 果然， 前人给埋的优化点被我发现了， 哈哈， 这个季度绩效有指望了。