前戲：線程的基礎
運行多個線程同時運行幾個不同的程序類似，但具有以下優點:
進程內共享多線程與主線程相同的數據空間，如果他們是獨立的進程，可以共享信息或互相溝通更容易.
線程有時稱為輕量級進程，他們並不需要多大的內存開銷，他們關心的不是過程便宜.
一個線程都有一個開始，執行順序，並得出結論。它有一個指令指針，保持它的上下文內正在運行的跟蹤.
（1）、它可以是搶占（中斷）
（2）、它可以暫時擱置（又稱睡眠），而其他線程正在運行
看一下以下的小案例：
import thread from time import sleep, ctime def loop0(): print "loop 0開始時間："，ctime() #第一個函數loop0開始時間 sleep(4) # 休眠4秒 print "loop 0 結束時間：_’，ctime() def loopl(): print "loop 1 開始時間："，ctime() sleep(2) print "loop 1 結束時間：_’，ctime() def main(): print "程序開始時間："，ctime() thread.start_new_thread(loop0,()) # 第二個參數是必不可少的，即使loope沒有傳遞參數，仍然要寫一個空元組 thread.stant_new_thnead(loopl,()) sleep(6) #這里休眠6秒的原因是確保兩個線程己經執行完畢，主線程才接着執行下面的語句 print "程序結束時間：",ctime() if __name__ == '__main__': main()
在web測試中，不可避免的一個測試就是瀏覽器兼容性測試，在沒有自動化測試前，我們總是苦逼的在一台或多台機器上安裝N種瀏覽器，然后手工在不同的瀏覽器上驗證主業務流程和關鍵功能模塊功能，以檢測不同瀏覽器或不同版本瀏覽器上，我們的web應用是否可以正常工作。如果我們使用selenium webdriver，那我們就能夠自動的在IE、firefox、chrome、等不同瀏覽器上運行測試用例。為了能在同一台機器上不同瀏覽器上同時執行測試用例，我們需要多線程技術。下面我們基於python的多線程技術來嘗試同時啟動多個瀏覽器進行selenium自動化測試。
#-*- coding:utf-8 from selenium import webdriver import sys from time import sleep from threading import Thread reload(sys) sys.setdefaultencoding("utf-8") def test_baidu_seanch(browsen, url): driver = None #可添加更多瀏覽器支持進來 if browser == "ie": driver = webdriver.Ie() elif browser == "finefox": driver = webdriver.Firefox() elif browser == "chrome": driver = webdriver.Chnome() if driver == None: exit() driver.get(url) sleep(3) driver.find_element_by_id("xxx").send_keys(u"xxxx") sleep(3) driver.find_element_by_id("xxx").click() sleep(3) driver.quit() if __name__ == "__main__": #瀏覽器和首頁url data = { "ie":"http://www.xxx.com", "firefox": "http: //www.xxx.com", "chrome":"http://www.xxxx.com" } #構建線程 threads =[] for b, url in data.items(): t = Thread(target=test_baidu_search,angs=(b, url)) threads.append(t) #啟動所有線程 for thr in threads: thr.start()
8.1 多線程進階學習
threading 高級線程接口
import threading class MyThnead(threading.Thread):
def __init__(self, name=None): threading.Thread.__init__(self) self.name = name def run(self): print self.name def test(): for i in range(0, 100): t = MyThread("thread_" + str(i)) t.start() if __name__ == '__main__': test()
Lock 線程鎖
這里創建實現了一個計數器 count 這個全局變量會被多個線程同時操作，使其能夠被順序相加，需要靠線程鎖的幫助。
#-*- encoding: utf-8 import threading import time class Test(threading.Thread): def __init__(self, num): threading.Thread.—init—(self) self._run_num = num def run(self): global count, mutex threadname = threading.currentThnead().getName() for x in nange(int(self._run_num)): mutex.acquire() count = count + 1 mutex.release() print (thneadname, x, count) time.sleep(l) if __name__ == '__main__': global count^ mutex threads =[] num = 5 count =0 #創建鎖 mutex = threading.Lock() #創建線程對象 for x in nange(num): threads.append(Test(10)) #啟動線程 for t in threads: t. start() #等待子線程結束 for t in threads: t.join()
Queue隊列
#!/usr/bin/env python import Queue import threading import urllib2 import time hosts = ["http://xxxx.com", "http://xxxxx.com","http://xxxxxx.com","http://xxxxx.com", "http://xxxxx.com"] queue = Queue.Queue() class ThreadUrl(thneading.Thread): ""”Threaded Uni Grab def __init__(self, queue): threading.Thread.__init__(self) self.queue = queue def run(self): while True: #gnabs host from queue host = self.queue.get() url = urllib2.urlopen(host) #gnabs urls of hosts and prints first 1024 bytes of page uni = urllib2.urlopen(host) #signals to queue job is done self.queue.task_done() start = time.time() def main(): #spawn a pool of threads, and pass them queue instance for i in nange(5): t = ThreadUrl(queue) t.setDaemon(True) t.start() #populate queue with data for host in hosts: queue.put(host) #wait on the queue until everything has been processed queue.join() main() print "Elapsed Time: %s" % (time.time() - start)
8.2 使用隊列與線程
當線程需要共享數據或資源時，線程可能會變得復雜。線程模塊提供許多同步原語，包括信號量，條件變量，事件和鎖。雖然存在這些選項，但它被認為是最佳做法，而是專注於使用隊列。隊列更容易處理，並且使線程編程更安全，因為它們有效地將資源訪問單個線程，並允許更清晰和更可讀的設計模式。
首先創建一個程序，該程序將按順序或一個接一個地獲取網站的URL，並打印頁面的前1024個字節。這是一個經典的例子，可以使用線程更快地完成任務。首先，讓我們一起使用這個urllib2 模塊來抓住這些頁面，然后再使用代碼：
import urllib2 import time hosts = ["http://xxxx.com", "http://xxxxx.com","http://xxxxxx.com","http://xxxxx.com", "http://xxxxx.com"] start = time.time() for host in hosts: url = urllib2.urlopen(host) print url.read(1024) print "Elapsed Time: %s" % (time.time() - start)
導入兩個模塊首先， urllib2模塊是什么是繁重的抓住網頁。其次，通過調用創建開始時間值 time.time()，然后再次調用它，並減去初始值以確定程序執行多長時間。最后，在查看程序的速度時，“兩個半秒”的結果是不可怕的，但是如果您有數百個網頁來檢索，則考慮到目前的平均值，大概需要50秒。看看如何創建一個線程版本加快速度：
import Queue import threading import urllib2 import time hosts = ["http://xxxx.com", "http://xxxxx.com","http://xxxxxx.com","http://xxxxx.com", "http://xxxxx.com"] queue = Queue.Queue() class ThreadUrl(thneading.Thread): ""”Threaded Uni Grab def __init__(self, queue): threading.Thread.__init__(self) self.queue = queue def run(self): while True: #gnabs host from queue host = self.queue.get() url = urllib2.urlopen(host) #gnabs urls of hosts and prints first 1024 bytes of page print url.read(1024) #signals to queue job is done self.queue.task_done() def main(): #spawn a pool of threads, and pass them queue instance for i in nange(5): t = ThreadUrl(queue) t.setDaemon(True) t.start() for host in hosts: queue.put(host) queue.join() main() print "Elapsed Time: %s" % (time.time() - start)
上面的案例並不比第一個線程示例復雜得多，這要歸功於使用排隊模塊。這種模式是使用Python的線程的一種非常常見的推薦方式。步驟描述如下：
1. 創建一個實例，Queue.Queue()然后用數據填充它。
2.將填充數據的實例傳遞到您從繼承中創建的線程類threading.Thread。
3.產生一個守護進程池線程。
4. 一次將一個項目拉出隊列，並使用線程內的數據，運行方法來完成工作。
5.完成工作后，向queue.task_done()任務完成發送一個信號到隊列。
6. 加入隊列，這意味着等到隊列為空，然后退出主程序。
只是一個關於這種模式的注釋：通過將守護進程線程設置為true，它允許主線程或程序退出，如果只有守護進程線程存活。這將創建一種控制程序流程的簡單方法，因為您可以在退出之前連接隊列，或等到隊列為空。具體過程最好在隊列模塊的文檔中描述，如相關主題所示：
join()
“塊直到隊列中的所有項目已經被處理完畢，每當一個項目被添加到隊列中時，未完成任務的計數就會上升，當消費者線程調用task_done（）來指示項目被檢索時，所有的工作都是完成的，當未完成任務的計數下降到零時，join()解除阻塞。