多線程、多進程和線程池編程

一.python中的GIL（Global Interpreter Lock）

　　詳情：https://www.cnblogs.com/SuKiWX/p/8804974.html

　　介紹：

　　　　GIL：全局解釋器鎖（Cpython中才有，Jpython沒有，pypy是去gil的）；

　　　　cpython：pyhon中的一個線程對應C語言中的一個線程；

　　　    gil使得同一個時刻只有一個線程在一個cpu上執行字節碼，無法將多個線程映射到多cpu上；
　　　    gil在一些情況下會釋放，是結合字節碼和時間片釋放（Python2和Python3有差別），gil在遇到io操作的時候會主動釋放

#gil會釋放，最后的結果不定，釋放的位置不定
import threading
total=1
def add():
    global total
    for i in range(1000000):
        total+=1

def decs():
    global total
    for i in range(1000000):
        total-=1

thread1=threading.Thread(target=add)
thread2=threading.Thread(target=decs)
thread1.start()
thread2.start()
thread1.join()
thread2.join()
print(total)

 

　　注：

　　　　Python GIL其實是功能和性能之間權衡后的產物，它尤其存在的合理性，也有較難改變的客觀因素。從本分的分析中，我們可以做以下一些簡單的總結：

• • • • 因為GIL的存在，只有IO Bound場景下得多線程會得到較好的性能
      • 如果對並行計算性能較高的程序可以考慮把核心部分也成C模塊，或者索性用其他語言實現
      • GIL在較長一段時間內將會繼續存在，但是會不斷對其進行改進

 

二.python多線程編程

　　1.利用Thread實例實現多線程：

　　　　這里子線程默認為非守護線程（主線程運行完，子線程不會退出，繼續運行完）

# 對於io操作，多線程和多進程性能差別不大
# 通過Thread實例化

import time
import threading
def get_detail_html(url):
    print('我獲取詳情內容了')
    time.sleep(2)
    print('我獲取內容完了')
def get_detail_url(url):
    print('我獲取url了')
    time.sleep(2)
    print('我獲取url完了')
if __name__=='__main__':
    thread1=threading.Thread(target=get_detail_html,args=('',))
    thread2=threading.Thread(target=get_detail_url,args=('',))
    start_time=time.time()
    thread1.start()
    thread2.start()
    #時間非常小，是運行代碼的時間差，而不是2秒
    #這樣運行一共有三個線程，主線程和其他兩個子線程（thread1，thread2），而且是並行的，子線程啟動后，主線程仍然往下運行，因此時間不是2秒
    #守護線程（主線程退出，子線程就會kill掉）
    print('last time:{}'.format(time.time()-start_time))

 

 

共三個進程，主線程和兩個子線程

　　2.守護線程：（主線程退出，子線程就會被kill掉）

import time
import threading
def get_detail_html(url):
    print('我獲取詳情內容了')
    time.sleep(4)
    print('我獲取內容完了')
def get_detail_url(url):
    print('我獲取url了')
    time.sleep(2)
    print('我獲取url完了')
if __name__=='__main__':
    thread1=threading.Thread(target=get_detail_html,args=('',))
    thread2=threading.Thread(target=get_detail_url,args=('',))
    #將線程1設置成守護線程（主線程退出，該線程就會被kill掉），但會等線程2運行完（非守護線程）
    thread1.setDaemon(True)
    start_time=time.time()
    thread1.start()
    thread2.start()
    print('last time:{}'.format(time.time()-start_time))

 

線程1未運行完退出

　　3.join():等某個子線程執行完在繼續執行主線程代碼：

import time
import threading
def get_detail_html(url):
    print('我獲取詳情內容了')
    time.sleep(4)
    print('我獲取內容完了')
def get_detail_url(url):
    print('我獲取url了')
    time.sleep(2)
    print('我獲取url完了')
if __name__=='__main__':
    thread1=threading.Thread(target=get_detail_html,args=('',))
    thread2=threading.Thread(target=get_detail_url,args=('',))
    start_time=time.time()
    thread1.start()
    thread2.start()
    #等待兩個線程執行完
    thread1.join()
    thread2.join()
    print('last time:{}'.format(time.time()-start_time))

 

執行時間是線程最大時間（並發執行）

　　4.繼承Thread實現多線程（代碼較復雜時）：

