快捷導(dǎo)航

python 實(shí)現(xiàn)線程之間的通信示例

更新時(shí)間：2020年02月14日 11:46:14 作者：三只松鼠

這篇文章主要介紹了python 實(shí)現(xiàn)線程之間的通信示例，文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧

前言：因?yàn)镚IL的限制，python的線程是無法真正意義上并行的。相對(duì)于異步編程，其性能可以說不是一個(gè)等量級(jí)的。為什么我們還要學(xué)習(xí)多線程編程呢，雖然說異步編程好處多，但編程也較為復(fù)雜，邏輯不容易理解，學(xué)習(xí)成本和維護(hù)成本都比較高。畢竟我們大部分人還是適應(yīng)同步編碼的，除非一些需要高性能處理的地方采用異步。

首先普及下進(jìn)程和線程的概念：

進(jìn)程：進(jìn)程是操作系統(tǒng)資源分配的基本單位。

線程：線程是任務(wù)調(diào)度和執(zhí)行的基本單位。

一個(gè)應(yīng)用程序至少一個(gè)進(jìn)程，一個(gè)進(jìn)程至少一個(gè)線程。

兩者區(qū)別：同一進(jìn)程內(nèi)的線程共享本進(jìn)程的資源如內(nèi)存、I/O、cpu等，但是進(jìn)程之間的資源是獨(dú)立的。

一、多線程

python 可以通過 thread 或 threading 模塊實(shí)現(xiàn)多線程，threading 相比 thread 提供了更高階、更全面的線程管理。我們下文主要以 threading 模塊介紹多線程的基本用法。

import threading
import time

class thread(threading.Thread):
 def __init__(self, threadname):
 threading.Thread.__init__(self, name='線程' + threadname)

 def run(self):
 print('%s:Now timestamp is %s'%(self.name,time.time()))

threads = []
for a in range(int(5)): # 線程個(gè)數(shù)
 threads.append(thread(str(a)))
for t in threads: # 開啟線程
 t.start()
for t in threads: # 阻塞線程
 t.join()
print('END')

輸出：
線程3:Now timestamp is 1557386184.7574518
線程2:Now timestamp is 1557386184.7574518
線程0:Now timestamp is 1557386184.7574518
線程1:Now timestamp is 1557386184.7574518
線程4:Now timestamp is 1557386184.7582724
END

start() 方法開啟子線程。運(yùn)行多次 start() 方法代表開啟多個(gè)子線程。

join() 方法用來阻塞主線程，等待子線程執(zhí)行完成。舉個(gè)例子，主線程A創(chuàng)建了子線程B，并使用了 join() 方法，主線程A在 join() 處就被阻塞了，等待子線程B完成后，主線程A才能執(zhí)行 print('END')。如果沒有使用 join() 方法，主線程A創(chuàng)建子線程B后，不會(huì)等待子線程B，直接執(zhí)行 print('END')，如下：

import threading
import time

class thread(threading.Thread):
 def __init__(self, threadname):
 threading.Thread.__init__(self, name='線程' + threadname)

 def run(self):
 time.sleep(1)
 print('%s:Now timestamp is %s'%(self.name,time.time()))

threads = []
for a in range(int(5)): # 線程個(gè)數(shù)
 threads.append(thread(str(a)))
for t in threads: # 開啟線程
 t.start()
# for t in threads: # 阻塞線程
# t.join()
print('END')

輸出：
END
線程0:Now timestamp is 1557386321.376941
線程3:Now timestamp is 1557386321.377937
線程1:Now timestamp is 1557386321.377937
線程2:Now timestamp is 1557386321.377937
線程4:Now timestamp is 1557386321.377937

二、線程之間的通信

1.threading.Lock()

如果多個(gè)線程對(duì)某一資源同時(shí)進(jìn)行修改，可能會(huì)存在不可預(yù)知的情況。為了修改數(shù)據(jù)的正確性，需要把這個(gè)資源鎖住，只允許線程依次排隊(duì)進(jìn)去獲取這個(gè)資源。當(dāng)線程A操作完后，釋放鎖，線程B才能進(jìn)入。如下腳本是開啟多個(gè)線程修改變量的值，但輸出結(jié)果每次都不一樣。

import threading

money = 0
def Order(n):
 global money
 money = money + n
 money = money - n

class thread(threading.Thread):
 def __init__(self, threadname):
 threading.Thread.__init__(self, name='線程' + threadname)
 self.threadname = int(threadname)

 def run(self):
 for i in range(1000000):
  Order(self.threadname)

t1 = thread('1')
t2 = thread('5')
t1.start()
t2.start()
t1.join()
t2.join()
print(money)

接下來我們用 threading.Lock() 鎖住這個(gè)變量，等操作完再釋放這個(gè)鎖。lock.acquire() 給資源加一把鎖，對(duì)資源處理完成之后，lock.release() 再釋放鎖。以下腳本執(zhí)行結(jié)果都是一樣的，但速度會(huì)變慢，因?yàn)榫€程只能一個(gè)個(gè)的通過。

