python 测试CPU密集型任务的单进程，多线程和多进程性能

Posted 2021-05-09

#!/usr/bin/python
# -*- coding: UTF-8 -*-
'''
Auther      : Hu Wenchao
Description : Test the time consumed of CPU-bound tasks in single thread,
              multi thread and multi process.

Results run on my Mac Pro:

$ python multicore.py multithread
time is 5.34328889847
$ python multicore.py multiprocess
time is 2.07748579979
$ python multicore.py single
time is 4.0395720005
'''
import time
from multiprocessing import Pool as ProcessPool
from multiprocessing.dummy import Pool as ThreadPool

def sum_prime_below(n):
    '''
    >>> sum_prime_below(1)
    0
    >>> sum_prime_below(2)
    2
    >>> sum_prime_below(3)
    5
    >>> sum_prime_below(4)
    5
    >>> sum_prime_below(5)
    10
    '''
    return sum([i for i in xrange(2, n+1) if is_prime(i)])

def is_prime(n):
    '''
    >>> is_prime(2)
    True
    >>> is_prime(3)
    True
    >>> is_prime(4)
    False
    >>> is_prime(5)
    True
    '''
    for i in xrange(2, n):
        if n % i == 0:
            return False
    return True

if __name__ == '__main__':
    import sys
    start_time = time.time()
    commands = ['single', 'multithread', 'multiprocess']
    try:
        mode = sys.argv[1]
        assert mode in commands
    except (AssertionError, IndexError):
        sys.exit('usage: python multicore.py [ single | multithread | multiprocess ]')
    if mode == 'single':
        for i in range(10):
            sum_prime_below(10000)
    else:
        if mode == 'multithread':
            pool = ThreadPool()
        elif mode == 'multiprocess':
            pool = ProcessPool()
        pool.map(sum_prime_below, [10000]*10)
        pool.close()
        pool.join()
    end_time = time.time()
    print 'time is %s' % (end_time-start_time)