Python全棧之線程詳解

更新時(shí)間：2021年12月24日 17:14:53 作者：熬夜泡枸杞

這篇文章主要為大家介紹了Python全棧之線程，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下，希望能夠給你帶來幫助

1. 線程的概念

1.1 Manager_進(jìn)程通信

# ### Manager ( list 列表  ,  dict 字典 ) 進(jìn)程之間共享數(shù)據(jù)
from multiprocessing import Process , Manager ,Lock
def mywork(data,lock):
	# 共享字典
	"""
	lock.acquire()
	data["count"] -= 10
	lock.release()
	"""
	# 共享列表
	data[0] += 1
if __name__ == "__main__":
	lst = []
	m = Manager()
	lock = Lock()
	# 多進(jìn)程中的共享字典
	# data = m.dict(  {"count":5000}  )
	# print(data , type(data) )
	# 多進(jìn)程中的共享列表
	data =  m.list( [100,200,300] )
	# print(data , type(data) )
	""""""
	# 進(jìn)程數(shù)超過1000,處理該數(shù)據(jù),死機(jī)(謹(jǐn)慎操作)
	for i in range(10):
		p = Process(target=mywork,args=(data,lock))
		p.start()
		lst.append(p)
	# 必須等待子進(jìn)程所有計(jì)算完畢之后,再去打印該字典,否則報(bào)錯(cuò);
	for i in lst:
		i.join()
	print(data)

1.2 線程的概念

線程概念：

#進(jìn)程是資源分配的最小單位
#線程是計(jì)算機(jī)中調(diào)度的最小單位
#線程的緣起
資源分配需要分配內(nèi)存空間,分配cpu:
分配的內(nèi)存空間存放著臨時(shí)要處理的數(shù)據(jù)等,比如要執(zhí)行的代碼,數(shù)據(jù)
而這些內(nèi)存空間是有限的,不能無限分配
目前配置高的主機(jī),5萬個(gè)并發(fā)已是上限.線程概念應(yīng)用而生.
#線程的特點(diǎn)
線程是比較輕量級(jí),能干更多的活,一個(gè)進(jìn)程中的所有線程資源是共享的.
一個(gè)進(jìn)程至少有一個(gè)線程在工作

線程的缺陷：

#python中的線程可以并發(fā),但是不能并行(同一個(gè)進(jìn)程下的多個(gè)線程不能分開被多個(gè)cpu同時(shí)執(zhí)行)
#原因:
   全局解釋器鎖(Cpython解釋器特有) GIL鎖:
   同一時(shí)間,一個(gè)進(jìn)程下的多個(gè)線程只能被一個(gè)cpu執(zhí)行,不能實(shí)現(xiàn)線程的并行操作   
   python是解釋型語言,執(zhí)行一句編譯一句,而不是一次性全部編譯成功,不能提前規(guī)劃,都是臨時(shí)調(diào)度
   容易造成cpu執(zhí)行調(diào)度異常.所以加了一把鎖叫GIL   	
#想要并行的解決辦法:
    (1)用多進(jìn)程間接實(shí)現(xiàn)線程的并行
    (2)換一個(gè)Pypy,Jpython解釋器
#程序分為計(jì)算密集型和io密集型
	對(duì)于計(jì)算密集型程序會(huì)過度依賴cpu,但網(wǎng)頁,爬蟲,OA辦公,這種io密集型的程序里,python綽綽有余

小結(jié)：

進(jìn)程中的線程，同一時(shí)間只能有一個(gè)cup（單核來回切換進(jìn)行處理線程）來執(zhí)行，單核之間可以進(jìn)程切換
工作，以避免一個(gè)單核持續(xù)工作，過熱導(dǎo)致頻率降低。（java可以在同一時(shí)間進(jìn)行多核操作，也就是同步操作）
線程是執(zhí)行調(diào)度的最小單位
一個(gè)進(jìn)程包含多個(gè)線程，一個(gè)進(jìn)程至少一個(gè)線程
線程在一個(gè)進(jìn)程當(dāng)中可以共享一份資源
線程的缺陷：不能并行（java可以）
一個(gè)程序，至少一個(gè)主進(jìn)程和一個(gè)主線程

在這里插入圖片描述

2. 線程的基本使用