import threading

money = 0
def Order(n):
 global money
 money = money + n
 money = money - n

class thread(threading.Thread):
 def __init__(self, threadname):
 threading.Thread.__init__(self, name='線程' + threadname)
 self.threadname = int(threadname)

 def run(self):
 for i in range(1000000):
  lock.acquire()
  Order(self.threadname)
  lock.release()
# print('%s:Now timestamp is %s'%(self.name,time.time()))

lock = threading.Lock()
t1 = thread('1')
t2 = thread('5')
t1.start()
t2.start()
t1.join()
t2.join()
print(money)

2.threading.Rlock()

用法和 threading Lock() 一致，區(qū)別是 threading.Rlock() 允許多次鎖資源，acquire() 和 release() 必須成對(duì)出現(xiàn)，也就是說加了幾把鎖就得釋放幾把鎖。

lock = threading.Lock()
# 死鎖
lock.acquire()
lock.acquire()
print('...')
lock.release()
lock.release()

rlock = threading.RLock()
# 同一線程內(nèi)不會(huì)阻塞線程
rlock.acquire()
rlock.acquire()
print('...')
rlock.release()
rlock.release()

3.threading.Condition()

threading.Condition() 可以理解為更加高級(jí)的鎖，比 Lock 和 Rlock 的用法更高級(jí)，能處理一些復(fù)雜的線程同步問題。threading.Condition() 創(chuàng)建一把資源鎖（默認(rèn)是Rlock），提供 acquire() 和 release() 方法，用法和 Rlock 一致。此外 Condition 還提供 wait()、Notify() 和 NotifyAll() 方法。

wait()：線程掛起，直到收到一個(gè) Notify() 通知或者超時(shí)（可選參數(shù)），wait() 必須在線程得到 Rlock 后才能使用。

Notify() ：在線程掛起的時(shí)候，發(fā)送一個(gè)通知，讓 wait() 等待線程繼續(xù)運(yùn)行，Notify() 也必須在線程得到 Rlock 后才能使用。 Notify(n=1)，最多喚醒 n 個(gè)線程。

NotifyAll() ：在線程掛起的時(shí)候，發(fā)送通知，讓所有 wait() 阻塞的線程都繼續(xù)運(yùn)行。

舉例說明下 Condition() 使用

import threading,time

def TestA():
 cond.acquire()
 print('李白：看見一個(gè)敵人，請(qǐng)求支援')
 cond.wait()
 print('李白：好的')
 cond.notify()
 cond.release()

def TestB():
 time.sleep(2)
 cond.acquire()
 print('亞瑟：等我...')
 cond.notify()
 cond.wait()
 print('亞瑟：我到了，發(fā)起沖鋒...')

if __name__=='__main__':
 cond = threading.Condition()
 testA = threading.Thread(target=TestA)
 testB = threading.Thread(target=TestB)
 testA.start()
 testB.start()
 testA.join()
 testB.join()

輸出
李白：看見一個(gè)敵人，請(qǐng)求支援
亞瑟：等我...
李白：好的
亞瑟：我到了，發(fā)起沖鋒...

4.threading.Event()

threading.Event() 原理是在線程中立了一個(gè) Flag ，默認(rèn)值是 False ，當(dāng)一個(gè)或多個(gè)線程遇到 event.wait() 方法時(shí)阻塞，直到 Flag 值變?yōu)?True 。threading.Event() 通常用來實(shí)現(xiàn)線程之間的通信，使一個(gè)線程等待其他線程的通知，把 Event 傳遞到線程對(duì)象中。

event.wait() ：阻塞線程，直到 Flag 值變?yōu)?True

event.set() ：設(shè)置 Flag 值為 True

event.clear() ：修改 Flag 值為 False

event.isSet() : 僅當(dāng) Flag 值為 True 時(shí)返回

下面這個(gè)例子，主線程啟動(dòng)子線程后 sleap 2秒，子線程因?yàn)?event.wait() 被阻塞。當(dāng)主線程醒來后執(zhí)行 event.set() ，子線程才繼續(xù)運(yùn)行，兩者輸出時(shí)間差 2s。

import threading
import datetime,time

class thread(threading.Thread):
 def __init__(self, threadname):
 threading.Thread.__init__(self, name='線程' + threadname)
 self.threadname = int(threadname)

 def run(self):
 event.wait()
 print('子線程運(yùn)行時(shí)間：%s'%datetime.datetime.now())

if __name__ == '__main__':
 event = threading.Event()
 t1 = thread('0')
 #啟動(dòng)子線程
 t1.start()
 print('主線程運(yùn)行時(shí)間：%s'%datetime.datetime.now())
 time.sleep(2)
 # Flag設(shè)置成True
 event.set()
 t1.join()