python網絡-多線程(22)

一、什么是線程

線程（英語：thread）是操作系統能夠進行運算調度的最小單位。它被包含在進程之中，是進程中的實際運作單位。同一進程中的多條線程將共享該進程中的全部系統資源，一個進程可以有很多線程，每條線程並行執行不同的任務。

 

二、線程和進程的區別

1、舉例：

• 進程，能夠完成多任務，比如 在一台電腦上能夠同時運行多個QQ
• 線程，能夠完成多任務，比如 一個QQ中的多個聊天窗口

2、定義的不同

• 進程是系統進行資源分配和調度的一個獨立單位.重點是資源分配和調度
• 線程是進程的一個實體,是CPU調度和分派的基本單位它是比進程更小的能獨立運行的基本單位.

3、功能不同

• 一個程序至少有一個進程,一個進程至少有一個線程.
• 線程的划分尺度小於進程(資源比進程少)，使得多線程程序的並發性高。
• 進程在執行過程中擁有獨立的內存單元，而多個線程共享內存，從而極大地提高了程序的運行效率
• 線程不能夠獨立執行，必須依存在進程中

4、優缺點

• 線程和進程在使用上各有優缺點：線程執行開銷小，但不利於資源的管理和保護；而進程正相反。

 

三、threading模塊

1、單線程執行

#coding=utf-8
import time

def sayHi():
    print("Hello I am Se7eN_HOU")
    time.sleep(1)

if __name__ == "__main__":
    for i in range(5):
        sayHi()

運行結果為：

2、多線程執行

#coding=utf-8
import threading
import time

def sayHi():
    print("Hello I am Se7eN_HOU")
    time.sleep(1)

if __name__ == "__main__":
    for i in range(5):
        t = threading.Thread(target = sayHi)
        t.start()

運行結果為：

說明

1. 可以明顯看出使用了多線程並發的操作，花費時間要短很多
2. 創建好的線程，需要調用start()方法來啟動

3、查看線程數量

#coding=utf-8
import threading
from time import sleep,ctime

def sing():
    for i in range(3):
        print("正在唱歌...%d"%i)
        sleep(1)

def dance():
    for i in range(3):
        print("正在跳舞...%d"%i)
        sleep(1)

if __name__ == '__main__':
    print("---開始---")

    t1 = threading.Thread(target=sing)
    t2 = threading.Thread(target=dance)

    t1.start()
    t2.start()

    while True:
        length = len(threading.enumerate())
        print("當前運行的線程數為：%d"%length)
        if length<=1:
            break
        sleep(0.5)

運行結果為:

 

四、Thread子類的封裝

通過使用threading模塊能完成多任務的程序開發，為了讓每個線程的封裝性更完美，所以使用threading模塊時，往往會定義一個新的子類class，只要繼承threading.Thread就可以了，然后重寫run方法

#coding=utf-8
import threading
import time

class MyThread1(threading.Thread):
    def run(self):
        for i in range(3):
            time.sleep(1)
            msg = "I'm "+self.name+' @ '+str(i) #name屬性中保存的是當前線程的名字
            print(msg)

class MyThread2(threading.Thread):
    def run(self):
        for i in range(3):
            time.sleep(1)
            msg = "I'm "+self.name+' @ '+str(i) #name屬性中保存的是當前線程的名字
            print(msg)

if __name__ == '__main__':
    t1 = MyThread1()
    t2 = MyThread2()

    t1.start()
    t2.start()

運行結果為：

說明

• python的threading.Thread類有一個run方法，用於定義線程的功能函數，可以在自己的線程類中覆蓋該方法。
• 創建自己的線程實例后，通過Thread類的start方法，可以啟動該線程，當該線程獲得執行的機會時，就會調用run方法執行線程。

2. 線程的執行順序

#coding=utf-8
import threading
import time

class MyThread(threading.Thread):
    def run(self):
        for i in range(3):
            time.sleep(1)
            msg = "I'm "+self.name+' @ '+str(i) #name屬性中保存的是當前線程的名字
            print(msg)

