python多線程比單線程效率低的原因及其解決方案

更新時間：2023年08月15日 08:51:52 作者：易爻64

這篇文章主要介紹了python多線程比單線程效率低的原因及其解決方案,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教

python多線程比單線程效率低的原因

Python語言的標準實現(xiàn)叫作CPython，它分兩步來運行Python程序

步驟1：解析源代碼文本，并將其編譯成字節(jié)碼（bytecode）

字節(jié)碼是一種底層代碼，可以把程序表示成8位的指令
從Python 3.6開始，這種底層代碼實際上已經(jīng)變成16位了

步驟2：CPython采用基于棧的解釋器來運行字節(jié)碼。

字節(jié)碼解釋器在執(zhí)行Python程序的過程中，必須確保相關的狀態(tài)不受干擾，
CPython會用一種叫作全局解釋器鎖（global interpreter lock，GIL）的機制來實現(xiàn)運行的python程序的相關狀態(tài)不受干擾

GIL

GIL實際上就是一種互斥鎖（mutual-exclusion lock，mutex），用來防止CPython的狀態(tài)在搶占式的多線程環(huán)境（preemptive multithreading）之中受到干擾，因為在這種環(huán)境下，一條線程有可能突然打斷另一條線程搶占程序的控制權。如果這種搶占行為來得不是時候，那么解釋器的狀態(tài)（例如為垃圾回收工作而設立的引用計數(shù)等）就會遭到破壞。

CPython要通過GIL阻止這樣的動作，以確保它自身以及它的那些C擴展模塊能夠正確地執(zhí)行每一條字節(jié)碼指令。

GIL會產(chǎn)生一個很不好的影響。在C++與Java這樣的語言里面，如果程序之中有多個線程能夠分頭執(zhí)行任務，那么就可以把CPU的各個核心充分地利用起來。盡管Python也支持多線程，但這些線程受GIL約束，所以每次或許只能有一條線程向前推進，而無法實現(xiàn)多頭并進。

所以，想通過多線程做并行計算或是給程序提速的開發(fā)者，恐怕要失望了。

并發(fā) concurrency : 指計算機似乎能在同一時刻做許多不同的事情
并行 parallelism : 指計算機確實能夠在同一時刻做許多不同的事情

多線程下的線程執(zhí)行

獲取GIL
執(zhí)行代碼直到sleep或者是 python虛擬機將其掛起。
釋放 GIL

多線程效率低于單線程原因

如上我們可以知道，在 python中想要某個線程要執(zhí)行必須先拿到 GIL這把鎖，且 python只有一個 GIL，拿到這個 GIL才能進入 CPU執(zhí)行，在遇到 I/O操作時會釋放這把鎖。如果是純計算的程序，沒有 I/O 操作，解釋器會每隔 100次操作就釋放這把鎖，讓別的線程有機會執(zhí)行（這個次數(shù)可以通sys.setcheckinterval來調整）。所以雖然 CPython 的線程庫直接封裝操作系統(tǒng)的原生線程，但 CPython 進程做為一個整體，同一時間只會有一個獲得了 GIL 的線程在跑，其它的線程都處于等待狀態(tài)等著 GIL 的釋放。

而每次釋放 GIL鎖，線程進行鎖競爭、切換線程，會消耗資源。并且由于 GIL鎖存在，python里一個進程永遠只能同時執(zhí)行一個線程 (拿到 GIL的線程才能執(zhí)行 )，這就是為什么在多核 CPU上， python的多線程效率并不高

多線程效率低于或高于單線程原因

相同的代碼，為何有時候多線程會比單線程慢，有時又會比單線程快？這主要跟運行的代碼有關：

CPU密集型代碼(各種循環(huán)處理、計數(shù)等等 )，在這種情況下，由于計算工作多， ticks計數(shù)很快就會達到 100閾值，然后觸發(fā) GIL的釋放與再競爭（多個線程來回切換當然是需要消耗資源的），所以 python下的多線程遇到 CPU密集型代碼時，單線程比多線程效率高。

IO密集型代碼 (文件處理、網(wǎng)絡爬蟲等 )，多線程能夠有效提升效率單線程下有 IO操作會進行 IO等待，造成不必要的時間浪費。開啟多線程能在線程 A等待時，自動切換到線程 B，可以不浪費 CPU的資源，從而能提升程序執(zhí)行效率。進行IO密集型的時候可以進行分時切換所有這個時候多線程快過單線程

如果python想充分利用多核 CPU，可以采用多進程

每個進程有各自獨立的 GIL，互不干擾，這樣就可以真正意義上的并行執(zhí)行。

在 python中，多進程的執(zhí)行效率優(yōu)于多線程 (僅僅針對多核 CPU而言 )。所以在多核 CPU下，想做并行提升效率，比較通用的方法是使用多進程，能夠有效提高執(zhí)行效率

代碼示例：