# ### 線程 
"""
進(jìn)程是資源分配的最小單元
線程是cpu執(zhí)行調(diào)度的最小單元
"""
# (1) 一個(gè)進(jìn)程里包含了多個(gè)線程,線程之間是異步并發(fā)
from threading import Thread
from multiprocessing import Process
import os , time , random
"""
def func(i):
	time.sleep(random.uniform(0.1,0.9))
	print("當(dāng)前進(jìn)程號(hào):{}".format(os.getpid()) , i)
if __name__ == "__main__":
	for i in range(10):
		t = Thread(target=func,args=(i,))
		t.start()
print(os.getpid())
"""
# (2) 并發(fā)的多進(jìn)程和多線程之間,多線程的速度更快
# 多線程速度
def func(i):
	print( "當(dāng)前進(jìn)程號(hào):{} , 參數(shù)是{} ".format(os.getpid() , i)  )
"""
if __name__ == "__main__":
	lst = []
	startime = time.time()
	for i in range(10000):
		t = Thread(target=func,args=(i,))
		t.start()
		lst.append(t)
	# print(lst)
	for i in lst:
		i.join()
	endtime = time.time()
	print("運(yùn)行的時(shí)間是{}".format(endtime - startime) ) # 運(yùn)行的時(shí)間是1.8805944919586182
"""
# 多進(jìn)程速度
"""
if __name__ == "__main__":
	lst = []
	startime = time.time()
	for i in range(10000):
		p = Process(target=func,args=(i,))
		p.start()
		lst.append(p)
	# print(lst)
	for i in lst:
		i.join()
	endtime = time.time()
	print("運(yùn)行的時(shí)間是{}".format(endtime - startime) ) # 運(yùn)行的時(shí)間是101.68004035949707
"""
# (3) 多線程之間,數(shù)據(jù)共享
num = 100
lst = []
def func():
	global num
	num -= 1
for i in range(100):
	t = Thread(target=func)
	t.start()
	lst.append(t)
for i in lst:
	i.join()
print(num)

在這里插入圖片描述

小提示：一個(gè)線程對(duì)變量進(jìn)行操作的時(shí)候，其他的線程就不能在對(duì)這個(gè)變量進(jìn)行操作。這個(gè)時(shí)候用鎖把線程鎖住。

3. 自定義線程_守護(hù)線程

3.1 自定義線程

# ### 用類定義線程
from threading import Thread
import os,time
# (1)必須繼承父類Thread,來自定義線程類
"""
class MyThread(Thread):
	def __init__(self,name):
		# 手動(dòng)調(diào)用父類的構(gòu)造方法
		super().__init__()
		# 自定義當(dāng)前類需要傳遞的參數(shù)
		self.name = name
	def run(self):
		print(  "當(dāng)前進(jìn)程號(hào){},name={}".format(os.getpid() , self.name)  )
if __name__ == "__main__":
	t = MyThread("我是線程")
	t.start()
	print( "當(dāng)前進(jìn)程號(hào){}".format(os.getpid()) )
"""
# ### 線程中的相關(guān)屬性
"""
# 線程.is_alive()    檢測線程是否仍然存在
# 線程.setName()     設(shè)置線程名字
# 線程.getName()     獲取線程名字
# 1.currentThread().ident 查看線程id號(hào) 
# 2.enumerate()        返回目前正在運(yùn)行的線程列表
# 3.activeCount()      返回目前正在運(yùn)行的線程數(shù)量
"""
"""
def func():
	time.sleep(1)
if __name__ == "__main__":
	t = Thread(target=func)
	t.start()
	# 檢測線程是否仍然存在
	print( t.is_alive() )
	# 線程.getName()     獲取線程名字
	print(t.getName())
	# 設(shè)置線程名字
	t.setName("抓API接口")
	print(t.getName())
"""
from threading import currentThread
from threading import enumerate
from threading import activeCount
def func():
	time.sleep(0.1)
	print("當(dāng)前子線程號(hào)id是{},進(jìn)程號(hào){}".format( currentThread().ident ,os.getpid()) )
if __name__ == "__main__":
	t = Thread(target=func)
	t.start()
	print("當(dāng)前主線程號(hào)id是{},進(jìn)程號(hào){}".format( currentThread().ident ,os.getpid()) )
	for i in range(5):
		t = Thread(target=func)
		t.start()
	# 返回目前正在運(yùn)行的線程列表
	lst = enumerate()
	print(lst,len(lst))
	# 返回目前正在運(yùn)行的線程數(shù)量 (了解)
	print(activeCount())

3.2 守護(hù)線程

# ### 守護(hù)線程 : 等待所有線程全部執(zhí)行完畢之后,自己在終止程序,守護(hù)所有線程
from threading import Thread
import time
def func1():
	while True:
		time.sleep(1)		
		print("我是函數(shù)func1")
def func2():
	print("我是func2  start ... ")
	time.sleep(3)
	print("我是func2  end ... ")
def func3():
	print("我是func3 start ... ")
	time.sleep(6)	
	print("我是func3  end ... ")