def test():
    for i in range(1,5):
        t=MyThread()
        t.start()

if __name__ == '__main__':
   test()

運行結果為：

說明：

從代碼和執行結果我們可以看出，多線程程序的執行順序是不確定的。當執行到sleep語句時，線程將被阻塞（Blocked），到sleep結束后，線程進入就緒（Runnable）狀態，等待調度。而線程調度將自行選擇一個線程執行。上面的代碼中只能保證每個線程都運行完整個run函數，但是線程的啟動順序、run函數中每次循環的執行順序都不能確定。

總結：

1. 每個線程一定會有一個名字，盡管上面的例子中沒有指定線程對象的name，但是python會自動為線程指定一個名字。
2. 當線程的run()方法結束時該線程完成。
3. 無法控制線程調度程序，但可以通過別的方式來影響線程調度的方式。

線程的幾種狀態

 

 

五、多線程-全局變量(共享)、局部變量(不共享)

1、共享全局變量

from threading import Thread
import time

g_num = 100

def work1():
    global g_num
    for i in range(3):
        g_num += 1
        print("----in work1, g_num is %d---"%g_num)

def work2():
    global g_num
    print("----in work2, g_num is %d---"%g_num)

print("---線程創建之前,g_num is %d---"%g_num)

t1 = Thread(target=work1)
t1.start()
#延時一會，保證t1線程中的事情做完
time.sleep(1)
t2 = Thread(target=work2)
t2.start()

運行結果為：

---線程創建之前,g_num is 100---
----in work1, g_num is 101---
----in work1, g_num is 102---
----in work1, g_num is 103---
----in work2, g_num is 103---

2、列表當做實參傳遞到線程中

from threading import Thread
import time

def work1(nums):
    nums.append(44)
    print("----in work1---",nums)

def work2(nums):
    #延時一會，保證t1線程中的事情做完
    time.sleep(1)
    print("----in work2---",nums)

g_nums = [11,22,33]

t1 = Thread(target=work1, args=(g_nums,))
t1.start()

t2 = Thread(target=work2, args=(g_nums,))
t2.start()

運行結果為：

----in work1--- [11, 22, 33, 44]
----in work2--- [11, 22, 33, 44]

2、不共享局部變量

import threading
from time import sleep

def test(sleepTime):
    num=1
    sleep(sleepTime)
    num+=1
    print('---(%s)--num=%d'%(threading.current_thread(), num))


t1 = threading.Thread(target = test,args=(5,))
t2 = threading.Thread(target = test,args=(1,))

t1.start()
t2.start()

運行結果為：

---(<Thread(Thread-2, started 12236)>)--num=2
---(<Thread(Thread-1, started 5644)>)--num=2

總結：

• 在一個進程內的所有線程共享全局變量，能夠在不使用其他方式的前提下完成多線程之間的數據共享（這點要比多進程要好）
• 缺點就是，線程是對全局變量隨意遂改可能造成多線程之間對全局變量的混亂（即線程非安全）
• 在多線程開發中，全局變量是多個線程都共享的數據，而局部變量等是各自線程的，是非共享的

 

六、同步

1、線程沖突

假設兩個線程t1和t2都要對num=0進行增1運算，t1和t2都各對num修改10次，num的最終的結果應該為20。但是由於是多線程訪問，有可能出現下面情況：

在num=0時，t1取得num=0。此時系統把t1調度為”sleeping”狀態，把t2轉換為”running”狀態，t2也獲得num=0。然后t2對得到的值進行加1並賦給num，使得num=1。然后系統又把t2調度為”sleeping”，把t1轉為”running”。線程t1又把它之前得到的0加1后賦值給num。這樣，明明t1和t2都完成了1次加1工作，但結果仍然是num=1。

from threading import Thread
import time

g_num = 0

def test1():
    global g_num
    for i in range(1000000):
        g_num += 1

    print("---test1---g_num=%d"%g_num)

def test2():
    global g_num
    for i in range(1000000):
        g_num += 1

    print("---test2---g_num=%d"%g_num)


