一、threading模塊
multiprocess模塊的完全模仿了threading模塊的接口，二者在使用層面，有很大的相似性。
1.開(kāi)啟線程的兩種方式（同Process）
方法一
from threading import Threadimport timedef sayhi(name):
time.sleep(2) print('%s say hello' %name)if __name__ == '__main__':
t=Thread(target=sayhi,args=('hh',))
t.start() print('主線程')
主線程 hh say hello
方法二
from threading import Threadimport timeclass Sayhi(Thread): def __init__(self,name):
super().__init__()
self.name=name def run(self):
time.sleep(2) print('%s say hello' % self.name)if __name__ == '__main__':
t = Sayhi('hh')
t.start() print('主線程')
主線程 hh say hello
2.在一個(gè)進(jìn)程下開(kāi)啟多個(gè)線程與在一個(gè)進(jìn)程下開(kāi)啟多個(gè)子進(jìn)程的區(qū)別
開(kāi)啟速度，主進(jìn)程下開(kāi)啟線程速度較快。
from threading import Threadfrom multiprocessing import Processimport osdef work(): print('hello')if __name__ == '__main__': #在主進(jìn)程下開(kāi)啟線程
t=Thread(target=work)
t.start() print('主進(jìn)程-->線程') '''
打印結(jié)果:
hello
主進(jìn)程-->線程 '''
#在主進(jìn)程下開(kāi)啟子進(jìn)程
t=Process(target=work)
t.start() print('主進(jìn)程-->子進(jìn)程') '''
打印結(jié)果:
主進(jìn)程-->子進(jìn)程
hello '''
hello 主進(jìn)程-->線程 主進(jìn)程-->子進(jìn)程 hello
開(kāi)啟PID
在主進(jìn)程下開(kāi)啟多個(gè)線程,每個(gè)線程都跟主進(jìn)程的pid一樣，開(kāi)多個(gè)進(jìn)程,每個(gè)進(jìn)程都有不同的pid
from threading import Threadfrom multiprocessing import Processimport osdef work(): print('hello',os.getpid())if __name__ == '__main__': #part1:在主進(jìn)程下開(kāi)啟多個(gè)線程,每個(gè)線程都跟主進(jìn)程的pid一樣
t1=Thread(target=work)
t2=Thread(target=work)
t1.start()
t2.start() print('主進(jìn)程-->線程pid',os.getpid()) #part2:開(kāi)多個(gè)進(jìn)程,每個(gè)進(jìn)程都有不同的pid
p1=Process(target=work)
p2=Process(target=work)
p1.start()
p2.start() print('主進(jìn)程-->子進(jìn)程pid',os.getpid())
hello 13002hello 13002主進(jìn)程-->線程pid 13002主進(jìn)程-->子進(jìn)程pid 13002hello 13003hello 13004
多線程并發(fā)的socket服務(wù)端
客戶端
三個(gè)任務(wù)，一個(gè)接收用戶輸入，一個(gè)將用戶輸入的內(nèi)容格式化成大寫，一個(gè)將格式化后的結(jié)果存入文件
from threading import Thread
msg_l=[]
format_l=[]def talk(): while True:
msg=input('>>: ').strip() if not msg:continue
msg_l.append(msg)def format_msg(): while True: if msg_l:
res=msg_l.pop()
format_l.append(res.upper())def save(): while True: if format_l:
with open('db.txt','a',encoding='utf-8') as f:
res=format_l.pop()
f.write('%s\n' %res)if __name__ == '__main__':
t1=Thread(target=talk)
t2=Thread(target=format_msg)
t3=Thread(target=save)
t1.start()
t2.start()
t3.start()
3.線程的join與setdaemon
與進(jìn)程的方法都是類似的，其實(shí)是multiprocessing模仿threading的接口
from threading import Threadimport timedef sayhi(name):
time.sleep(2) print('%s say hello' %name)if __name__ == '__main__':
t=Thread(target=sayhi,args=('h',))
t.setDaemon(True)
t.start()
t.join() print('主線程') print(t.is_alive())
4.線程相關(guān)的其他方法補(bǔ)充
Thread實(shí)例對(duì)象的方法 # isAlive(): 返回線程是否活動(dòng)的。 # getName(): 返回線程名。 # setName(): 設(shè)置線程名。 threading模塊提供的一些方法： # threading.currentThread(): 返回當(dāng)前的線程變量。 # threading.enumerate(): 返回一個(gè)包含正在運(yùn)行的線程的list。正在運(yùn)行指線程啟動(dòng)后、結(jié)束前，不包括啟動(dòng)前和終止后的線程。 # threading.activeCount(): 返回正在運(yùn)行的線程數(shù)量，與len(threading.enumerate())有相同的結(jié)果。
from threading import Threadimport threadingfrom multiprocessing import Processimport osdef work(): import time
time.sleep(3) print(threading.current_thread().getName())if __name__ == '__main__': #在主進(jìn)程下開(kāi)啟線程
t=Thread(target=work)
t.start() print(threading.current_thread().getName()) print(threading.current_thread()) #主線程
print(threading.enumerate()) #連同主線程在內(nèi)有兩個(gè)運(yùn)行的線程
print(threading.active_count()) print('主線程/主進(jìn)程')
MainThread<_MainThread(MainThread, started 140736678523840)>[<_MainThread(MainThread, started 140736678523840)>, <Thread(Thread-1, started 123145314349056)>]2主線程/主進(jìn)程 Thread-1
二、 Python GIL(Global Interpreter Lock)
全局解釋器鎖。在Cpython解釋器中，同一個(gè)進(jìn)程下開(kāi)啟的多線程，同一時(shí)刻只能有一個(gè)線程執(zhí)行，無(wú)法利用多核優(yōu)勢(shì)。
首先需要明確的一點(diǎn)是GIL并不是Python的特性，它是在實(shí)現(xiàn)Python解析器(CPython)時(shí)所引入的一個(gè)概念。就好比C++是一套語(yǔ)言（語(yǔ)法）標(biāo)準(zhǔn)，但是可以用不同的編譯器來(lái)編譯成可執(zhí)行代碼。有名的編譯器例如GCC，INTEL C++，Visual C++等。Python也一樣，同樣一段代碼可以通過(guò)CPython，PyPy，Psyco等不同的Python執(zhí)行環(huán)境來(lái)執(zhí)行。像其中的JPython就沒(méi)有GIL。然而因?yàn)镃Python是大部分環(huán)境下默認(rèn)的Python執(zhí)行環(huán)境。所以在很多人的概念里CPython就是Python，也就想當(dāng)然的把GIL歸結(jié)為Python語(yǔ)言的缺陷。所以這里要先明確一點(diǎn)：GIL并不是Python的特性，Python完全可以不依賴于GIL
補(bǔ)充：
1. cpu到底是用來(lái)做計(jì)算的，還是用來(lái)做I/O的？
1. 多cpu，意味著可以有多個(gè)核并行完成計(jì)算，所以多核提升的是計(jì)算性能
2. 每個(gè)cpu一旦遇到I/O阻塞，仍然需要等待，所以多核對(duì)I/O操作沒(méi)什么用處
一個(gè)工人相當(dāng)于cpu，此時(shí)計(jì)算相當(dāng)于工人在干活，I/O阻塞相當(dāng)于為工人干活提供所需原材料的過(guò)程，工人干活的過(guò)程中如果沒(méi)有原材料了，則工人干活的過(guò)程需要停止，直到等待原材料的到來(lái)。
如果你的工廠干的大多數(shù)任務(wù)都要有準(zhǔn)備原材料的過(guò)程（I/O密集型），那么你有再多的工人，意義也不大，還不如一個(gè)人，在等材料的過(guò)程中讓工人去干別的活，
反過(guò)來(lái)講，如果你的工廠原材料都齊全，那當(dāng)然是工人越多，效率越高
結(jié)論：
對(duì)計(jì)算來(lái)說(shuō)，cpu越多越好，但是對(duì)于I/O來(lái)說(shuō)，再多的cpu也沒(méi)用
我們有四個(gè)任務(wù)需要處理，處理方式肯定是要玩出并發(fā)的效果，解決方案可以是：
方案一：開(kāi)啟四個(gè)進(jìn)程
方案二：一個(gè)進(jìn)程下，開(kāi)啟四個(gè)線程
單核情況下，分析結(jié)果:
如果四個(gè)任務(wù)是計(jì)算密集型，沒(méi)有多核來(lái)并行計(jì)算，方案一徒增了創(chuàng)建進(jìn)程的開(kāi)銷，方案二勝
如果四個(gè)任務(wù)是I/O密集型，方案一創(chuàng)建進(jìn)程的開(kāi)銷大，且進(jìn)程的切換速度遠(yuǎn)不如線程，方案二勝
多核情況下，分析結(jié)果：
如果四個(gè)任務(wù)是計(jì)算密集型，多核意味著并行計(jì)算，在python中一個(gè)進(jìn)程中同一時(shí)刻只有一個(gè)線程執(zhí)行用不上多核，方案一勝
如果四個(gè)任務(wù)是I/O密集型，再多的核也解決不了I/O問(wèn)題，方案二勝
結(jié)論：現(xiàn)在的計(jì)算機(jī)基本上都是多核，python對(duì)于計(jì)算密集型的任務(wù)開(kāi)多線程的效率并不能帶來(lái)多大性能上的提升，甚至不如串行(沒(méi)有大量切換)，但是，對(duì)于IO密集型的任務(wù)效率還是有顯著提升的。
#計(jì)算密集型from threading import Threadfrom multiprocessing import Processimport osimport timedef work():
res=0 for i in range(1000000):
res+=iif __name__ == '__main__':
t_l=[]
start_time=time.time() # for i in range(300): #串行
# work()
for i in range(300): # t=Thread(target=work) #多線程49.64094281196594
t=Process(target=work) #多進(jìn)程11.664679050445557 t_l.append(t)
t.start() for i in t_l:
i.join()
stop_time=time.time() print('run time is %s' %(stop_time-start_time)) print('主線程')# run time is 49.64094281196594# 主線程# run time is 11.664679050445557# 主線程
#I/O密集型from threading import Threadfrom multiprocessing import Processimport timeimport osdef work():
time.sleep(2) #模擬I/O操作，可以打開(kāi)一個(gè)文件來(lái)測(cè)試I/O,與sleep是一個(gè)效果
print(os.getpid())if __name__ == '__main__':
t_l=[]
start_time=time.time() for i in range(1000):
t=Thread(target=work) #耗時(shí)大概為2秒
# t=Process(target=work) #耗時(shí)大概為25秒,創(chuàng)建進(jìn)程的開(kāi)銷遠(yuǎn)高于線程，而且對(duì)于I/O密集型，多cpu根本不管用 t_l.append(t)
t.start() for t in t_l:
t.join()
stop_time=time.time() print('run time is %s' %(stop_time-start_time))
結(jié)論：
多線程用于IO密集型，如socket，爬蟲，web
多進(jìn)程用于計(jì)算密集型，如金融分析
三、同步鎖
import timeimport threadingdef addNum(): global num #在每個(gè)線程中都獲取這個(gè)全局變量
#num-=1
temp=num
time.sleep(0.1)
num =temp-1 # 對(duì)此公共變量進(jìn)行-1操作num = 100 #設(shè)定一個(gè)共享變量thread_list = []for i in range(100):
t = threading.Thread(target=addNum)
t.start()
thread_list.append(t)for t in thread_list: #等待所有線程執(zhí)行完畢 t.join()print('Result: ', num)
Result: 99
鎖通常被用來(lái)實(shí)現(xiàn)對(duì)共享資源的同步訪問(wèn)。為每一個(gè)共享資源創(chuàng)建一個(gè)Lock對(duì)象，當(dāng)你需要訪問(wèn)該資源時(shí)，調(diào)用acquire方法來(lái)獲取鎖對(duì)象（如果其它線程已經(jīng)獲得了該鎖，則當(dāng)前線程需等待其被釋放），待資源訪問(wèn)完后，再調(diào)用release方法釋放鎖：
import threading R=threading.Lock() R.acquire()'''對(duì)公共數(shù)據(jù)的操作'''R.release()
補(bǔ)充：
GIL VS Lock
Python已經(jīng)有一個(gè)GIL來(lái)保證同一時(shí)間只能有一個(gè)線程來(lái)執(zhí)行了，為什么這里還需要lock?
首先我們需要達(dá)成共識(shí)：鎖的目的是為了保護(hù)共享的數(shù)據(jù)，同一時(shí)間只能有一個(gè)線程來(lái)修改共享的數(shù)據(jù)
然后，我們可以得出結(jié)論：保護(hù)不同的數(shù)據(jù)就應(yīng)該加不同的鎖。
最后，問(wèn)題就很明朗了，GIL 與Lock是兩把鎖，保護(hù)的數(shù)據(jù)不一樣，前者是解釋器級(jí)別的（當(dāng)然保護(hù)的就是解釋器級(jí)別的數(shù)據(jù)，比如垃圾回收的數(shù)據(jù)），后者是保護(hù)用戶自己開(kāi)發(fā)的應(yīng)用程序的數(shù)據(jù)，很明顯GIL不負(fù)責(zé)這件事，只能用戶自定義加鎖處理，即Lock
四、死鎖與遞歸鎖
死鎖： 是指兩個(gè)或兩個(gè)以上的進(jìn)程或線程在執(zhí)行過(guò)程中，因爭(zhēng)奪資源而造成的一種互相等待的現(xiàn)象，若無(wú)外力作用，它們都將無(wú)法推進(jìn)下去。此時(shí)稱系統(tǒng)處于死鎖狀態(tài)或系統(tǒng)產(chǎn)生了死鎖，這些永遠(yuǎn)在互相等待的進(jìn)程稱為死鎖進(jìn)程，如下就是死鎖
from threading import Thread,Lockimport time
mutexA=Lock()
mutexB=Lock()class MyThread(Thread): def run(self):
self.func1()
self.func2() def func1(self):
mutexA.acquire() print('\033[41m%s 拿到A鎖\033[0m' %self.name)
mutexB.acquire() print('\033[42m%s 拿到B鎖\033[0m' %self.name)
mutexB.release()
mutexA.release() def func2(self):
mutexB.acquire() print('\033[43m%s 拿到B鎖\033[0m' %self.name)
time.sleep(2)
mutexA.acquire() print('\033[44m%s 拿到A鎖\033[0m' %self.name)
mutexA.release()
mutexB.release()if __name__ == '__main__': for i in range(10):
t=MyThread()
t.start()'''Thread-1 拿到A鎖
Thread-1 拿到B鎖
Thread-1 拿到B鎖
Thread-2 拿到A鎖
然后就卡住，死鎖了'''
解決方法，遞歸鎖，在Python中為了支持在同一線程中多次請(qǐng)求同一資源，python提供了可重入鎖RLock。
這個(gè)RLock內(nèi)部維護(hù)著一個(gè)Lock和一個(gè)counter變量，counter記錄了acquire的次數(shù)，從而使得資源可以被多次require。直到一個(gè)線程所有的acquire都被release，其他的線程才能獲得資源。上面的例子如果使用RLock代替Lock，則不會(huì)發(fā)生死鎖：
mutexA=mutexB=threading.RLock() #一個(gè)線程拿到鎖，counter加1,該線程內(nèi)又碰到加鎖的情況，則counter繼續(xù)加1，這期間所有其他線程都只能等待，等待該線程釋放所有鎖，即counter遞減到0為止
五、信號(hào)量Semahpore
Semaphore管理一個(gè)內(nèi)置的計(jì)數(shù)器，
每當(dāng)調(diào)用acquire()時(shí)內(nèi)置計(jì)數(shù)器-1；
調(diào)用release() 時(shí)內(nèi)置計(jì)數(shù)器+1；
計(jì)數(shù)器不能小于0；當(dāng)計(jì)數(shù)器為0時(shí)，acquire()將阻塞線程直到其他線程調(diào)用release()。
實(shí)例：(同時(shí)只有5個(gè)線程可以獲得semaphore,即可以限制最大連接數(shù)為5)：
import threadingimport time semaphore = threading.Semaphore(5)def func(): if semaphore.acquire(): print (threading.currentThread().getName() + ' get semaphore') time.sleep(2) semaphore.release()for i in range(20): t1 = threading.Thread(target=func) t1.start()
與進(jìn)程池是完全不同的概念，進(jìn)程池Pool(4)，最大只能產(chǎn)生4個(gè)進(jìn)程，而且從頭到尾都只是這四個(gè)進(jìn)程，不會(huì)產(chǎn)生新的，而信號(hào)量是產(chǎn)生一堆線程/進(jìn)程
六、Event
線程的一個(gè)關(guān)鍵特性是每個(gè)線程都是獨(dú)立運(yùn)行且狀態(tài)不可預(yù)測(cè)。如果程序中的其 他線程需要通過(guò)判斷某個(gè)線程的狀態(tài)來(lái)確定自己下一步的操作,這時(shí)線程同步問(wèn)題就 會(huì)變得非常棘手。為了解決這些問(wèn)題,我們需要使用threading庫(kù)中的Event對(duì)象。 對(duì)象包含一個(gè)可由線程設(shè)置的信號(hào)標(biāo)志,它允許線程等待某些事件的發(fā)生。在 初始情況下,Event對(duì)象中的信號(hào)標(biāo)志被設(shè)置為假。如果有線程等待一個(gè)Event對(duì)象, 而這個(gè)Event對(duì)象的標(biāo)志為假,那么這個(gè)線程將會(huì)被一直阻塞直至該標(biāo)志為真。一個(gè)線程如果將一個(gè)Event對(duì)象的信號(hào)標(biāo)志設(shè)置為真,它將喚醒所有等待這個(gè)Event對(duì)象的線程。如果一個(gè)線程等待一個(gè)已經(jīng)被設(shè)置為真的Event對(duì)象,那么它將忽略這個(gè)事件, 繼續(xù)執(zhí)行
event.isSet()：返回event的狀態(tài)值； event.wait()：如果 event.isSet()==False將阻塞線程； event.set()： 設(shè)置event的狀態(tài)值為True，所有阻塞池的線程激活進(jìn)入就緒狀態(tài)， 等待操作系統(tǒng)調(diào)度； event.clear()：恢復(fù)event的狀態(tài)值為False。
可以考慮一種應(yīng)用場(chǎng)景（僅僅作為說(shuō)明），例如，我們有多個(gè)線程從Redis隊(duì)列中讀取數(shù)據(jù)來(lái)處理，這些線程都要嘗試去連接Redis的服務(wù)，一般情況下，如果Redis連接不成功，在各個(gè)線程的代碼中，都會(huì)去嘗試重新連接。如果我們想要在啟動(dòng)時(shí)確保Redis服務(wù)正常，才讓那些工作線程去連接Redis服務(wù)器，那么我們就可以采用threading.Event機(jī)制來(lái)協(xié)調(diào)各個(gè)工作線程的連接操作：主線程中會(huì)去嘗試連接Redis服務(wù)，如果正常的話，觸發(fā)事件，各工作線程會(huì)嘗試連接Redis服務(wù)。
import threadingimport timeimport logging
logging.basicConfig(level=logging.DEBUG, format='(%(threadName)-10s) %(message)s',)def worker(event):
logging.debug('Waiting for redis ready...')
event.wait()http://www.cnblogs.com/smallmars/p/7149507.html
延伸閱讀
- 要想為事業(yè)多添一把火，自己就得多添一捆材 2023-04-18
- 多敲代碼，不斷試錯(cuò) 2022-09-08
- 多積累才能有進(jìn)步——?jiǎng)?聰 2019-10-28
- 方法總比困難多——蘇*誠(chéng) 2019-10-28
- 極少成多，聚沙成塔——呂*飛 2018-12-26
- JDBC&&c3p0、事務(wù)、批處理、多線程 于一體的經(jīng)典秘方QueryRunner （common-dbutils） 2017-07-26
- 認(rèn)識(shí)cpu、核與線程 2017-07-26
- 多線程之策略模式 2017-07-26
- Android系統(tǒng)--輸入系統(tǒng)（十四）Dispatcher線程情景分析_dispatch前處理 2017-07-25
- nopCommerce 3.9 大波浪系列 之 路由擴(kuò)展 [多語(yǔ)言Seo的實(shí)現(xiàn)] 2017-07-25
要想為事業(yè)多添一把火，自己就得多添一捆材