if __name__ == "__main__":
	t = Thread(target=func1)
	t2 = Thread(target=func2)
	t3 = Thread(target=func3)
	# 設(shè)置守護(hù)線程 (啟動(dòng)前設(shè)置)
	t.setDaemon(True)
	t.start()
	t2.start()
	t3.start()
	print("主線程執(zhí)行結(jié)束.... ")

4. 線程安全問題

4.1 線程安全問題

# ### 線程中的數(shù)據(jù)安全問題
from threading import  Thread  , Lock
import time
n = 0
def func1(lock):
	global n
	lock.acquire() 
	for i in range(1000000):		
		n += 1
	lock.release()
def func2(lock):
	global n
	# with語法可以簡化上鎖+解鎖的操作,自動(dòng)完成
	with lock:
		for i in range(1000000):
			n -= 1
if __name__ == "__main__":
	lst = []
	lock = Lock()
	start = time.time()
	for i in range(10):
		t1 = Thread(target=func1 ,args=(lock,) )
		t1.start()
		t2 = Thread(target=func2 ,args=(lock,) )
		t2.start()
		lst.append(t1)
		lst.append(t2)
	for i in lst:
		i.join()
	# print(lst,len(lst))
	end = time.time()
	print("主線程執(zhí)行結(jié)束... 當(dāng)前n結(jié)果為{} ,用時(shí){}".format(n , end-start))

4.2 Semaphore_信號(hào)量

# ### 信號(hào)量 Semaphore (線程)
"""同一時(shí)間對(duì)多個(gè)線程上多把鎖"""
from threading import Thread,Semaphore
import time , random
def func(i,sem):
	time.sleep(random.uniform(0.1,0.7))
	# with語法自動(dòng)實(shí)現(xiàn)上鎖 + 解鎖
	with sem:		
		print("我在電影院拉屎 .... 我是{}號(hào)".format(i))
if __name__ == "__main__":
	sem = Semaphore(5)
	for i in range(30):
		Thread(target=func,args=(i,sem)).start()
	print(1)
"""
	創(chuàng)建線程是異步的,
	上鎖的過程會(huì)導(dǎo)致程序變成同步;
"""

5. 死鎖_互斥鎖_遞歸鎖

加一把鎖,就對(duì)應(yīng)解一把鎖.形成互斥鎖.
從語法上來說,鎖可以互相嵌套,但不要使用,
不要因?yàn)檫壿媶栴}讓上鎖分成兩次.導(dǎo)致死鎖
遞歸鎖用于解決死鎖,但只是一種應(yīng)急的處理辦法

# ### 互斥鎖 死鎖 遞歸鎖
from threading import Thread , Lock , RLock
import time
# (1) 語法上的死鎖
"""語法上的死鎖: 是連續(xù)上鎖不解鎖"""
"""
lock = Lock()
lock.acquire()
# lock.acquire() error
print("代碼執(zhí)行中 ... 1")
lock.release()
lock.release()
"""
"""是兩把完全不同的鎖"""
lock1 = Lock()
lock2 = Lock()
lock1.acquire()
lock2.acquire()
print("代碼執(zhí)行中 ... 2")
lock2.release()
lock1.release()
# (2) 邏輯上的死鎖
""""""
noodles_lock = Lock()
kuaizi_lock = Lock()
def eat1(name):
	noodles_lock.acquire()
	print("{}搶到面條了 ... ".format(name))
	kuaizi_lock.acquire()
	print("{}搶到筷子了 ... ".format(name))
	print("開始享受香菇青菜面 ... ")
	time.sleep(0.5)
	kuaizi_lock.release()
	print("{}吃完了,滿意的放下了筷子".format(name))
	noodles_lock.release()
	print("{}吃完了,滿意的放下了面條".format(name))
def eat2(name):
	kuaizi_lock.acquire()
	print("{}搶到筷子了 ... ".format(name))
	noodles_lock.acquire()
	print("{}搶到面條了 ... ".format(name))
	print("開始享受香菇青菜面 ... ")
	time.sleep(0.5)	
	noodles_lock.release()
	print("{}吃完了,滿意的放下了面條".format(name))
	# kuaizi_lock.release()
	print("{}吃完了,滿意的放下了筷子".format(name))