p1 = Thread(target=test1)
p1.start()
#time.sleep(3) #取消屏蔽之后 再次運行程序，結果會不一樣

p2 = Thread(target=test2)
p2.start()

print("---g_num=%d---"%g_num)
time.sleep(3)
print("---g_num=%d---"%g_num)

運行結果為：

---g_num=138526---
---test1---g_num=1264273
---test2---g_num=1374945
---g_num=1374945---

取消屏蔽之后，再次運行結果如下：

---test1---g_num=1000000
---g_num=1029220---
---test2---g_num=2000000
---g_num=2000000---

問題產生的原因就是沒有控制多個線程對同一資源的訪問，對數據造成破壞，使得線程運行的結果不可預期。這種現象稱為“線程不安全”。

2. 什么是同步

• 同步就是協同步調，按預定的先后次序進行運行。如:你說完，我再說。
• "同"字從字面上容易理解為一起動作，其實不是，"同"字應是指協同、協助、互相配合。
• 如進程、線程同步，可理解為進程或線程A和B一塊配合，A執行到一定程度時要依靠B的某個結果，於是停下來，示意B運行;B依言執行，再將結果給A;A再繼續操作。

3. 解決問題的思路

對於上面提出的那個計算錯誤的問題，可以通過線程同步來進行解決思路，如下:

1. 系統調用t1，然后獲取到num的值為0，此時上一把鎖，即不允許其他現在操作num
2. 對num的值進行+1
3. 解鎖，此時num的值為1，其他的線程就可以使用num了，而且是num的值不是0而是1
4. 同理其他線程在對num進行修改時，都要先上鎖，處理完后再解鎖，在上鎖的整個過程中不允許其他線程訪問，就保證了數據的正確性

 

七、線程互斥鎖

一、線程互斥鎖介紹

當多個線程幾乎同時修改某一個共享數據的時候，需要進行同步控制，線程同步能夠保證多個線程安全訪問競爭資源，最簡單的同步機制是引入互斥鎖。

互斥鎖為資源引入一個狀態：鎖定/非鎖定

某個線程要更改共享數據時，先將其鎖定，此時資源的狀態為“鎖定”，其他線程不能更改；直到該線程釋放資源，將資源的狀態變成“非鎖定”，其他的線程才能再次鎖定該資源。互斥鎖保證了每次只有一個線程進行寫入操作，從而保證了多線程情況下數據的正確性。

threading模塊中定義了Lock類，可以方便的處理鎖定：

#創建鎖
mutex = threading.Lock()
#鎖定
mutex.acquire([blocking])
#釋放
mutex.release()

其中，鎖定方法acquire可以有一個blocking參數。

• 如果設定blocking為True，則當前線程會堵塞，直到獲取到這個鎖為止（如果沒有指定，那么默認為True）
• 如果設定blocking為False，則當前線程不會堵塞

使用互斥鎖實現上面的例子的代碼如下：

from threading import Thread, Lock
import time

g_num = 0

def test1():
    global g_num
    for i in range(1000000):
        #True表示堵塞 即如果這個鎖在上鎖之前已經被上鎖了，那么這個線程會在這里一直等待到解鎖為止 
        #False表示非堵塞，即不管本次調用能夠成功上鎖，都不會卡在這,而是繼續執行下面的代碼
        mutexFlag = mutex.acquire(True) 
        if mutexFlag:#鎖住
            g_num += 1
            mutex.release()#解鎖
    print("---test1---g_num=%d"%g_num)

def test2():
    global g_num
    for i in range(1000000):
        mutexFlag = mutex.acquire(True) #True表示堵塞
        if mutexFlag:#鎖住
            g_num += 1
            mutex.release()#解鎖
    print("---test2---g_num=%d"%g_num)

#創建一個互斥鎖
#這個所默認是未上鎖的狀態
mutex = Lock()

p1 = Thread(target=test1)
p1.start()


p2 = Thread(target=test2)
p2.start()