# 多線程
# 最后完成的線程的耗時
# [TIME MEASURE] execute function: gene_1000_field took 3840.604ms
@time_measure
def mult_thread(rows):
    # 總行數(shù)
    rows = rows
    # 線程數(shù)
    batch_size = 4
    cell = math.ceil(rows / batch_size)
    # 處理數(shù)據(jù)生成
    print('數(shù)據(jù)生成中，線程數(shù):' + str(batch_size))
    threads = []
    for i in range(batch_size):
        starts = i * cell
        ends = (i + 1) * cell
        file = f"my_data_{str(i)}.csv"
        # t = threading.Thread(target=gene_1000_field_test, args=(starts, ends, file))
        t = threading.Thread(target=gene_1000_field, args=(starts, ends, file))
        t.start()
        threads.append(t)
    # for t in threads:
    #     t.join()

# 多進程
# [TIME MEASURE] execute function: gene_1000_field took 1094.776ms
# 執(zhí)行時間和單個線程的執(zhí)行時間差不多，目的達到
@time_measure
def mult_process(rows):
    # 總行數(shù)
    rows = rows
    # 線程數(shù)
    batch_size = 4
    cell = math.ceil(rows / batch_size)
    # 處理數(shù)據(jù)生成
    print('數(shù)據(jù)生成中，線程數(shù):' + str(batch_size))
    process = []
    for i in range(batch_size):
        starts = i * cell
        ends = (i + 1) * cell
        file = f"my_data_{str(i)}.csv"
        # p = Process(target=f, args=('bob',))
        # p.start()
        # p_lst.append(p)
        # t = threading.Thread(target=gene_1000_field_test, args=(starts, ends, file))
        p = Process(target=gene_1000_field, args=(starts, ends, file))
        p.start()
        process.append(p)

python中多線程與單線程的對比

# 做一個簡單的爬蟲：
import threading
import time
import functools
from urllib.request import urlopen
# 寫一個時間函數(shù)的裝飾器
def timeit(f):
    @functools.wraps(f)
    def wrapper(*args,**kwargs):
        start_time=time.time()
        res=f(*args,**kwargs)
        end_time=time.time()
        print("%s函數(shù)運行時間:%.2f" % (f.__name__, end_time - start_time))
        return res
    return wrapper
def get_addr(ip):
    url="http://ip-api.com/json/%s"%(ip)
    urlobj=urlopen(url)
    # 服務端返回的頁面信息, 此處為字符串類型
    pagecontent=urlobj.read().decode('utf-8')
    # 2. 處理Json數(shù)據(jù)
    import json
    # 解碼: 將json數(shù)據(jù)格式解碼為python可以識別的對象;
    dict_data = json.loads(pagecontent)
    print("""
    ip : %s
    所在城市: %s
    所在國家: %s
    """ % (ip, dict_data['city'], dict_data['country']))
#不使用多線程
@timeit
def main1():
    ips = ['12.13.14.%s' % (i + 1) for i in range(10)]
    for ip in ips:
        get_addr(ip)
# 多線程的方法一
@timeit
def main2():
    ips=['12.13.14.%s'%(i+1) for i in range(10)]
    threads=[]
    for ip in ips:
        t=threading.Thread(target=get_addr,args=(ip,))
        threads.append(t)
        t.start()
    [thread.join() for thread in threads]
# 多線程的方法二
class MyThread(threading.Thread):
    def __init__(self, ip):
        super(MyThread, self).__init__()
        self.ip = ip
    def run(self):
        url = "http://ip-api.com/json/%s" % (self.ip)
        urlObj = urlopen(url)
        # 服務端返回的頁面信息, 此處為字符串類型
        pageContent = urlObj.read().decode('utf-8')
        # 2. 處理Json數(shù)據(jù)
        import json
        # 解碼: 將json數(shù)據(jù)格式解碼為python可以識別的對象;
        dict_data = json.loads(pageContent)
        print("""
                            %s
        所在城市: %s
        所在國家: %s
        """ % (self.ip, dict_data['city'], dict_data['country']))
@timeit
def main3():
    ips = ['12.13.14.%s' % (i + 1) for i in range(10)]
    threads = []
    for ip in ips:
        t = MyThread(ip)
        threads.append(t)
        t.start()
    [thread.join() for thread in threads]
if __name__ == '__main__':
    main1()
    main2()
    main3()

---->輸出：
# main1函數(shù)運行時間:55.06
# main2函數(shù)運行時間:5.64
# main3函數(shù)運行時間:11.06

由次可以看出多線程確實速度快了很多，然而這只是適合I/O密集型，當計算密集型中cpu一直在占用的時候，多線程反而更慢。

下面舉例

import threading
import time
def my_counter():
    i = 1
    for count in range(200000000):
        i = i + 2*count
    return True
# 采用單線程
@timeit
def main1():
    thread_array = {}
    for tid in range(2):
        t = threading.Thread(target=my_counter)
        t.start()
        t.join()
# 采用多線程
@timeit
def main2():
    thread_array = {}
    for tid in range(2):
        t = threading.Thread(target=my_counter)
        t.start()
        thread_array[tid] = t
    for i in range(2):
        thread_array[i].join()
if __name__ == '__main__':
    main1()
    main2()