if __name__ == "__main__":
	lst1 = ["康?？?,"張宇"]
	lst2 = ["張保張","趙沈陽"]
	for name in lst1:
		Thread(target=eat1,args=(name,)).start()
	for name in lst2:
		Thread(target=eat2,args=(name,)).start()
# (3) 使用遞歸鎖
"""
	遞歸鎖的提出專門用來解決死鎖現(xiàn)象
	用于快速解決線上項(xiàng)目死鎖問題
	即使連續(xù)上鎖,使用遞歸鎖后也形同虛設(shè),因?yàn)檫f歸鎖的作用在于解鎖;
"""
"""
# 基本語法
rlock = RLock()
rlock.acquire()
rlock.acquire()
rlock.acquire()
rlock.acquire()
print("代碼執(zhí)行中 ... 3")
rlock.release()
rlock.release()
rlock.release()
rlock.release()
"""
"""
noodles_lock = Lock()
kuaizi_lock = Lock()
# 讓noodles_lock和kuaizi_lock 都等于遞歸鎖
noodles_lock = kuaizi_lock = RLock()
def eat1(name):
	noodles_lock.acquire()
	print("{}搶到面條了 ... ".format(name))
	kuaizi_lock.acquire()
	print("{}搶到筷子了 ... ".format(name))
	print("開始享受香菇青菜面 ... ")
	time.sleep(0.5)
	kuaizi_lock.release()
	print("{}吃完了,滿意的放下了筷子".format(name))
	noodles_lock.release()
	print("{}吃完了,滿意的放下了面條".format(name))
def eat2(name):
	kuaizi_lock.acquire()
	print("{}搶到筷子了 ... ".format(name))
	noodles_lock.acquire()
	print("{}搶到面條了 ... ".format(name))
	print("開始享受香菇青菜面 ... ")
	time.sleep(0.5)	
	noodles_lock.release()
	print("{}吃完了,滿意的放下了筷子".format(name))
	kuaizi_lock.release()
	print("{}吃完了,滿意的放下了筷子".format(name))
if __name__ == "__main__":
	lst1 = ["康裕康","張宇"]
	lst2 = ["張保張","趙沈陽"]
	for name in lst1:
		Thread(target=eat1,args=(name,)).start()
	for name in lst2:
		Thread(target=eat2,args=(name,)).start()
"""
# (4) 盡量使用一把鎖解決問題,(少用鎖嵌套,容易邏輯死鎖)
"""
lock = Lock()
def eat1(name):
	lock.acquire()
	print("{}搶到面條了 ... ".format(name))
	print("{}搶到筷子了 ... ".format(name))
	print("開始享受香菇青菜面 ... ")
	time.sleep(0.5)	
	print("{}吃完了,滿意的放下了筷子".format(name))	
	print("{}吃完了,滿意的放下了面條".format(name))
	lock.release()
def eat2(name):
	lock.acquire()
	print("{}搶到筷子了 ... ".format(name))
	print("{}搶到面條了 ... ".format(name))
	print("開始享受香菇青菜面 ... ")
	time.sleep(0.5)	
	print("{}吃完了,滿意的放下了筷子".format(name))
	print("{}吃完了,滿意的放下了筷子".format(name))
	lock.release()
if __name__ == "__main__":
	lst1 = ["康?？?,"張宇"]
	lst2 = ["張保張","趙沈陽"]
	for name in lst1:
		Thread(target=eat1,args=(name,)).start()
	for name in lst2:
		Thread(target=eat2,args=(name,)).start()
"""

6. 線程事件

# ### 事件 Event
from threading import Thread , Event
import time,random
"""
wait   : 動(dòng)態(tài)加阻塞 (True => 放行  False => 阻塞)
is_set : 獲取內(nèi)部成員屬性值是True 還是 False
set    : 把False -> True
clear  : 把True  -> False
"""
# (1) 基本語法
"""
e = Event()
print(e.is_set())
e.set()
print(e.is_set())
e.wait()
e.clear()
# 最多阻塞三秒,放行
e.wait(3)
print("代碼執(zhí)行中 ... ")
"""
# (2) 模擬連接遠(yuǎn)程數(shù)據(jù)庫
"""最多連接三次,如果三次都連接不上,直接報(bào)錯(cuò)."""
def check(e):	
	print("目前正在檢測您的賬號(hào)和密碼 .... ")
	# 模擬延遲的場景
	time.sleep(random.randrange(1,7)) # 1 ~ 6
	# 把成員屬性值從False -> True
	e.set()
def connect(e):
	sign = False
	for i in range(1,4):
		e.wait(1)
		if e.is_set():
			print("數(shù)據(jù)庫連接成功 ... ")
			sign = True
			break
		else:
			print("嘗試連接數(shù)據(jù)庫第{}次失敗了...".format(i))
	# 三次都不成功,報(bào)錯(cuò)
	if sign == False:
		# 主動(dòng)拋出異常  超時(shí)錯(cuò)誤
		raise TimeoutError
# if __name__ == "__main__":
e = Event()
t1 = Thread(target=check,args=(e,))
t1.start()
t2 = Thread(target=connect,args=(e,))
t2.start()