原理:用Python代碼模擬實現性能測試中的集合點
作者:強官濤   類型:Python開發    類別:原理   日期:2019-09-19    閱讀:273 次   消耗積分:0 分


小伙伴們,大家好~

今天的這篇文章來自蝸牛學院在線教育部卿淳俊老師。


蝸牛學院資深講師,15年測試開發及項目管理經驗,認證敏捷項目Scrum Master,精通各種測試及開發工具、框架的設計和實現,在移動互聯網、敏捷項目管理、電商測試等領域有豐富的項目經驗。

對大數據、機器學習、區塊鏈等新興技術均有深入研究。6年講師經驗鑄就了深入淺出的授課風格,秉持著“授人以魚不如授人以漁”的理念,讓學員達到“知其然更要知其所以然”,真正掌握技術的本質。



·  正  ·  文  ·  來  ·  啦  ·



做過性能測試的童鞋應該都知道集合點這個概念,也就是在多線程的環境中,我們通過設置集合點,使得這些創建好的線程并不立即執行,而是讓所有線程等待到某一個點之后再統一一起執行,這樣做的目的是對服務器造成真正的并發壓力。在性能測試工具Jmeter或LR中可以輕松地做到這點,但如果是編寫純Python腳本的話,該如何實現多線程的類似集合點這樣的并發請求呢?


其實要在Python中實現多線程的集合也是比較容易的,我們主要要用到線程同步方法中的Event來實現。Event是在threading模塊中的一個類,這個事件類可以通過設置一個事件信號,讓所有設置了該事件的線程暫停等待或激活執行,這個類主要有三個方法,大家可以從代碼注釋中看到這些方法的作用,非常簡單:

event = threading.Event()


# 重置event,使得所有該event事件都處于待命狀態

event.clear()

# 等待接收event的指令,決定是否阻塞程序執行
event.wait()

# 發送event指令,使所有設置該event事件的線程執行
event.set()


知道了這些方法,我們就可以利用這些方法達到模擬集合點目的。廢話不多說,直接上代碼:


import threading


class MyThread(threading.Thread):
   def __init__(self, event):
       super().__init__()
       self.event = event
   

  def run(self):
    print("線程{}已初始化完成,隨時準備啟動....".format(self.name))
       # 設置線程等待
       self.event.wait()
       print("{}開始執行...".format(self.name))

if __name__ == '__main__':
   

   event = threading.Event()
   threads = []

 [threads.append(MyThread(event)) for i in range(1, 11)]
   # 必須在子線程start之前先清空所有的event設置,讓子線程的event.wait生效
   event.clear()
   [t.start() for t in threads]
   # 設置event事件,事件設置后將通知所有設置了事件對象的線程激活
   event.set()
   [t.join() for t in threads]


在上面這段中,我們在主線程中新建了10個子線程,并且在每個線程方法中都設置了線程等待語句,這樣當線程完成初始化進入run方法后,首先會打印出“線程xx已初始化完畢,隨時準備啟動..."的輸出語句,隨后就進入等待狀態,后面的語句將不會執行,直到在主線程中運行了event.set()方法,通知所有的線程進入激活狀態之后,所有的線程才開始真正執行。

代碼運行后,結果如下。從結果中我們可以看到,所有的線程確實是先生成好后,然后再統一啟動,成功實現了集合點的效果。


20190919_105548_380.jpg



為了答謝大家對蝸牛學院的支持,蝸牛學院將會定期對大家免費發放干貨,敬請關注蝸牛學院的官方微信。


20190320_095757_834.jpg




版權所有,轉載本站文章請注明出處:蝸牛學院在線課堂, http://www.bnwkbz.icu/note/380
上一篇: 資訊:從180斤到130斤,從月薪2.6K到13K,這是他的個人分享!
下一篇: 實驗:我是如何在2小時內,成功滲透蝸牛學院Boss系統的?
提示:登錄后添加有效評論可享受積分哦!
? 下载重庆时时