Python IO密集型任務、計算密集型任務，以及多線程、多進程

對於IO密集型任務：

• 直接執行用時：10.0333秒
• 多線程執行用時：4.0156秒
• 多進程執行用時：5.0182秒

說明多線程適合IO密集型任務。

 

對於計算密集型任務

• 直接執行用時：10.0273秒
• 多線程執行用時：13.247秒
• 多進程執行用時：6.8377秒

說明多進程適合計算密集型任務。

#coding=utf-8
import sys
import multiprocessing
import time
import threading


# 定義全局變量Queue
g_queue = multiprocessing.Queue()

def init_queue():
    print("init g_queue start")
    while not g_queue.empty():
        g_queue.get()
    for _index in range(10):
        g_queue.put(_index)
    print("init g_queue end")
    return

# 定義一個IO密集型任務：利用time.sleep()
def task_io(task_id):
    print("IOTask[%s] start" % task_id)
    while not g_queue.empty():
        time.sleep(1)
        try:
            data = g_queue.get(block=True, timeout=1)
            print("IOTask[%s] get data: %s" % (task_id, data))
        except Exception as excep:
            print("IOTask[%s] error: %s" % (task_id, str(excep)))
    print("IOTask[%s] end" % task_id)
    return

g_search_list = list(range(10000))
# 定義一個計算密集型任務：利用一些復雜加減乘除、列表查找等
def task_cpu(task_id):
    print("CPUTask[%s] start" % task_id)
    while not g_queue.empty():
        count = 0
        for i in range(10000):
            count += pow(3*2, 3*2) if i in g_search_list else 0
        try:
            data = g_queue.get(block=True, timeout=1)
            print("CPUTask[%s] get data: %s" % (task_id, data))
        except Exception as excep:
            print("CPUTask[%s] error: %s" % (task_id, str(excep)))
    print("CPUTask[%s] end" % task_id)
    return task_id

if __name__ == '__main__':
    print("cpu count:", multiprocessing.cpu_count(), "\n")

    print(u"========== 直接執行IO密集型任務 ==========")
    init_queue()
    time_0 = time.time()
    task_io(0)
    print(u"結束：", time.time() - time_0, "\n")

    print("========== 多線程執行IO密集型任務 ==========")
    init_queue()
    time_0 = time.time()
    thread_list = [threading.Thread(target=task_io, args=(i,)) for i in range(10)]
    for t in thread_list:
        t.start()
    for t in thread_list:
        if t.is_alive():
            t.join()
    print("結束：", time.time() - time_0, "\n")

    print("========== 多進程執行IO密集型任務 ==========")
    init_queue()
    time_0 = time.time()
    process_list = [multiprocessing.Process(target=task_io, args=(i,)) for i in range(multiprocessing.cpu_count())]
    for p in process_list:
        p.start()
    for p in process_list:
        if p.is_alive():
            p.join()
    print("結束：", time.time() - time_0, "\n")

    print("========== 直接執行CPU密集型任務 ==========")
    init_queue()
    time_0 = time.time()
    task_cpu(0)
    print("結束：", time.time() - time_0, "\n")

    print("========== 多線程執行CPU密集型任務 ==========")
    init_queue()
    time_0 = time.time()
    thread_list = [threading.Thread(target=task_cpu, args=(i,)) for i in range(10)]
    for t in thread_list:
        t.start()
    for t in thread_list:
        if t.is_alive():
            t.join()
    print("結束：", time.time() - time_0, "\n")

    print("========== 多進程執行cpu密集型任務 ==========")
    init_queue()
    time_0 = time.time()
    process_list = [multiprocessing.Process(target=task_cpu, args=(i,)) for i in range(multiprocessing.cpu_count())]
    for p in process_list:
        p.start()
    for p in process_list:
        if p.is_alive():
            p.join()
    print("結束：", time.time() - time_0, "\n")

參考：https://zhuanlan.zhihu.com/p/24283040