time.sleep(5)
print("---g_num=%d---"%g_num)

運行結果為：

---test1---g_num=1942922
---test2---g_num=2000000
---g_num=2000000---

2、上鎖解鎖過程

• 當一個線程調用鎖的acquire()方法獲得鎖時，鎖就進入“locked”狀態。
• 每次只有一個線程可以獲得鎖。如果此時另一個線程試圖獲得這個鎖，該線程就會變為“blocked”狀態，稱為“阻塞”，直到擁有鎖的線程調用鎖的release()方法釋放鎖之后，鎖進入“unlocked”狀態。
• 線程調度程序從處於同步阻塞狀態的線程中選擇一個來獲得鎖，並使得該線程進入運行（running）狀態。

總結

鎖的好處：

• 確保了某段關鍵代碼只能由一個線程從頭到尾完整地執行

鎖的壞處：

• 阻止了多線程並發執行，包含鎖的某段代碼實際上只能以單線程模式執行，效率就大大地下降了
• 由於可以存在多個鎖，不同的線程持有不同的鎖，並試圖獲取對方持有的鎖時，可能會造成死鎖

3、死鎖

在線程間共享多個資源的時候，如果兩個線程分別占有一部分資源並且同時等待對方的資源，就會造成死鎖。

盡管死鎖很少發生，但一旦發生就會造成應用的停止響應。下面看一個死鎖的例子

#coding=utf-8
import threading
import time

class MyThread1(threading.Thread):
    def run(self):
        if mutexA.acquire():
            print(self.name+'----do1---up----')
            time.sleep(1)

            if mutexB.acquire():
                print(self.name+'----do1---down----')
                mutexB.release()
            mutexA.release()

class MyThread2(threading.Thread):
    def run(self):
        if mutexB.acquire():
            print(self.name+'----do2---up----')
            time.sleep(1)
            if mutexA.acquire():
                print(self.name+'----do2---down----')
                mutexA.release()
            mutexB.release()

mutexA = threading.Lock()
mutexB = threading.Lock()

if __name__ == '__main__':
    t1 = MyThread1()
    t2 = MyThread2()
    t1.start()
    t2.start()

運行結果為：

Thread-1----do1---up----
Thread-2----do2---up----

此時已經進入到了死鎖狀態

 

 

4、 避免死鎖

• 程序設計時要盡量避免（銀行家算法）
• 添加超時時間等

 

八、同步應用

from threading import Thread,Lock
from time import sleep

class Task1(Thread):
    def run(self):
        while True:
            if lock1.acquire():
                print("------Task 1 -----")
                sleep(0.5)
                lock2.release()

class Task2(Thread):
    def run(self):
        while True:
            if lock2.acquire():
                print("------Task 2 -----")
                sleep(0.5)
                lock3.release()

class Task3(Thread):
    def run(self):
        while True:
            if lock3.acquire():
                print("------Task 3 -----")
                sleep(0.5)
                lock1.release()

#使用Lock創建出的鎖默認沒有“鎖上”
lock1 = Lock()
#創建另外一把鎖，並且“鎖上”
lock2 = Lock()
lock2.acquire()
#創建另外一把鎖，並且“鎖上”
lock3 = Lock()
lock3.acquire()

t1 = Task1()
t2 = Task2()
t3 = Task3()

t1.start()
t2.start()
t3.start()

運行結果為：

可以使用互斥鎖完成多個任務，有序的進程工作，這就是線程的同步

 

九、ThreadLocal

在多線程環境下，每個線程都有自己的數據。一個線程使用自己的局部變量比使用全局變量好，因為局部變量只有線程自己能看見，不會影響其他線程，而全局變量的修改必須加鎖。

import threading

# 創建全局ThreadLocal對象:
local_school = threading.local()

def process_student():
    # 獲取當前線程關聯的student:
    std = local_school.student
    print('Hello, %s (in %s)' % (std, threading.current_thread().name))

def process_thread(name):
    # 綁定ThreadLocal的student:
    local_school.student = name
    process_student()

