一焰轻、線程
在一個(gè)進(jìn)程的內(nèi)部乳讥，要同時(shí)干多件事，就需要同時(shí)運(yùn)行多個(gè)“子任務(wù)”知残，我們把進(jìn)程內(nèi)的這些“子任務(wù)”叫做線程靠瞎。
線程通常叫做輕型的進(jìn)程。線程是共享內(nèi)存空間的并發(fā)執(zhí)行的多任務(wù)求妹，每一個(gè)線程都共享一個(gè)進(jìn)程的資源乏盐。
線程是最小的執(zhí)行單元，而進(jìn)程由至少一個(gè)線程組成制恍。如何調(diào)度進(jìn)程和線程父能，完全由操作系統(tǒng)決定，程序自己不能決定什么時(shí)候執(zhí)行净神，執(zhí)行多長(zhǎng)時(shí)間何吝。
注意進(jìn)程與線程的區(qū)分，進(jìn)程之間是CPU資源分配的最小單位且進(jìn)程與進(jìn)程之間的資源是不共享的
二强挫、線程的使用
- Python中的使用
一般我們通過(guò)Python的_thread模塊和threading模塊來(lái)實(shí)現(xiàn)多線程
_thread模塊是比較低級(jí)的模塊，更接近底層薛躬。threading模塊是高級(jí)模塊俯渤，對(duì)_thread模塊進(jìn)行了封裝
from threading import Thread, current_thread
import time
def run(num):
print('子線程（%s）開始' % (current_thread().name,))
time.sleep(2)
print('打印', num)
time.sleep(2)
print('子線程（%s）結(jié)束' % (current_thread().name,))
if __name__ == '__main__':
# current_thread()：返回返回當(dāng)前線程的實(shí)例
print('主線程（%s）開始' % (current_thread().name,))
# 創(chuàng)建子線程
t = Thread(target=run, args=(1,), name='runThread')
t.start()
# 等待線程結(jié)束
t.join()
print("主線程(%s)結(jié)束" % (current_thread().name))
任何進(jìn)程默認(rèn)就會(huì)啟動(dòng)一個(gè)線程，稱為主線程型宝，主線程可以啟動(dòng)新的子線程
- 常用的方法
| 方法名 | 說(shuō)明 |
|---|---|
| isAlive() | 返回線程是否在運(yùn)行八匠。正在運(yùn)行指啟動(dòng)后絮爷、終止前。 |
| get/setName(name) | 獲取/設(shè)置線程名梨树。 |
| start() | 線程準(zhǔn)備就緒坑夯，等待CPU調(diào)度 |
| is/setDaemon(bool) | 獲取/設(shè)置是守護(hù)線程（默認(rèn)前臺(tái)線程（False））。（在start之前設(shè)置） |
| join([timeout]) | 阻塞當(dāng)前上下文環(huán)境的線程抡四，直到調(diào)用此方法的線程終止或到達(dá)指定的timeout（可選參數(shù)） |
- 線程之間的數(shù)據(jù)共享
多線程之間的各個(gè)線程資源是共享的
from threading import Thread
from time import sleep
# 全局?jǐn)?shù)據(jù)
num = 100
def run():
print('子線程開始')
global num
num += 1
print('子線程結(jié)束')
if __name__ == '__main__':
print('主線程開始')
# 創(chuàng)建主線程
t = Thread(target=run)
t.start()
t.join()
print(num)
print('主線程結(jié)束')
# 輸出結(jié)果
主線程開始
子線程開始
子線程結(jié)束
101
主線程結(jié)束
子線程修改了全局變量num以后柜蜈，主線程輸出的num是已經(jīng)被修改過(guò)的num。注意指巡，雖然我只聲明了一個(gè)線程淑履，但是實(shí)際有兩個(gè)線程，即它的主線程和我開啟的子線程藻雪。
- 線程鎖
既然多線程之間資源是共享的,那當(dāng)我們的線程都修改某一個(gè)變量或者文件時(shí)會(huì)發(fā)生什么情況秘噪？
from threading import Thread
# 全局變量
num = 100
def run(n):
global num
for i in range(100000000):
num = num + n
num = num - n
if __name__ == '__main__':
t1 = Thread(target=run, args=(6,))
t2 = Thread(target=run, args=(9,))
t1.start()
t2.start()
t1.join()
t2.join()
print('num=',num)
運(yùn)行之后發(fā)現(xiàn)num不等于100了。這就是因?yàn)閮蓚€(gè)線程都對(duì)num進(jìn)行操作勉耀，中間發(fā)生了紊亂指煎。
因?yàn)榫€程的開啟沒有先后順序而言，當(dāng)有個(gè)線程+n后還沒來(lái)得及-n時(shí)便斥，另外一個(gè)線程開啟又對(duì)num進(jìn)行了+n造成num的值發(fā)生了改變至壤，后續(xù)的結(jié)果都會(huì)產(chǎn)生變化。如果你運(yùn)行后num的值沒有發(fā)生改變椭住，請(qǐng)多增加線程或是循環(huán)的次數(shù)崇渗。
所以這個(gè)時(shí)候我們就需要應(yīng)用到了線程鎖來(lái)保證同時(shí)只能有一個(gè)線程來(lái)讀取我們的數(shù)據(jù)，但鎖確保了這段代碼只能由一個(gè)線程從頭到尾的完整執(zhí)行京郑。阻止了多線程的并發(fā)執(zhí)行宅广，包含鎖的某段代碼實(shí)際上只能以單線程模式執(zhí)行，所以效率大大滴降低了些举。
由于可以存在多個(gè)鎖跟狱，不同線程持有不同的鎖，并試圖獲取其他的鎖户魏，可能造成死鎖驶臊，導(dǎo)致多個(gè)線程掛起。只能靠操作系統(tǒng)強(qiáng)制終止叼丑。
接下來(lái)关翎，我們給之前的代碼加上鎖 來(lái)解決數(shù)據(jù)混亂的問(wèn)題。
from threading import Thread, Lock
# 全局變量
num = 100
# 鎖對(duì)象
lock = Lock()
def run(n):
global num
global lock
for i in range(100000000):
# 獲取鎖
lock.acquire()
try:
num = num + n
num = num - n
finally:
# 修改完一定 要釋放鎖
lock.release()
if __name__ == '__main__':
t1 = Thread(target=run, args=(6,))
t2 = Thread(target=run, args=(9,))
t1.start()
t2.start()
t1.join()
t2.join()
print('num=',num)
注意：每次獲取鎖以后一定要釋放鎖
每次使用又要獲取鎖又要釋放鎖是不是很麻煩鸠信？
我們可以和操作文件一樣纵寝，使用上下文管理器 with，可以自動(dòng)獲取鎖星立，釋放鎖爽茴≡岬剩可以將上面代碼的try語(yǔ)句改成如下：
with lock:
num = num + n
num = num - n
三、線程的進(jìn)階
- 實(shí)現(xiàn)進(jìn)程的局部變量
出了上面介紹的進(jìn)程鎖以外室奏，我們還能通過(guò)ThreadLocal實(shí)現(xiàn)線程數(shù)據(jù)不共享火焰。即線程的局部變量。
import threading
num = 0
# 創(chuàng)建一個(gè)全局的ThreadLocal對(duì)象
# 每個(gè)線程有獨(dú)立的存儲(chǔ)空間
# 每個(gè)線程對(duì)ThreadLocal對(duì)象都可以讀寫胧沫， 但是互不影響
local = threading.local()
def run(x, n):
x = x + n
x = x - n
def func(n):
# 每個(gè)線程都有l(wèi)ocal.x ,就是線程的局部變量
local.x = num
for i in range(1000000):
run(local.x, n)
print('%s--%d' % (threading.current_thread().name, local.x))
if __name__ == '__main__':
t1 = threading.Thread(target=func, args=(6,))
t2 = threading.Thread(target=func, args=(9,))
t1.start()
t2.start()
t1.join()
t2.join()
print("num=",num)
我們能從結(jié)果看到昌简。num最后結(jié)果是0，那是因?yàn)閚um被設(shè)置成了線程的局部變量琳袄，每個(gè)線程操作的都是自己的局部變量江场，所以不會(huì)和其他線程的數(shù)據(jù)發(fā)生混亂。
- 使用信號(hào)量來(lái)限制線程并發(fā)數(shù)
Semphore叫做信號(hào)量窖逗，可以限制線程的并發(fā)數(shù)址否，是一種帶計(jì)數(shù)的線程同步機(jī)制，當(dāng)調(diào)用release時(shí)碎紊，增加計(jì)算佑附，當(dāng)acquire時(shí)，減少計(jì)數(shù)仗考，當(dāng)計(jì)數(shù)為0時(shí)音同，自動(dòng)阻塞，等待release被調(diào)用秃嗜。
在大部分情況下权均，信號(hào)量用于守護(hù)有限容量的資源。
import threading
import time
# 創(chuàng)建一個(gè)信號(hào)量實(shí)例锅锨，限制并行的線程數(shù)為3個(gè)
sem = threading.Semaphore(3)
def run(i):
# 獲取信號(hào)量叽赊，信號(hào)量減1
sem.acquire()
print('%s--%d' % (threading.current_thread().name, i))
time.sleep(2)
# 釋放信號(hào)量，信號(hào)量加1
sem.release()
if __name__ == '__main__':
# 創(chuàng)建5個(gè)線程
for i in range(5):
threading.Thread(target=run, args=(i,)).start()
我們能看到前三個(gè)線程會(huì)先執(zhí)行必搞，前三個(gè)線程執(zhí)行完了之后才會(huì)執(zhí)行后2個(gè)線程必指。
還有另一種信號(hào)量叫做BoundedSemaphore，比普通的Semphore更加嚴(yán)謹(jǐn)恕洲，叫做有界信號(hào)量塔橡。有界信號(hào)量會(huì)確保它當(dāng)前的值不超過(guò)它的初始值。如果超過(guò)霜第，則引發(fā)ValueError葛家。
- Barrier對(duì)象
Barrier翻譯成柵欄，可以理解為線程數(shù)量不夠時(shí)泌类，會(huì)被攔住不讓執(zhí)行癞谒。我們來(lái)看一個(gè)例子：
import threading
import time
bar = threading.Barrier(3)
def run(i):
print('%s--%d--start' % (threading.current_thread().name, i))
time.sleep(2)
bar.wait()
print('%s--%d--end' % (threading.current_thread().name, i))
if __name__ == '__main__':
# 創(chuàng)建5個(gè)線程
for i in range(5):
threading.Thread(target=run, args=(i,)).start()
#輸出結(jié)果：
Thread-1--0--start
Thread-2--1--start
Thread-3--2--start
Thread-4--3--start
Thread-5--4--start
Thread-4--3--end
Thread-2--1--end
Thread-1--0--end
我們能看到執(zhí)行完3個(gè)線程之后，程序一直停著。那是因?yàn)楹竺娴木€程不夠3個(gè)扯俱，被柵欄攔住了沒法繼續(xù)執(zhí)行。如果我們開啟六個(gè)線程喇澡，那么程序就會(huì)全部執(zhí)行迅栅。
- 定時(shí)執(zhí)行線程
threading中的Timer可以讓線程在指定時(shí)間之后執(zhí)行 ，實(shí)現(xiàn)定時(shí)執(zhí)行的效果晴玖。
import threading
import time
def run():
print('start...')
time.sleep(2)
print('end...')
if __name__ == '__main__':
timer = threading.Timer(5, run)
timer.start()
print('end main...')
- 最簡(jiǎn)單的線程通信機(jī)制——Event
Event（事件）是最簡(jiǎn)單的線程通信機(jī)制之一：一個(gè)線程通知事件读存，其他線程等待事件。Event內(nèi)置了一個(gè)初始為False的標(biāo)志（flag）呕屎，當(dāng)調(diào)用set()時(shí)設(shè)為True让簿，調(diào)用clear()時(shí)重置為 False。wait()將阻塞線程至等待阻塞狀態(tài)秀睛。
來(lái)看一個(gè)簡(jiǎn)單的例子：
import threading
import time
# 創(chuàng)建一個(gè)事件
event = threading.Event()
def run():
print('start...')
# 進(jìn)入到等待事件的阻塞狀態(tài)中
event.wait()
time.sleep(2)
print('end...')
if __name__ == '__main__':
t1 = threading.Thread(target=run)
t1.start()
time.sleep(2)
# 發(fā)送事件
print('sending event...')
event.set()
注意：clear會(huì)把內(nèi)置的標(biāo)志重新設(shè)為False尔当。
- 生產(chǎn)者消費(fèi)者模型
生產(chǎn)者消費(fèi)者模型是多線程中常見的一種模型。先來(lái)看一個(gè)簡(jiǎn)單的例子蹂安。
import threading, queue, time, random
# 生產(chǎn)者
def product(id, q):
while True:
num = random.randint(0, 10000)
q.put(num)
print('生產(chǎn)者%d生成了%d數(shù)據(jù)放入了隊(duì)列' % (id, num))
time.sleep(1)
# 告訴隊(duì)列任務(wù)完成
q.task_done()
# 消費(fèi)者
def consumer(id, q):
while True:
item = q.get()
if item is None:
break
print('消費(fèi)者%d消費(fèi)了%d數(shù)據(jù)' % (id, item))
time.sleep(1)
# 任務(wù)完成
q.task_done()
if __name__ == '__main__':
# 消息隊(duì)列
q = queue.Queue()
# 啟動(dòng)生產(chǎn)者
for i in range(4):
threading.Thread(target=product, args=(i, q)).start()
# 啟動(dòng)消費(fèi)者
for i in range(3):
threading.Thread(target=consumer, args=(i, q)).start()
生產(chǎn)者生成的數(shù)據(jù)存放在隊(duì)列q中椭迎，消費(fèi)者去隊(duì)列中取數(shù)據(jù)。
- 線程調(diào)度
正常情況下線程的執(zhí)行順序是操作系統(tǒng)控制的田盈，如果需要讓我們自己來(lái)控制線程的執(zhí)行順序畜号，需要用到Condition（條件變量）類。
嘗試實(shí)現(xiàn)這么一個(gè)需求允瞧，有一個(gè)線程輸出 0 2 4 6 8 简软，另一個(gè)線程輸出 1 3 5 7 9 ，這兩個(gè)線程并行執(zhí)行時(shí)述暂，顯示的結(jié)果是混亂無(wú)序的痹升，要求輸出結(jié)果為0 1 2 3 4 5 6 7 8 9 。
import threading, time
# 線程條件變量
cond = threading.Condition()
def run1():
for i in range(0, 10, 2):
# condition自帶一把鎖
# 獲取鎖
if cond.acquire():
print(threading.current_thread().name, i)
time.sleep(1)
# 當(dāng)前線程執(zhí)行完成贸典，等待另一個(gè)線程執(zhí)行视卢，并釋放鎖
cond.wait()
# 通知另一個(gè)線程可以運(yùn)行了。
cond.notify()
def run2():
for i in range(1, 10, 2):
# 獲取鎖
if cond.acquire():
print(threading.current_thread().name, i)
time.sleep(1)
# 通知上面的線程你不要等了廊驼，該走了据过。
cond.notify()
# 然后等待上一個(gè)線程給自己繼續(xù)執(zhí)行的信號(hào)。
cond.wait()
if __name__ == '__main__':
threading.Thread(target=run1).start()
threading.Thread(target=run2).start()
Condition類的方法說(shuō)明：
acquire([timeout])/release(): 調(diào)用關(guān)聯(lián)的鎖的相應(yīng)方法妒挎。
wait([timeout]): 調(diào)用這個(gè)方法將使線程進(jìn)入Condition的等待池等待通知绳锅，并釋放鎖。使用前線程必須已獲得鎖定酝掩，否則將拋出異常鳞芙。
notify(): 調(diào)用這個(gè)方法將從等待池挑選一個(gè)線程并通知，收到通知的線程將自動(dòng)調(diào)用acquire()嘗試獲得鎖定（進(jìn)入鎖定池）；其他線程仍然在等待池中原朝。調(diào)用這個(gè)方法不會(huì)釋放鎖定驯嘱。使用前線程必須已獲得鎖定，否則將拋出異常喳坠。
notifyAll(): 調(diào)用這個(gè)方法將通知等待池中所有的線程鞠评，這些線程都將進(jìn)入鎖定池嘗試獲得鎖定。調(diào)用這個(gè)方法不會(huì)釋放鎖定壕鹉。使用前線程必須已獲得鎖定剃幌，否則將拋出異常。
condition同樣可以使用with上下文管理器來(lái)自動(dòng)管理鎖晾浴。
四负乡、多線程的應(yīng)用
socket編程中多線程的應(yīng)用。單線程中脊凰，socket服務(wù)器在監(jiān)聽時(shí)抖棘，處于阻塞狀態(tài)，只能同時(shí)處理一個(gè)客戶端的連接狸涌，使用多線程钉答，讓服務(wù)器可以同時(shí)處理多個(gè)客戶端連接。
服務(wù)器端代碼：
import socket
import threading
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(('10.36.137.19',8081))
server.listen(5)
def run(ck):
data = ck.recv(1024)
print("客戶端說(shuō)：" + data.decode("utf-8"))
ck.send("nice to meet you too".encode("utf-8"))
print("服務(wù)器啟動(dòng)成功杈抢，等待客戶端的鏈接")
while True:
clientSocket, clientAddress = server.accept()
t = threading.Thread(target=run, args=(clientSocket,))
t.start()
客戶端代碼：
import socket
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.connect(("10.36.137.19", 8081))
while True:
data = input("請(qǐng)輸入給服務(wù)器發(fā)送的數(shù)據(jù)")
client.send(data.encode("utf-8"))
info = client.recv(1024)
print("服務(wù)器說(shuō)：", info.decode("utf-8"))