import time
import threading
class GetDetailHtml(threading.Thread):
    def __init__(self, name):
        super().__init__(name=name)
    def run(self):
        print('我獲取詳情內容了')
        time.sleep(4)
        print('我獲取內容完了')

class GetDetailUrl(threading.Thread):
    def __init__(self,name):
        super().__init__(name=name)
    def run(self):
        print('我獲取url了')
        time.sleep(2)
        print('我獲取url完了')
if __name__=='__main__':
    thread1=GetDetailHtml('get_detail_html')
    thread2=GetDetailUrl('get_detail_url')
    start_time=time.time()
    thread1.start()
    thread2.start()
    #等待兩個線程執行完
    thread1.join()
    thread2.join()
    print('last time:{}'.format(time.time()-start_time))

三.線程間通信-Queue

　　1.線程通信方式——共享變量：（全局變量或參數等）

　　　　注：共享變量的方式是線程不安全的操作（不推薦）

import time
import threading
url_lists = []
def get_detail_html():
    #可以單獨放在某一個文件管理（注意引入時要引用文件）
    global url_lists
    url_lists=url_lists
    while True:
        if len(url_lists):
            url=url_lists.pop()
            print('我獲取詳情內容了')
            time.sleep(4)
            print('我獲取內容完了')
def get_detail_url(url_lists):
    while True:
        print('我獲取url了')
        time.sleep(2)
        for i in range(20):
            url_lists.append('url'+str(i))
        print('我獲取url完了')
if __name__ == '__main__':
    thread_url=threading.Thread(target=get_detail_url,args=(url_lists,))
    thread_url.start()
    #開啟十個線程爬取詳情
    for i in range(10):
        thread_html=threading.Thread(target=get_detail_html,)
        thread_html.start()

 

　　2.通過queue的方式進行線程同步：

　　　　注：是線程安全的（Queue本身就是線程安全的【使用了線程鎖的機制】，使用了雙端隊列，deque）

　　　　Queue中的方法：qsize()查看對列大小，empty()判斷隊列是否為空，full()判斷隊列是否滿，滿了的話put方法就會阻塞，等待有空位加入，put()將數據放入隊列，默認是阻塞的（block參數，可以設置成非阻塞，還有timeout等待時間），get()從隊列取數據

 

import time
import threading
from queue import Queue
url_lists = []
def get_detail_html(queue):
    while True:
        url=queue.get()
        print('我獲取詳情內容了')
        time.sleep(4)
        print('我獲取內容完了')
def get_detail_url(queue):
    while True:
        print('我獲取url了')
        time.sleep(2)
        for i in range(20):
            queue.put('url'+str(i))
        print('我獲取url完了')
if __name__ == '__main__':
    #設置隊列最大值1000，過大對內存會有很大影響
    urls_queue=Queue(maxsize=1000)
    thread_url=threading.Thread(target=get_detail_url,args=(urls_queue,))
    thread_url.start()
    #開啟十個線程爬取詳情
    for i in range(10):
        thread_html=threading.Thread(target=get_detail_html,args=(urls_queue,))
        thread_html.start()
    #執行該方法才能執行退出,和join成對出現
    urls_queue.task_done()
    urls_queue.join()

 

四. 線程同步（Lock、RLock、Semaphores、Condition）

　　問題：如例一中的問題，最后的結果不正確且不穩定，是因為在字節碼運行時，線程隨時可能跳轉，導致賦值不正確，因此需要一個鎖，讓某段代碼運行時，另一代碼段不運行

　　1.Lock：鎖住的代碼段都只能有一個代碼段運行

　　　　獲取（acquire）和釋放（release）鎖都需要時間：因此用鎖會影響性能；還有可能引起死鎖（互相等待，A和B都需要a,b兩個資源，A獲取了a，B獲取了B，A等待b，B等待a或則未釋放鎖再次獲取）;

　　　　產生死鎖的四個條件：互斥條件；不剝奪條件；請求和保持條件；循環等待條件

import threading
from threading import Lock
total=1
lock=Lock()
def add():
    global total
    for i in range(1000000):
        #獲取鎖
        lock.acquire()
        total+=1
        #釋放鎖，釋放后其他才能獲取
        lock.release()

def decs():
    global total
    for i in range(1000000):
        lock.acquire()
        total-=1
        lock.release()
thread1=threading.Thread(target=add)
thread2=threading.Thread(target=decs)
thread1.start()
thread2.start()
thread1.join()
thread2.join()
print(total)

 

