web端自動化——python多線程

Python通過兩個標准庫thread和threading提供對線程的支持。thread提供了低級別的、原始的線程以及一個簡單的鎖。threading基於Java的線程模型設計。

鎖（Lock）條件變量（Condition）在Java中是對象的基本行為（每一個對象都自帶了鎖和條件變量），而在 Python中則是獨立的對象。

注意：我們應該避免使用thread模塊，原因是thread模塊不支持守護線程。當主線程退出時，所有的子線程不管它們是否還在工作，都會被強行退出。

threading模塊支持守護線程。

 

1、進程與線程的區別：

 

2、multiprocessing多進程模塊

多進程multiprocessing模塊的使用與多線程threading模塊的用法類似。multiprocessing提供了本地和遠程的並發性，有效地通過全局解釋鎖（Global InterceptorLock,GIL)來使用進程（而不是線程）。由於GIL的存在，在CPU密集型的程序當中，使用多線程並不能有效地利用多核CPU的優勢，因為一個解釋器在同一時刻只會有一個線程在執行。所以，multiprocessing模塊可以充分利用硬件的多處理器來進行工作。它支持UNlX和Windows系統上的運行。

3、threading 多線程模塊

threading模塊支持守護線程。

 

4、Pipe和Queue

multiprocessing提供threading包中沒有的IPC(進程間通信），效率上更高。應優先考慮Pipe和Queue，避免使用Lock/Event/Semaphore/Condition等同步方式（因為它們根據的不是用戶進程的資源）。

multiprocessing包中有Pipe類和Queue類來分別支持這兩種IPC機制。Pipe和Queue 可以用來傳送常見的對象。

①    Pip可以是單向化(half-duplex)，也可以是雙向(duplex)。我們通過multiprocessing.Pipe (duplex=False)創建單向管道（默認為雙向）。一個進程從pipe—端輸入對象，然后被pipe 另一端的進程接收。單向管道只允許管道一端的進程輸入，而雙向管道則允許從兩端輸入。

Pipe.py

import multiprocessing

def proc1(pipe):

    pipe.send('hello')

    print('proc1 rec:', pipe.recv())

def proc2(pipe):

    print('proc2 rec:', pipe.recv())

    pipe.send('hello, too')

if __name__ == "__main__":

    multiprocessing.freeze_support()

    pipe = multiprocessing.Pipe()

    p1 = multiprocessing.Process(target=proc1,args=(pipe[0],))

    p2 = multiprocessing.Process(target=proc2,args=(pipe[1],))

    p1.start()

    p2.start()

    p1.join()

    p2.join()

這里的pipe是雙向的。pipe對象建立的時候，返回一個含有兩個元素的表，每個元素 代表pipe的一端（Connection對象）。我們對pipe的某一端調用send()方法來傳送對象，在另一端使用recv()來接收。

②     Queue類與Pipe相類似，都是先進先出結構。但Queue類允許多個進程放入，多個進程從隊列取出對象。Queue類使用Queue (maxsize)創建，maxsize表示隊列中可以存放對象的最大數量。

queue.py

import multiprocessing

import os, time

def inputQ(queue):

    #getpid是獲得當前進程的進程號。系統每開辟一個新進程就會為他分配一個進程號。

    info = str(os.getpid()) + '(put):'+str(time.time())

    #put方法用以插入數據到隊列中

    queue.put(info)

def outputQ(queue, lock):

    #get方法可以從隊列讀取並且刪除一個元素

    info = queue.get()

    lock.acquire()

    print((str(os.getpid()) + '(get):'+info))

    lock.release()

if __name__=='__main__':

    record1 =[]

    record2 =[]

    lock = multiprocessing.Lock()#加鎖，為防止散亂的打印

    queue = multiprocessing.Queue(3)

    for i in range(10):

        process = multiprocessing.Process(target=inputQ,args=(queue,))

        process.start()

        record1.append(process)

    for i in range(10):

        process = multiprocessing.Process(target=outputQ,args=(queue,lock))

        process.start()

        record2.append(process)

    for p in record1:

        p.join()

    for p in record2:

        p.join()

    queue.close() #沒有更多的對象進來，關閉queue

 

5、多線程分布式執行測試用例

Selenium Grid只是提供多系統、多瀏覽器的執行環境，Selenium Grid本身並不提供並行的執行測試用例，這個我們在前面己經反復強調。下面就通過演示使用多線程技術結合Selenium Grid實現分布式並行地執行測試用例。

啟動 Selenium Server

在本機打開兩個命令提示符窗口。

本機啟動一個主hub和一個node節點（端口號別分為4444和5555),本機IP地址為: 172.16.101.239。

>java -jar selenium-server-standalone-2.47.0.jar -role hub

>java -jar selenium-server-standalone-2.47.0.jar -role node -port 5555

啟動一個遠程的node(設罝端口號為6666),IP地址假設為：172.16.101.238。

>java -jar selenium-server-standalone-2.47.jar -role node -port 6666 -hub http://172.16.101.239:4444/grid/register

運行測試腳本。

grid_thread.py

from threading import Thread

from selenium import webdriver

from time import sleep,ctime

#測試用例

def test_baidu(host,browser):

    print('start:%s' %ctime())

    print(host, browser)

    dc = {'browserName':browser}

    driver = webdriver.Remote(conunand_executor=host,

                              desired_capabilities=dc)

    driver.get('http://www.baidu.com')

    driver.find_element_by_id("kw").send_keys(browser)

    driver.find_element_by_id("su").click()

    sleep(2)

    driver.quit()

if __name__ =='__main__'：   

#啟動參數(指定運行主機與瀏覽器)

    lists = {'http://127.0.0.1:4444/wd/hub': 'chrome',

             'http://127.0.0.1:5555/wd/hub': 'internet explorer',

             'http://172.16.101.238:6666/wd/hub':'firefox', # 遠程節點

             }

    threads =[]

    files = range(len(lists))

    #創建線程

    for host, browser in lists.items():

            t = Thread(target=test_baidu, args=(host, browser))

            threads.append(t)

    #啟動線程

    for i in files:

        threads[i].start()

    for i in files:

        threads [i].join()

    print('end:%s' %ctime())

 

 6、多線程執行測試用例

from threading import Thread
from selenium import webdriver
from time import ctime,sleep
#測試用例
def test_baidu(browser,search):
    print('start:%s' %ctime())
    print('browser:%s ,' %browser)
    if browser == "chrome":
        driver = webdriver.Chrome()
    elif browser == "ff":
        driver = webdriver.Firefox()
    else:
        print('browser參數有誤，只能為ff、chrome')
        driver.get('http://www.baidu.com')
        driver.find_element_by_id("kw").send_keys(search)
        driver.find_element_by_id('su').click()
        sleep(2)
        driver.quit()
if __name__ == "__main__":
#啟動參數（指瀏覽器與百度搜索內容）
    lists ={'chrome':'threading', 'ff':'python'}
    threads =[]
    files = range(len(lists))
#創建線程
for browser, search in lists.items():
    t = Thread(target=test_baidu,args=(browser,search))
    threads.append(t)
#啟動線程
for t in files:
    threads[t].start()
for t in files:
    threads[t].join()
print('end:%s' %ctime())