Python進程、線程、協(xié)程詳解

1、Python進程、線程、協(xié)程詳解進程與線程的歷史我們都知道計算機是由硬件和軟件組成的。硬件中的CPIM計算機的核心，它承擔計算機的所有任務(wù)。操作系統(tǒng)是運行在硬件之上的軟件，是計算機的管理者，它負責資源的管理和分配、任務(wù)的調(diào)度。程序是運行在系統(tǒng)上的具有某種功能的軟件，比如說瀏覽器，首樂播放器等。每次執(zhí)彳了程序的時候，都會完成一定的功能，比如說瀏覽器幫我們打開網(wǎng)頁，為了保證其獨立性，就需要一個專門的管理和控制執(zhí)行程序的數(shù)據(jù)結(jié)構(gòu)一一進程控制塊。進程就是一個程序在一個數(shù)據(jù)集上的一次動態(tài)執(zhí)行過程。進程一般由程序、數(shù)據(jù)集、進程控制塊三部分組成。我們編寫的程序用來描述進程要完成哪些功能以及如何完成；數(shù)據(jù)集則

2、是程序在執(zhí)行過程中所需要使用的資源；進程控制塊用來記錄進程的外部特征，描述進程的執(zhí)行變化過程，系統(tǒng)可以利用它來控制和管理進程，它是系統(tǒng)感知進程存在的唯一標志。在早期的操作系統(tǒng)里，計算機只有一個核心，進程執(zhí)行程序的最小單位，任務(wù)調(diào)度采用時間片輪轉(zhuǎn)的搶占式方式進行進程調(diào)度。每個進程都有各自的一塊獨立的內(nèi)存，保證進程彼此間的內(nèi)存地址空間的隔離。隨著計算機技術(shù)的發(fā)展，進程出現(xiàn)了很多弊端，一是進程的創(chuàng)建、撤銷和切換的開銷比較大，二是由于對稱多處理機(對稱多處理機(SymmetricalMulti-Processing)又叫SMP是指在一個計算機上匯集了一組處理器（多CPU各CP比間共享內(nèi)存子系統(tǒng)以及總線

3、結(jié)構(gòu)）的出現(xiàn)，可以滿足多個運行單位，而多進程并行開銷過大。這個時候就引入了線程的概念。線程也叫輕量級進程，它是一個基本的cpii行單元，也是程序執(zhí)行過程中的最小單元，由線程ID、程序計數(shù)器、寄存器集合和堆棧共同組成。線程的引入減小了程序并發(fā)執(zhí)行時的開銷，提高了操作系統(tǒng)的并發(fā)性能。線程沒有自己的系統(tǒng)資源，只擁有在運行時必不可少的資源。但線程可以與同屬與同一進程的其他線程共享進程所擁有的其他資源。進程與線程之間的關(guān)系線程是屬于進程的，線程運行在進程空間內(nèi)，同一進程所產(chǎn)生的線程共享同一內(nèi)存空間，當進程退出時該進程所產(chǎn)生的線程都會被強制退出并清除。線程可與屬于同一進程的其它線程共享進程所擁有的全部資源

4、，但是其本身基本上不擁有系統(tǒng)資源，只擁有一點在運行中必不可少的信息（如程序計數(shù)器、一組寄存器和棧）。python線程Threading用于提供線程相關(guān)的操作，線程是應(yīng)用程序中工作的最小單元。1、threading模塊threading模塊建立在_thread模塊之上。thread模塊以低級、原始的方式來處理和控制線程，而threading模塊通過對CPUthread進行二次封裝，提供了更方便的api來處理線程。1importthreading2importtime34defworker(num):5"""6threadworkerfunction7:return:

5、8"""9time.sleep(1)10print("Thenumis%d"%num)11return1213foriinrange(20):14t=threading.Thread(target=worker,args=(i,),name="t.%T%15t.start()上述代碼創(chuàng)建了20個“前臺”線程，然后控制器就交給了CPU艮據(jù)指定算法進行調(diào)度，分片執(zhí)行指令。Thread方法說明t.start():激活線程，t.getName():獲取線程的名稱t.setName():設(shè)置線程的名稱:獲取或設(shè)置線程的名稱t.is_

6、alive():判斷線程是否為激活狀態(tài)t.isAlive():判斷線程是否為激活狀態(tài)t.setDaemon()設(shè)置為后臺線程或前臺線程(默認：False)通過一個布爾值設(shè)置線程是否為守護線程，必須在執(zhí)行start()方法之后才可以使用。如果是后臺線程，主線程執(zhí)行過程中，后臺線程也在進行，主線程執(zhí)行完畢后，后臺線程不論成功與否，均停止；如果是前臺線程，主線程執(zhí)行過程中，前臺線程也在進行，主線程執(zhí)行完畢后，等待前臺線程也執(zhí)行完成后，程序停止t.isDaemon():判斷是否為守護線程t.ident:獲取線程的標識符。線程標識符是一個非零整數(shù)，只有在調(diào)用了start()方法之后該屬性才有效，否則它只

7、返回Nonet.join():逐個執(zhí)行每個線程，執(zhí)行完畢后繼續(xù)往下執(zhí)行，該方法使得多線程變得無意義t.run():線程被cpu調(diào)度后自動執(zhí)行線程對象的run方法2、線程缺threading.RLock和threading.Lock由于線程之間是進行隨機調(diào)度，并且每個線程可能只執(zhí)行n條執(zhí)行之后，CPUS著執(zhí)行其他線程。為了保證數(shù)據(jù)的準確性，引入了鎖的概念。所以，可能出現(xiàn)如下問題：例：假設(shè)列表A的所有元素就為0,當一個線程從前向后打印列表的所有元素，另外一個線程則從后向前修改列表的元素為1,那么輸出的時候，列表的元素就會一部分為0,一部分為1,這就導(dǎo)致了數(shù)據(jù)的不一致。鎖的出現(xiàn)解決了這個問題。1 i

8、mportthreading2 importtime4globals_num=056lock=threading.RLock()78defFunc():9lock.acquire()#獲得鎖10globalglobals_num11globals_num+=112time.sleep(1)13print(globals_num)14lock.release()#釋放鎖1516foriinrange(10):17t=threading.Thread(target18t.start()=Func)3、threading.RLock和threading.Lock的區(qū)別RLock允許在同一線程中被多次

9、acquire。而Lock卻不允許這種情況。如果使用RLock，那么acquire和release必須成對出現(xiàn)，即調(diào)用了n次acquire,必須調(diào)用n次的release才能真正釋放所占用的瑣。importthreading2lock=threading.Lock()#Lock對象3lock.acquire()4lock.acquire()#產(chǎn)生了死瑣。5lock.release()6lock.release()7importthreading8rLock=threading.RLock()#RLockM象9rLock.acquire()10rLock.acquire()#在同一線程內(nèi)，程序不會

10、堵塞。11rLock.release()12rLock.release()4、threading.Eventpython線程的事件用于主線程控制其他線程的執(zhí)行，事件主要提供了三個方法set、wait、clear。事件處理的機制：全局定義了一個“Flag”，如果“Flag”值為False，那么當程序執(zhí)行event.wait方法時就會阻基，如果方法時便不再阻塞?！癋lag”值為True，月8么event.waitclear:將“Flag”設(shè)置為Falseset:將“Flag”設(shè)置為TrueEvent.isSet():判斷標識位是否為Ture。1importthreading23defdo(even

11、t):4print('start')5event.wait()6print('execute')78event_obj=threading.Event()9foriinrange(10):10t=threading.Thread(target=do,args=(event_obj,)11t.start()1213event_obj.clear()14inp=input('input:')15ifinp='true':16event_obj.set()當線程執(zhí)行的時候，如果flag為False，則線程會阻塞，當flag為True的時

12、候，線程不會阻塞。它提供了本地和遠程的并發(fā)性。5、threading.Condition一個condition變量總是與某些類型的鎖相聯(lián)系，這個可以使用默認的情況或創(chuàng)建一個，當幾個condition變量必須共享和同一個鎖的時候，是很有用的。鎖是conditon對象的一部分：沒有必要分別跟蹤。condition變量服從上下文管理協(xié)議：with語句塊封閉之前可以獲取與鎖的聯(lián)系。acquire()和release()會調(diào)用與鎖相關(guān)聯(lián)的相應(yīng)的方法。其他和鎖關(guān)聯(lián)的方法必須被調(diào)用，wait()方法會釋放鎖，當另外一個線程使用notify()ornotify_all()喚醒它之前會一直阻塞。一旦被喚醒，wa

13、it()會重新獲得鎖并返回，Condition類實現(xiàn)了一個conditon變量。這個conditiaon變量允許一個或多個線程等待，直到他們被另一個線程通知。如果lock參數(shù)，被給定一個非空的值，那么他必須是一個lock或者Rlock對象，它用來做底層鎖。否則，會創(chuàng)建一個新的Rlock對象，用來做底層鎖。wait(timeout=None):等待通知，或者等到設(shè)定的超時時間。當調(diào)用這wait()方法時，如果調(diào)用它的線程沒有得到鎖，那么會拋出一個RuntimeError異常。wati()釋放鎖以后，在被調(diào)用相同條件的另一個進程用notify()ornotify_all()叫醒之前會一直阻塞。wa

14、it()還可以指定一個超時時間。如果有等待的線程，notify()方法會喚醒一個在等待conditon變量的線程。notify_all()則會喚醒所有在等待conditon變量的線程。注意：notify()和notify_all()不會釋放鎖，也就是說，線程被喚醒后不會立刻返回他們的wait()調(diào)用。除非線程調(diào)用notify()和notify_all()之后放棄了鎖的所有權(quán)。在典型的設(shè)計風格里，利用condition變量用鎖去通許訪問一些共享狀態(tài)，線程在獲取到它想得到的狀態(tài)前，會反復(fù)調(diào)用wait()修改狀態(tài)的線程在他們狀態(tài)改變時調(diào)用notify()ornotify_all(),用這種方式，線程

15、會盡可能的獲取到想要的一個等待者狀態(tài)。例子：生產(chǎn)者-消費者模型，12345importthreadingimporttimedefconsumer(cond):withcond:print("consumerbeforewait")78910111213141516171819202122232425print("consumerafterwait")defproducer(cond):withcond:print("producerbeforenotifyAll")cond.notifyAll()print("produc

16、erafternotifyAll")condition=threading.Condition()c1=threading.Thread(name="c1",target=consumer,args=(condition,)c2=threading.Thread(name="c2”,target=consumer,args=(condition,)p=threading.Thread(name="p",target=producer,args=(condition,)c1.start()time.sleep(2)c2.start()t

17、ime.sleep(2)p.start()6、queue模塊Queue就是對隊列，它是線程安全的舉例來說，我們?nèi)湲攧诔燥?。飯店里面有廚師職位，前臺負責把廚房做好的飯賣給顧客，顧客則去前臺領(lǐng)取做好的飯。這里的前臺就相當于我們的隊列。形成管道樣，廚師做好飯通過前臺傳送給顧客，所謂單向隊列這個模型也叫生產(chǎn)者-消費者模型。importqueueq=queue.Queue（maxsize=0）#構(gòu)造一個先進顯出隊列，maxsize指定隊列長q.join()q.qsize()q.empty()q.full()等到隊列為kong的時候，在執(zhí)行別的操作返回隊列的大?。ú豢煽浚┊旉犃袨榭盏臅r候，返回True當

18、隊列滿的時候，返回True,否則返回False（不可靠）否則返回False（不可靠）q.put(item,block=True,timeout=None)#將item放入Queued部，item必，10時，會等待隊列給出可用位置,11為False時為非阻塞，此時如12參數(shù)timeout,表示會阻塞設(shè)置的時間，過后,13如果隊列無法給出放入it14q.get(block=True,timeout=None)#移除并返回隊列頭部的一個值，可選參12345678910111213141515隊列為空，則阻塞，為False時，不阻塞,queue若此時隊列為空，則引發(fā)置的時候，過后，如果隊列為空，則引發(fā)

19、Empty異常。q.put_nowait(item)#等效于put(item,block=False)q.get_nowait()#等效于get(item,block=False)代碼如下:#!/usr/bin/envpythonimportQueueimportthreadingmessage=Queue.Queue(10)defproducer(i):whileTrue:message.put(i)defconsumer(i):whileTrue:msg=message.get()17foriinrange(12):18t=threading.Thread(target=producer,

20、args=(i,)19t.start()2021foriinrange(10):22t=threading.Thread(target=consumer,args=(i,)23t.start()那就自己做個線程池吧:#簡單往隊列中傳輸線程數(shù)importthreadingimporttimeimportqueueclassThreadingpool():def_init_(self,max_num=10):self.queuequeue.Queue(max_num)foriinrange(max_num):self.queue.put(threading.Thread)10defgetthrea

21、ding(self):11returnself.queue.get()12defaddthreading(self):13self.queue.put(threading.Thread)14deffunc(p,i):15time.sleep(1)16171819202122231234567891011print(i)p.addthreading()if_name_="_main_":p=Threadingpool()foriinrange(20):thread=p.getthreading()t=thread(target=func,args=(p,i)t.start()

22、#往隊列中無限添加任務(wù)importqueueimportthreadingimportcontextlibimporttimeStopEvent=object()classThreadPool(object):def_init_(self,max_num):self.q=queue.Queue()self.max_num=max_numself.terminal=False12self.generate_list131415161718192021222324252627282930313233self.free_list=defrun(self,func,args,callback=None

23、):"""線程池執(zhí)行一個任務(wù):paramfunc:任務(wù)函數(shù):paramargs:任務(wù)函數(shù)所需參數(shù):paramcallback:任務(wù)執(zhí)行失敗或成功后執(zhí)行的回調(diào)函數(shù)，回調(diào)函數(shù)返回值(默認為None,即：不執(zhí)行回調(diào)函數(shù)):return:如果線程池已經(jīng)終止，則返回True否則None»»»»»»iflen(self.free_list)=0andlen(self.generate_lisself.generate_thread()w=(func,args,callback,)self.q.put(w)defgen

24、erate_thread(self):"""創(chuàng)建一個線程"""t=threading.Thread(target=self.call)t.start()defcall(self):循環(huán)去獲取任務(wù)函數(shù)并執(zhí)行任務(wù)函數(shù)»»»»»»current_thread=threading.currentThreadself.generate_list.append(current_thread)event=self.q.get()#獲取線程whileevent!=StopEvent:#判斷獲

25、取的線程數(shù)不等于func,arguments,callback=event#拆務(wù)try:result=func(*arguments)#執(zhí)行函status=TrueexceptExceptionase:#函數(shù)執(zhí)行失敗status=Falseresult=eifcallbackisnotNonestry:callback(status,result)exceptExceptionase:passself.free_list.append(current_thread)event=self.q.get()35363738394041424344454647484950515253545556# s

26、elf.freelist.remove(currentthread)withself.work_state():575859606162636465666768697071727374757677event=self.q.get()else:self.generate_list.remove(current_thread)defclose(self):"""關(guān)閉線程，給傳輸全局非元祖的變量來進行關(guān)閉:return:"""foriinrange(len(self.generate_list):self.q.put(StopEvent)d

27、efterminate(self):"""突然關(guān)閉線程:return:»»»»»»self.terminal=Truewhileself.generate_list:self.q.put(StopEvent)self.q.empty()contextlib.contextmanager78defworkstate(self):self.free_list.append(threading.currentThread)79try:80yield81finally:82self.free_list.remov

28、e(threading.currentThread)83defwork(i):84print(i)85returni+1#返回給回調(diào)函數(shù)86defcallback(ret):87print(ret)88pool=ThreadPool(10)89foriteminrange(50):90pool.run(func=work,args=(item,),callback=callback)91pool.terminate()#pool.close()python進程multiprocessing是python的多進程管理包，和threading.Thread類似。1、multiprocessing模

29、塊直接從側(cè)面用subprocesses替換線程使用GIL的方式，由于這一占multiprocessing模塊可以讓程序員在給定的機器上充分的利用CPU在multiprocessing中，通過創(chuàng)建Process對象生成進程，然后調(diào)用它的start()方法,1frommultiprocessingimportProcess23deffunc(name):4print('hello',name)567if_name_="_main_":8p=Process(target=func,args=('zhangyanlin',)9p.start()10p

30、.join()#等待進程執(zhí)行完畢在使用并發(fā)設(shè)計的時候最好盡可能的避免共享數(shù)據(jù)，尤其是在使用多進程的時候。如果你真有需要要共享數(shù)據(jù),multiprocessing提供了兩種方式。(1)multiprocessing,Array,Value數(shù)據(jù)可以用Value或Array存儲在一個共享內(nèi)存地圖里，如下:1frommultiprocessingimportArray,Value,Process34567891011121314151617181920212223deffunc(a,b):a.value=3.333333333333333foriinrange(len(b):bi=-biif_name

31、_="_main_":num=Value('d',0.0)arr=Array('i',range(11)c=Process(target=func,args=(num,arr)d=Process(target=func,args=(num,arr)start()start()join()join()print(num.value)foriinarr:print(i)輸出:3.141592720,-1,-2,-3,-4,-5,-6,-7,-8,-9創(chuàng)建nunfnarr時，“d”和“i”參數(shù)由Array模塊使用的typecodes創(chuàng)建：“d”表示

32、一個雙精度的浮點數(shù)，“i”表示一個有符號的整數(shù)，這些共享對象將被線程安全的處理。'c':ctypes.c_char'b':ctypes.c_byte'h':ctypes.c_shortctypes.c_ushortctypes.c_uint'l':ctypes.c_long,'f':ctypes.c_floatArray('i',range(10)'u':ctypes.c_wchar'B':ctypes.c_ubyte'H':'i':c

33、types.c_int'I''L':ctypes.c_ulong'd':ctypes.c_double(2)multiprocessing,Manager由Manager。返回的manage醍供list,dict,Namespace,Lock,RLock,Semaphore,BoundedSemaphore,Condition,Event,Barrier,Queue,ValueandArray類型的支持。123456789101112131415161718frommultiprocessingimportProcess,Managerdeff(

34、d,l):d"name"="zhangyanlin"d"age"=18d"Job"="pythoner"l.reverse()if_name_="_main_":withManager()asman:d=man.dict()l=man.list(range(10)p=Process(target=f,args=(d,l)p.start()p.join()print(d)print(l)輸出:(0.25:None,1:'1','2':229,8,

35、7,6,5,4,3,2,1,0Serverprocessmanager比sharedmemory更靈活，因為它可以支持任意的對象類型。另外，一個單獨的manager可以通過進程在網(wǎng)絡(luò)上不同的計算機之間共享，不過他比sharedmemory要慢。2、進程池(Usingapoolofworkers)Pool類描述了一個工作進程池，他有幾種不同的方法讓任務(wù)卸載工作進程。進程池內(nèi)部維護一個進程序列，當使用時，則去進程池中獲取一個進程，如果進程池序列中沒有可供使用的進進程，那么程序就會等待，直到進程池中有可用進程為止。我們可以用Pool類創(chuàng)建一個進程池，展開提交的任務(wù)給進程池。例:#applyfromm

36、ultiprocessingimportPoolimporttimedeff1(i):67891011121314151617181920212223242526time.sleep(0.5)print(i)returni+100if_name_="_main_":pool=Pool(5)foriinrange(1,31):pool.apply(func=f1,args=(i,)#apply_asyncdeff1(i):time.sleep(0.5)print(i)returni+100deff2(arg):print(arg)if_name_="_main_&q

37、uot;:pool=Pool(5)foriinrange(1,31):27pool.apply_async(func=f1,args=(i,),callback=f2)pool.close()28pool.join()一個進程池對象可以控制工作進程池的哪些工作可以被提交，它支持超時和回調(diào)的異步結(jié)果，有一個類似map勺實現(xiàn)。processes:使用的工作進程的數(shù)量，如果processes是NoneBE么使用os.cpu_count()返回的數(shù)量。initializer:如果initializer是None,那么每一個工作進程在開始的時候會調(diào)用initializer(*initargs)。maxt

38、asksperchild:工作進程退出之前可以完成的任務(wù)數(shù)，完成后用一個心的工作進程來替代原進程，來讓閑置的資源被釋放。maxtasksperchild默認是None,意味著只要Pool存在工作進程就會一直存活。context:用在制定工作進程啟動時的上下文，一般使用multiprocessing.Pool()或者一個context對象的Pool()方法來創(chuàng)建一個池，兩種方法都適當?shù)脑O(shè)置了context注意：Pool對象的方法只可以被創(chuàng)建pool的進程所調(diào)用。Newinversion3.2:maxtasksperchildNewinversion3.4:context進程池的方法apply(f

39、unc,args,kwds):使用arg和kwds參數(shù)調(diào)用func函數(shù)，結(jié)果返回前會一直阻塞，由于這個原因，apply_async()更適合并發(fā)執(zhí)行，另外，func函數(shù)僅被pool中的一個進程運行。apply_async(func,args,kwds,callback,error_callback):apply()方法的一個變體，會返回一個結(jié)果對象。如果callback被指定，那么callback可以接收一個參數(shù)然后被調(diào)用，當結(jié)果準備好回調(diào)時會調(diào)用callback，調(diào)用失敗時，則用error_callback替換callback。Callbacks應(yīng)被立即完成，否則處理結(jié)果的線程會被阻塞。cl

40、ose():阻止更多的任務(wù)提交到pool,待任務(wù)完成后，工作進程會退出。terminate。:不管任務(wù)是否完成，立即停止工作進程。在對pool對象進程垃圾回收的時候，會立即調(diào)用terminate()。join():wait工作線程的退出，在調(diào)用join()前，必須調(diào)用close()orterminate()。這樣是因為被終止的進程需要被父進程調(diào)用wait(join等價與wait),否則進程會成為僵尸進程。map(func,iterable,chunksize)map_async(func,iterable,chunksize,callback,error_callback)?imap(func,iterable,chunksize)?imap_unordered(func,iterable,chunksize)starmap(func,iterable,chunksize)?starmap_async(func,iterable,chunksize,callback,error_back)python協(xié)程線程和進程的操作是由程序觸發(fā)系統(tǒng)接口，最后的執(zhí)行者是系統(tǒng)；協(xié)程的操作則是程序員。協(xié)程存在的意義：對于多線程應(yīng)用，CPU!過切片的方式來切換線程間的