　　2.RLock(可重入的鎖)：在一個線程中可以，可以連續多次acquire（獲取資源），一定要注意acquire的次數要和release的次數一致

 

 　　3.Condition:條件變量（用於復雜的線程間同步）

　　　　3.1使用鎖進行先后對話：發現先啟動的線程把話先說完（第一個線程啟動后運行完，第二個線程還沒有啟動，或者還未切換到另一個線程）

　　　　3.2通過condition實現：

　　　　　（通過調用with方法（實際是__enter__魔法函數），也可以使用acquire()方法【如self.conf.acquire()】，但記得一定要release()之后才能調用其他函數【wait(),notify()】，還有注意線程啟動順序【先接收方先啟動，否則接收不到】），Condition有兩層鎖，一把地層鎖，會在線程調用了wait()方法時釋放，上面的鎖會在每次調用wait()時分配一把鎖並放入到condition的等待隊列中，等待notify()方法的喚醒

　　　　　　Condition重要函數：acquire(),release()【都調用了Lock的acquire，release】，wait()【允許等待某個通知在操作,會獲取一把鎖並把Condition中的鎖釋放掉】，notify()【發送通知，釋放鎖】

import threading



class LYQ1(threading.Thread):
    def __init__(self, cond):
        super().__init__(name="LYQ1")
        self.cond = cond

    def run(self):
        with self.cond:
            print('LYQ1:你好，我是{}'.format(self.name))
            #輸出后發出通知
            self.cond.notify()
            self.cond.wait()
            print('LYQ1:哈哈')
            self.cond.notify()
            self.cond.wait()
            print('LYQ1:嘿嘿')
            self.cond.notify()


class LYQ2(threading.Thread):
    def __init__(self, cond):
        super().__init__(name="LYQ2")
        self.cond = cond

    def run(self):
        with self.cond:
            #等待通知
            self.cond.wait()
            print('LYQ2:你好，我是{}'.format(self.name))
            self.cond.notify()
            self.cond.wait()
            print('LYQ2:好的'.format(self.name))
            self.cond.notify()
            self.cond.wait()
            print('LYQ2：好久不見')
            self.cond.notify()


if __name__ == "__main__":
    cond = threading.Condition()
    lyq1 = LYQ1(cond=cond)
    lyq2 = LYQ2(cond=cond)
    #注意啟動順序，如果先啟動lyq1，發送通知確沒有接收（lyq2還沒有啟動）
    lyq2.start()
    lyq1.start()
    '''
    輸出：
    LYQ1:你好，我是LYQ1
    LYQ2:你好，我是LYQ2
    LYQ1:哈哈
    LYQ2:好的
    LYQ1:嘿嘿
    LYQ2：好久不見'''

 　　4.Semaphores:（有一個參數value可以控制線程（並發數），調用acquire方法value就會減一，如果減少到為0就會阻塞在那兒等待有空位，調用release()value就會加一）【線程數量過多會影響切換線程的效率】　　　

　　　　Semaphores內部實質是用Condition完成的，Queue實質也是；

　　　　用來控制進入數量的鎖（如文件寫一般只能一個線程，讀可以允許同時多個線程讀。）

# 控制線程的數量
from threading import Semaphore
import threading
import time


class UrlProducer(threading.Thread):
    def __init__(self, sem):
        super().__init__()
        self.sem = sem

    def run(self):
        for i in range(20):
            #調用acquire方法，Semaphore中的value就會減一（value），如果為0就阻塞在這兒
            self.sem.acquire()
            html_get = HtmlGet('url' + str(i),sem)
            html_get.start()


class HtmlGet(threading.Thread):
    def __init__(self, url,sem):
        super().__init__()
        self.url = url
        self.sem=sem

    def run(self):
        time.sleep(2)
        print('獲取網頁成功')
        #調用release方法，Semaphore中的value就會加一（value）
        self.sem.release()


if __name__ == '__main__':
    #允許並發的個數
    sem=Semaphore(3)
    urlproducer = UrlProducer(sem)
    urlproducer.start()

五.concurrent線程池編碼

　　1.為什么需要線程池：

　　　　提供了數量控制，獲取線程的狀態及返回值；當一個線程完成的時候主線程能立即知道；futures能讓多線程和多進程編碼接口一致

　　2.ThreadPoolExecutor中重要函數：