t1 = threading.Thread(target= process_thread, args=("Se7eN",), name='Thread-A')
t2 = threading.Thread(target= process_thread, args=("HOU",), name='Thread-B')
t1.start()
t2.start()
t1.join()
t2.join()

運行結果為：

Hello, Se7eN (in Thread-A)
Hello, HOU (in Thread-B)

說明：

1. 全局變量local_school就是一個ThreadLocal對象，每個Thread對它都可以讀寫student屬性，但互不影響。你可以把local_school看成全局變量，但每個屬性如local_school.student都是線程的局部變量，可以任意讀寫而互不干擾，也不用管理鎖的問題，ThreadLocal內部會處理。
2. ThreadLocal最常用的地方就是為每個線程綁定一個數據庫連接，HTTP請求，用戶身份信息等，這樣一個線程的所有調用到的處理函數都可以非常方便地訪問這些資源。
3.  一個ThreadLocal變量雖然是全局變量，但每個線程都只能讀寫自己線程的獨立副本，互不干擾。ThreadLocal解決了參數在一個線程中各個函數之間互相傳遞的問題

 

十、異步

• 同步調用就是你 喊 你朋友吃飯 ，你朋友在忙 ，你就一直在那等，等你朋友忙完了 ，你們一起去
• 異步調用就是你 喊 你朋友吃飯 ，你朋友說知道了 ，待會忙完去找你 ，你就去做別的了。

def test():
    print("---進程池中的進程---pid=%d,ppid=%d--"%(os.getpid(),os.getppid()))
    for i in range(3):
        print("----%d---"%i)
        time.sleep(1)
    return "hahah"

def test2(args):
    print("---callback func--pid=%d"%os.getpid())
    print("---callback func--args=%s"%args)
   
pool = Pool(3)
pool.apply_async(func=test,callback=test2)
time.sleep(5)
print("----主進程-pid=%d----"%os.getpid())

運行結果為：

---進程池中的進程---pid=9401,ppid=9400--
----0---
----1---
----2---
---callback func--pid=9400
---callback func--args=hahah
----主進程-pid=9400----

 十一、並行和並發

無論是並行還是並發，在用戶看來都是'同時'運行的，不管是進程還是線程，都只是一個任務而已，真是干活的是cpu，cpu來做這些任務，而一個cpu同一時刻只能執行一個任務。

並發是偽並行，即看起來是同時運行。單個cpu+多道技術就可以實現並發，（並行也屬於並發），簡單的可以理解為快速在多個線程來回切換，感覺好像同時在做多個事情。

只有具備多個cpu才能實現並行，單核下，可以利用多道技術，多個核，每個核也都可以利用多道技術（多道技術是針對單核而言的）。          

有四個核，六個任務，這樣同一時間有四個任務被執行，假設分別被分配給了cpu1，cpu2，cpu3，cpu4，一旦任務1遇到I/O就被迫中斷執行，此時任務5就拿到cpu1的時間片去執行，這就是單核下的多道技術 ，而一旦任務1的I/O結束了，操作系統會重新調用它(需知進程的調度、分配給哪個cpu運行，由操作系統說了算)，可能被分配給四個cpu中的任意一個去執行。

　　多道技術：內存中同時存入多道（多個）程序，cpu從一個進程快速切換到另外一個，使每個進程各自運行幾十或幾百毫秒，這樣，雖然在某一個瞬間，一個cpu只能執行一個任務，但在1秒內，cpu卻可以運行多個進程，這就給人產生了並行的錯覺，即偽並發，以此來區分多處理器操作系統的真正硬件並行（多個cpu共享同一個物理內存）

• 同步執行：一個進程在執行某個任務時，另外一個進程必須等待其執行完畢，才能繼續執行
• 異步執行：一個進程在執行某個任務時，另外一個進程無需等待其執行完畢，就可以繼續執行，當有消息返回時，系統會通知后者進行處理，這樣可以提高執行效率

    舉個例子，打電話時就是同步通信，發短息時就是異步通信。