快捷導(dǎo)航

python多線程并發(fā)測試過程

更新時間：2025年05月27日 15:25:35 作者：姑娘別禿頭

這篇文章主要介紹了python多線程并發(fā)測試過程,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教

一、并發(fā)與并行？

1、多任務(wù)概念：操作系統(tǒng)可以同時運行多個任務(wù)。

2、并發(fā)：任務(wù)數(shù)量多于cpu核數(shù)，通過操作系統(tǒng)的任務(wù)調(diào)度算法，多個任務(wù)可以一起執(zhí)行，實際總有一些任務(wù)不在一起執(zhí)行，因為切換任務(wù)的速度相當(dāng)快。宏觀上是一起執(zhí)行，微觀上不是一起執(zhí)行的。

3、并行：指的是任務(wù)數(shù)量小于cpu核數(shù)，即任務(wù)真的是一起執(zhí)行。

??多線程都是并發(fā)的狀態(tài)，缺點是比并行慢

二、同步與異步的概念？

1、同步：協(xié)同步調(diào)

def work():
    print("1")
def work2():
    print("2")
if __name__ == '__main__':
    work()
    work2()

2、異步：步調(diào)各異

三、線程與進程的區(qū)別？

1、線程被包含在進程內(nèi)，一個程序啟動時，先啟動進程，進程調(diào)用線程執(zhí)行任務(wù)。

2、線程依賴進程。

3、線程可以被搶占（中斷）

需求1：多線程執(zhí)行不同任務(wù)

1、函數(shù)

函數(shù)式：調(diào)用thread模塊中的start_new_thread()函數(shù)來產(chǎn)生新線程：

thread.start_new_thread ( function, args[, kwargs] )

參數(shù)說明：

function - 線程函數(shù)。
args - 傳遞給線程函數(shù)的參數(shù),他必須是個tuple類型。
kwargs - 可選參數(shù)

示例1:主線程不等待子線程

import time
import threading as td

#為線程定義一個函數(shù)

def asr_test1(url):
    print("start")
    print(time.ctime(time.time()))


def asr_test2(url):
    print("end")
    print(time.ctime(time.time()))

if __name__ == '__main__':
#target為函數(shù)名，args為元組，函數(shù)一定要輸入形參
    t1=td.Thread(target=asr_test1,args=("",))
    t2 = td.Thread(target=asr_test1, args=("",))
    t1.start()
    t2.start()
結(jié)果：
start
Fri Aug  5 16:03:05 2022
start
Fri Aug  5 16:03:05 2022
示例2、主線程等待子線程執(zhí)行完畢
import threading as th
def work():
    print("1")
def work2():
    print("2")


#原理：啟動時會調(diào)用python進程，進程調(diào)用主線程。
if __name__ == '__main__':
    '''子線程啟動'''
    start_time = time.time()
    t1=th.Thread(target=work)
    t2 = th.Thread(target=work2)
    t1.start()
    t2.start()
    #主線程不會等待
    # t1.join(5)#設(shè)置主線程等待子線程的執(zhí)行時間
    end = time.time() - start_time
    print(end)

thread提供了低級別的、原始的線程以及一個簡單的鎖。

Thread類提供以下方法：

run(): 用以表示線程活動的方法。
start():啟動線程活動。
join([time]): 等待至線程中止。這阻塞調(diào)用線程直至線程的join() 方法被調(diào)用中止-正常退出或者拋出未處理的異常-或者是可選的超時發(fā)生。
isAlive(): 返回線程是否活動的。
getName(): 返回線程名。
setName(): 設(shè)置線程名。

2、類包裝線程對象

需求2：多線程執(zhí)行相同任務(wù)

1.threading并發(fā)性

線程并發(fā)運行并共享內(nèi)存。

代碼如下（重寫run方法，繼承Thread類）：

import time
from threading import Thread
'''
Thread其他初始化參數(shù)：
target:指定任務(wù)函數(shù)
name:指定線程分組
'''
class MyThread(Thread):
    
    def run(self):
        for i in range(100):
            time.sleep(0.1)
            print("請求第{}次".format(self.name,i))


if __name__ == '__main__':
    for i in range(5):
        t=MyThread(name=f"線程{i+1}")
        t.start()

傳遞參數(shù)一：

def work(name,age):
    for i in range(5):
        # time.sleep(1)
        print("{}work1----{}---{}".format(name,i,age))


if __name__ == '__main__':
    #給任務(wù)函數(shù)傳遞參數(shù)，方式一：args參數(shù)，里面是元組
    # t=Thread(target=work,args=("",))
    # t.start()
    #方式2：kargs
    t2 = Thread(target=work, kwargs={"name":"wupig","age":18})
    t2.start()

傳遞參數(shù)：繼承類

2.多線程并發(fā)—資源共享，資源競爭問題：

1、無法并行，原因是python解釋器有一把鎖：GIL，在同一時間只能執(zhí)行一個線程。多線程是并發(fā)執(zhí)行的，經(jīng)過python解釋器后進行線程調(diào)度，線程切換機制：

2、線程執(zhí)行遇到IO耗時操作（sleep,網(wǎng)絡(luò)io,文件io），python解釋器釋放鎖，進行線程切換，再次獲取全局鎖。

3、線程執(zhí)行時間達到一定的閾值（面試）

解決方法一：加鎖

解決方法二：隊列

死鎖案例：

需求：

1、并發(fā)讀取本地文件。

2、并發(fā)獲取url地址。

xxx.txt文件中有很多URL地址，需求多并發(fā)獲取。

'''需求：并發(fā)讀取txt中的URL地址'''
list_test=[]
filename="../../url.txt"
file=open(filename,"r")
#列表推導(dǎo)式
list_test=[i.strip("\n") for i in file]
def audio_test():
    while list_test:
        url=list_test.pop()
        print(url)

def atest_thread(THREAD_MAX, Test_Script):
    #創(chuàng)建空列表
    thread_list = []
    for i in range(int(THREAD_MAX)):
        # print("**************",i)
        thread_name = 'thread-' + str(i)
        print("thread_name",thread_name)
        new_thread = threading.Thread(target=Test_Script, args="", name=thread_name)
        thread_list.append(new_thread)
        new_thread.start()
    for thread_obj in thread_list:
        #thread_obj.join：等待至線程中止
        thread_obj.join()

if __name__ == '__main__':
    starttime=time.time()
    count = 10
    # THREAD_MAX=10
    atest_thread(count, audio_test)
    endtime=time.time()-starttime
    print(endtime)

占坑