　　　　submit():通過submit函數提交執行的函數到線程池,立即返回值（不會阻塞）;

　　　　done():done()判斷某個任務是否執行成功；

　　　　result():獲取返回值;

　　　　cance():取消某個任務（還未執行，執行中不能取消）；

　　　　wait()讓主線程阻塞等待子線程完成，可以添加參數等待多長時間就不等待了

　　3.獲取已經完成的任務：

　　　　as_completed() [from concurrent.futures import ThreadPoolExecutor,as_completed]；

　　　　map(屬於ThreadPoolExecutor)　　

　　

#as_completed會將成功的url的返回值yield出去
from concurrent.futures import ThreadPoolExecutor,as_completed,wait
import time
def get_html(times):
    time.sleep(times)
    print('get success {}'.format(times))
    return times

excutor=ThreadPoolExecutor(max_workers=2)
# #通過submit函數提交執行的函數到線程池,立即返回值（不會阻塞）
# ret1=excutor.submit(get_html,(3))
# #done()判斷函數是否執行成功
# print(ret1.done())
# print(ret1.cancel())
# ret2=excutor.submit(get_html,(2))
# print(ret2.done())
# #result()獲取返回值
# print(ret1.result())
urls=[2,3,4]
#獲取已經完成的任務
all_task=[excutor.submit(get_html,(url)) for url in urls]
#wait(讓主線程阻塞等待子線程完成)，可以
wait(all_task,return_when='FIRST_COMPLETED')
print('haha')
# for future in as_completed(all_task):
#     data=future.result()
#     print(data)
# for data in excutor.map(get_html,urls):
#     print(data)

 　　4:ThreadPoolExecutor源碼簡介：

　　　　submit方法會加一把鎖，創建Future，然后放入WorkItem（執行單元），將執行單元放入執行隊列中，_adjust_thread_count（）會比較啟動得線程數量和最大線程數量，如果小於就會啟動一個線程放入_threads里面，_threads里面執行的是_worker()函數，參數為work_que，獲取WorkItem，然后調用里面的run()方法，將值設置到future中去。

Future中的方法，cancel()，done()等等都是判斷狀態，result()是會阻塞的，調用Condition()

六.多進程編程-multiprocessing

　　1.和多線程對比：

　　　　·1.1多進程開銷大，多線程開銷小；

　　　　 1.2耗CPU的操作，多進程編程比多線程編程好很多，對於IO操作來說，使用多線程操作比多進程好（線程切換比進程切換性能高）

　　2.例：

　　　　1.1對於耗CPU的操作（多進程優於多線程）：

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor,as_completed
import time

def fib(n):
    if n<=2:
        return 1
    return fib(n-2)+fib(n-1)
if __name__=='__main__':
    #代碼要放在這里面，不然可能拋異常
    with ThreadPoolExecutor(3) as excutor:
        start_time=time.time()
        all_task=[excutor.submit(fib,num) for num in range(25,40)]
        for future in as_completed(all_task):
            data=future.result()
            print('結果：'+str(data))
        print('多線程所需時間：'+str(time.time()-start_time))
    '''
    多線程所需時間：72.10901117324829
    '''

    with ProcessPoolExecutor(3) as excutor:
        start_time=time.time()
        all_task=[excutor.submit(fib,num) for num in range(25,40)]
        for future in as_completed(all_task):
            data=future.result()
            print('結果：'+str(data))
        print('多進程所需時間：'+str(time.time()-start_time))
    '''
    多進程所需時間：43.14996862411499
    '''

 

　　　　1.2對於IO操作，多線程由於多進程：　　　　

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor,as_completed
import time

def random_sleep(n):
    time.sleep(n)
    return n
if __name__=='__main__':
    #代碼要放在這里面，不然可能拋異常
    with ThreadPoolExecutor(3) as excutor:
        start_time=time.time()
        all_task=[excutor.submit(random_sleep,num) for num in [2]*30]
        for future in as_completed(all_task):
            data=future.result()
            print('休息：'+str(data)+'秒')
        print('多線程所需時間：'+str(time.time()-start_time))
    '''
    多線程所需時間：20.010841131210327
    '''

    with ProcessPoolExecutor(3) as excutor:
        start_time=time.time()
        all_task=[excutor.submit(random_sleep,num) for num in [2]*30]
        for future in as_completed(all_task):
            data=future.result()
            print('休息：'+str(data)+'秒')
        print('多進程所需時間：'+str(time.time()-start_time))
    '''
    20.755817651748657
    '''

　　3.進程有趣的例子：

import os
import time
#fork只能用於Linux/Unix
#進程間數據是隔離的，每個進程都有完整的數據，fork()會將父進程代碼復制一遍在運行（fork后的代碼）
pid=os.fork()
print('LYQ')
if pid==0:
    print('子進程:{},父進程：{}'.format(os.getpid(),os.getppid()))
else:
    print('我是父進程：{}'.format(pid))
    #暫停兩秒，父進程還沒有退出，子進程可以運行完，父進程退出就可以kill掉子進程
    #不暫停的話，父進程退出，子進程仍然在運行，輸出
    time.sleep(2)
'''
暫停的輸出：
LYQ
我是父進程：8587
LYQ
子進程:8587,父進程：8578

不暫停的輸出：
LYQ
我是父進程：8587
[root@izwz97n253zzwjtudbqt5uz ~]# LYQ
子進程:8587,父進程：1
'''

 　　4.multiprocessing:(比ProcessPoolExecutor更底層【基於multiprocessing實現】，推薦ProcessPoolExecutor更好的設計，和ThreadPoolExecutor相似):

　　　　

import time
import multiprocessing
def get_html(n):
    time.sleep(n)
    print('sub_process success')
    return n

if __name__=='__main__':
    process=multiprocessing.Process(target=get_html,args=(1,))
    #獲取進程號，沒有start之前為None
    print(process.pid)
    process.start()
    print(process.pid)
    process.join()
    print('main_process  success')

 

　　5進程池：

......
pool=multiprocessing.Pool(3)
    #異步提交任務
   #  result=pool.apply_async(get_html,args=(2,))
   # #關閉不在進入進程池
   #  pool.close()
   #  pool.join()
   #  print(result.get())
   #和執行順序一樣
   for result in pool.imap(get_html,[1,5,3]):
       print('{} sleep success'.format(result))
   #和先后完成順序一樣
   for result in pool.imap_unordered(get_html, [1, 5, 3]):
       print('{} sleep success'.format(result))

 

七.進程間通信

　　1.共享全局變量在多進程中不適用（會把數據復制到子進程中，數據是獨立的，修改也不會影響），quue中的Queue也不行，需要做一些處理

from multiprocessing import Queue,Process
import time
def producer(queue):
     queue.put('a')
     time.sleep(2)

def consumer(queue):
    time.sleep(2)
    data=queue.get()
    print(data)

if __name__=='__main__':
    queue=Queue(10)
    pro_producer=Process(target=producer,args=(queue,))
    pro_consumer=Process(target=consumer,args=(queue,))
    pro_producer.start()
    pro_consumer.start()
    pro_producer.join()
    pro_consumer.join()

 

　　2.multiprocessing中的Queue不能用於進程池（需要用到manager）：

　　　queue=Manager().Queue(10)
　　　　

from queue import Queue——>用於多線程
from multiprocessing import Queue——>用於非進程池的多進程通信
from multiprocessing import Manager——>manager.Queue()用於進程池通信

 

 　　3.通過Pipe進行進程間通信（管道），pipe只能適用於兩個進程 ，Pipe性能高於queue

from multiprocessing import Pipe
import time
def producer(pipe):
     pipe.send('a')
     time.sleep(2)

def consumer(pipe):
    time.sleep(2)
    data=pipe.recv()
    print(data)

if __name__=='__main__':
    #通過Pipe進行進程間通信（管道），pipe只能適用於兩個進程
    recv_pipe,send_pipe=Pipe()
    queue=Manager().Queue(10)
    pro_producer=Process(target=producer,args=(send_pipe,))
    pro_consumer=Process(target=consumer,args=(recv_pipe,))
    pro_producer.start()
    pro_consumer.start()
    pro_producer.join()
    pro_consumer.join()

 

　　4.進程間共享內存（Manager）： 

from multiprocessing import Manager,Process

def add_data(pro_dict, key, value):
    pro_dict[key] = value

if __name__=='__main__':
    #常用的類型都有

    process_dict=Manager().dict()
    fir=Process(target=add_data,args=(process_dict,'name1','LYQ1'))
    sed = Process(target=add_data, args=(process_dict, 'name2', 'LYQ2'))
    fir.start()
    sed.start()
    fir.join()
    sed.join()
    print(process_dict)