Python進程、線程、協(xié)程詳解進程與線程的歷史我們都知道計算機是由硬件和軟件組成的。硬件中的CPU是計算機的核心，它承擔計算機的所有任務。操作系統(tǒng)是運行在硬件之上的軟件，是計算機的管理者，它負責資源的管理和分配、任務的調(diào)度。程序是運行在系統(tǒng)上的具有某種功能的軟件，比如說瀏覽器，音樂播放器等。每次執(zhí)行程序的時候，都會完成一定的功能，比如說瀏覽器幫我們打開網(wǎng)頁，為了保證其獨立性，就需要一個專門的管理和控制執(zhí)行程序的數(shù)據(jù)結(jié)構(gòu)——進程控制塊。進程就是一個程序在一個數(shù)據(jù)集上的一次動態(tài)執(zhí)行過程。進程一般由程序、數(shù)據(jù)集、進程控制塊三部分組成。我們編寫的程序用來描述進程要完成哪些功能以及如何完成；數(shù)據(jù)集則是程序在執(zhí)行過程中所需要使用的資源；進程控制塊用來記錄進程的外部特征，描述進程的執(zhí)行變化過程，系統(tǒng)可以利用它來控制和管理進程，它是系統(tǒng)感知進程存在的唯一標志。在早期的操作系統(tǒng)里，計算機只有一個核心，進程執(zhí)行程序的最小單位，任務調(diào)度采用時間片輪轉(zhuǎn)的搶占式方式進行進程調(diào)度。每個進程都有各自的一塊獨立的內(nèi)存，保證進程彼此間的內(nèi)存地址空間的隔離。隨著計算機技術(shù)的發(fā)展，進程出現(xiàn)了很多弊端，一是進程的創(chuàng)建、撤銷和切換的開銷比較大，二是由于對稱多處理機（對稱多處理機（SymmetricalMulti-Processing）又叫SMP，是指在一個計算機上匯集了一組處理器(多CPU)，各CPU之間共享內(nèi)存子系統(tǒng)以及總線結(jié)構(gòu)）的出現(xiàn)，可以滿足多個運行單位，而多進程并行開銷過大。這個時候就引入了線程的概念。線程也叫輕量級進程，它是一個基本的CPU執(zhí)行單元，也是程序執(zhí)行過程中的最小單元，由線程ID、程序計數(shù)器、寄存器集合和堆棧共同組成。線程的引入減小了程序并發(fā)執(zhí)行時的開銷，提高了操作系統(tǒng)的并發(fā)性能。線程沒有自己的系統(tǒng)資源，只擁有在運行時必不可少的資源。但線程可以與同屬與同一進程的其他線程共享進程所擁有的其他資源。進程與線程之間的關系線程是屬于進程的，線程運行在進程空間內(nèi)，同一進程所產(chǎn)生的線程共享同一內(nèi)存空間，當進程退出時該進程所產(chǎn)生的線程都會被強制退出并清除。線程可與屬于同一進程的其它線程共享進程所擁有的全部資源，但是其本身基本上不擁有系統(tǒng)資源，只擁有一點在運行中必不可少的信息(如程序計數(shù)器、一組寄存器和棧)。python線程Threading用于提供線程相關的操作，線程是應用程序中工作的最小單元。1、threading模塊threading模塊建立在_thread模塊之上。thread模塊以低級、原始的方式來處理和控制線程，而threading模塊通過對thread進行二次封裝，提供了更方便的api來處理線程。123456789101112131415import

threadingimport

time

def

worker(num):

"""

thread

worker

function

:return:

"""

time.sleep(1)

print("The

num

is

%d"

%

num)

return

for

i

in

range(20):

t

=

threading.Thread(target=worker,args=(i,)，name=“t.%d”

%

i)

t.start()上述代碼創(chuàng)建了20個“前臺”線程，然后控制器就交給了CPU，CPU根據(jù)指定算法進行調(diào)度，分片執(zhí)行指令。Thread方法說明t.start():激活線程，t.getName():獲取線程的名稱t.setName()：設置線程的名稱

:獲取或設置線程的名稱t.is_alive()：判斷線程是否為激活狀態(tài)t.isAlive()：判斷線程是否為激活狀態(tài)t.setDaemon()設置為后臺線程或前臺線程（默認：False）;通過一個布爾值設置線程是否為守護線程，必須在執(zhí)行start()方法之后才可以使用。如果是后臺線程，主線程執(zhí)行過程中，后臺線程也在進行，主線程執(zhí)行完畢后，后臺線程不論成功與否，均停止；如果是前臺線程，主線程執(zhí)行過程中，前臺線程也在進行，主線程執(zhí)行完畢后，等待前臺線程也執(zhí)行完成后，程序停止t.isDaemon()：判斷是否為守護線程t.ident：獲取線程的標識符。線程標識符是一個非零整數(shù)，只有在調(diào)用了start()方法之后該屬性才有效，否則它只返回None。t.join()：逐個執(zhí)行每個線程，執(zhí)行完畢后繼續(xù)往下執(zhí)行，該方法使得多線程變得無意義t.run()：線程被cpu調(diào)度后自動執(zhí)行線程對象的run方法2、線程鎖threading.RLock和threading.Lock由于線程之間是進行隨機調(diào)度，并且每個線程可能只執(zhí)行n條執(zhí)行之后，CPU接著執(zhí)行其他線程。為了保證數(shù)據(jù)的準確性，引入了鎖的概念。所以，可能出現(xiàn)如下問題：例：假設列表A的所有元素就為0，當一個線程從前向后打印列表的所有元素，另外一個線程則從后向前修改列表的元素為1,那么輸出的時候，列表的元素就會一部分為0，一部分為1,這就導致了數(shù)據(jù)的不一致。鎖的出現(xiàn)解決了這個問題。123456789101112131415161718import

threadingimport

time

globals_num

=

0

lock

=

threading.RLock()

def

Func():

lock.acquire()

#

獲得鎖

global

globals_num

globals_num

+=

1

time.sleep(1)

print(globals_num)

lock.release()

#

釋放鎖

for

i

in

range(10):

t

=

threading.Thread(target=Func)

t.start()3、threading.RLock和threading.Lock的區(qū)別RLock允許在同一線程中被多次acquire。而Lock卻不允許這種情況。如果使用RLock，那么acquire和release必須成對出現(xiàn)，即調(diào)用了n次acquire，必須調(diào)用n次的release才能真正釋放所占用的瑣。123456789101112import

threadinglock

=

threading.Lock()

#Lock對象lock.acquire()lock.acquire()

#產(chǎn)生了死瑣。lock.release()lock.release()import

threadingrLock

=

threading.RLock()

#RLock對象rLock.acquire()rLock.acquire()

#在同一線程內(nèi)，程序不會堵塞。rLock.release()rLock.release()4、threading.Eventpython線程的事件用于主線程控制其他線程的執(zhí)行，事件主要提供了三個方法set、wait、clear。事件處理的機制：全局定義了一個“Flag”，如果“Flag”值為False，那么當程序執(zhí)行event.wait方法時就會阻塞，如果“Flag”值為True，那么event.wait方法時便不再阻塞。clear：將“Flag”設置為Falseset：將“Flag”設置為TrueEvent.isSet()：判斷標識位是否為Ture。12345678910111213141516import

threading

def

do(event):

print('start')

event.wait()

print('execute')

event_obj

=

threading.Event()for

i

in

range(10):

t

=

threading.Thread(target=do,

args=(event_obj,))

t.start()

event_obj.clear()inp

=

input('input:')if

inp

==

'true':

event_obj.set()當線程執(zhí)行的時候，如果flag為False，則線程會阻塞，當flag為True的時候，線程不會阻塞。它提供了本地和遠程的并發(fā)性。5、threading.Condition一個condition變量總是與某些類型的鎖相聯(lián)系，這個可以使用默認的情況或創(chuàng)建一個，當幾個condition變量必須共享和同一個鎖的時候，是很有用的。鎖是conditon對象的一部分：沒有必要分別跟蹤。condition變量服從上下文管理協(xié)議：with語句塊封閉之前可以獲取與鎖的聯(lián)系。acquire()和release()會調(diào)用與鎖相關聯(lián)的相應的方法。其他和鎖關聯(lián)的方法必須被調(diào)用，wait()方法會釋放鎖，當另外一個線程使用notify()ornotify_all()喚醒它之前會一直阻塞。一旦被喚醒，wait()會重新獲得鎖并返回，Condition類實現(xiàn)了一個conditon變量。這個conditiaon變量允許一個或多個線程等待，直到他們被另一個線程通知。如果lock參數(shù)，被給定一個非空的值，，那么他必須是一個lock或者Rlock對象，它用來做底層鎖。否則，會創(chuàng)建一個新的Rlock對象，用來做底層鎖。wait(timeout=None)：等待通知，或者等到設定的超時時間。當調(diào)用這wait()方法時，如果調(diào)用它的線程沒有得到鎖，那么會拋出一個RuntimeError異常。wati()釋放鎖以后，在被調(diào)用相同條件的另一個進程用notify()ornotify_all()叫醒之前會一直阻塞。wait()還可以指定一個超時時間。如果有等待的線程，notify()方法會喚醒一個在等待conditon變量的線程。notify_all()則會喚醒所有在等待conditon變量的線程。注意：notify()和notify_all()不會釋放鎖，也就是說，線程被喚醒后不會立刻返回他們的wait()調(diào)用。除非線程調(diào)用notify()和notify_all()之后放棄了鎖的所有權(quán)。在典型的設計風格里，利用condition變量用鎖去通許訪問一些共享狀態(tài)，線程在獲取到它想得到的狀態(tài)前，會反復調(diào)用wait()。修改狀態(tài)的線程在他們狀態(tài)改變時調(diào)用notify()ornotify_all()，用這種方式，線程會盡可能的獲取到想要的一個等待者狀態(tài)。例子：生產(chǎn)者-消費者模型，12345678910111213141516171819202122232425import

threadingimport

timedef

consumer(cond):

with

cond:

print("consumer

before

wait")

cond.wait()

print("consumer

after

wait")

def

producer(cond):

with

cond:

print("producer

before

notifyAll")

cond.notifyAll()

print("producer

after

notifyAll")

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()time.sleep(2)p.start()6、queue模塊Queue就是對隊列，它是線程安全的舉例來說，我們?nèi)湲攧诔燥垺ｏ埖昀锩嬗袕N師職位，前臺負責把廚房做好的飯賣給顧客，顧客則去前臺領取做好的飯。這里的前臺就相當于我們的隊列。形成管道樣，廚師做好飯通過前臺傳送給顧客，所謂單向隊列這個模型也叫生產(chǎn)者-消費者模型。123456789101112131415import

queue

q

=

queue.Queue(maxsize=0)

#

構(gòu)造一個先進顯出隊列，maxsize指定隊列長度，為0

時，表示隊列長度無限制。

q.join()

#

等到隊列為kong的時候，在執(zhí)行別的操作q.qsize()

#

返回隊列的大小

（不可靠）q.empty()

#

當隊列為空的時候，返回True

否則返回False

（不可靠）q.full()

#

當隊列滿的時候，返回True，否則返回False

（不可靠）q.put(item,

block=True,

timeout=None)

#

將item放入Queue尾部，item必須存在，可以參數(shù)block默認為True,表示當隊列滿時，會等待隊列給出可用位置，

為False時為非阻塞，此時如果隊列已滿，會引發(fā)queue.Full

異常。

可選參數(shù)timeout，表示

會阻塞設置的時間，過后，

如果隊列無法給出放入item的位置，則引發(fā)

queue.Full

異常q.get(block=True,

timeout=None)

#

移除并返回隊列頭部的一個值，可選參數(shù)block默認為True，表示獲取值的時候，如果隊列為空，則阻塞，為False時，不阻塞，若此時隊列為空，則引發(fā)

queue.Empty異常。

可選參數(shù)timeout，表示會阻塞設置的時候，過后，如果隊列為空，則引發(fā)Empty異常。q.put_nowait(item)

#

等效于

put(item,block=False)q.get_nowait()

#

等效于

get(item,block=False)代碼如下：1234567891011121314151617181920212223#!/usr/bin/env

pythonimport

Queueimport

threadingmessage

=

Queue.Queue(10)

def

producer(i):

while

True:

message.put(i)

def

consumer(i):

while

True:

msg

=

message.get()

for

i

in

range(12):

t

=

threading.Thread(target=producer,

args=(i,))

t.start()

for

i

in

range(10):

t

=

threading.Thread(target=consumer,

args=(i,))

t.start()那就自己做個線程池吧：1234567891011121314151617181920212223#

簡單往隊列中傳輸線程數(shù)import

threadingimport

timeimport

queueclass

Threadingpool():

def

__init__(self,max_num

=

10):

self.queue

=

queue.Queue(max_num)

for

i

in

range(max_num):

self.queue.put(threading.Thread)

def

getthreading(self):

return

self.queue.get()

def

addthreading(self):

self.queue.put(threading.Thread)def

func(p,i):

time.sleep(1)

print(i)

p.addthreading()if

__name__

==

"__main__":

p

=

Threadingpool()

for

i

in

range(20):

thread

=

p.getthreading()

t

=

thread(target

=

func,

args

=

(p,i))

t.start()12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091#往隊列中無限添加任務import

queueimport

threadingimport

contextlibimport

timeStopEvent

=

object()class

ThreadPool(object):

def

__init__(self,

max_num):

self.q

=

queue.Queue()

self.max_num

=

max_num

self.terminal

=

False

self.generate_list

=

[]

self.free_list

=

[]

def

run(self,

func,

args,

callback=None):

"""

線程池執(zhí)行一個任務

:param

func:

任務函數(shù)

:param

args:

任務函數(shù)所需參數(shù)

:param

callback:

任務執(zhí)行失敗或成功后執(zhí)行的回調(diào)函數(shù)，回調(diào)函數(shù)有兩個參數(shù)1、任務函數(shù)執(zhí)行狀態(tài)；2、任務函數(shù)返回值（默認為None，即：不執(zhí)行回調(diào)函數(shù)）

:return:

如果線程池已經(jīng)終止，則返回True否則None

"""

if

len(self.free_list)

==

0

and

len(self.generate_list)

<

self.max_num:

self.generate_thread()

w

=

(func,

args,

callback,)

self.q.put(w)

def

generate_thread(self):

"""

創(chuàng)建一個線程

"""

t

=

threading.Thread(target=self.call)

t.start()

def

call(self):

"""

循環(huán)去獲取任務函數(shù)并執(zhí)行任務函數(shù)

"""

current_thread

=

threading.currentThread

self.generate_list.append(current_thread)

event

=

self.q.get()

#

獲取線程

while

event

!=

StopEvent:

#

判斷獲取的線程數(shù)不等于全局變量

func,

arguments,

callback

=

event

#

拆分元祖，獲得執(zhí)行函數(shù)，參數(shù)，回調(diào)函數(shù)

try:

result

=

func(*arguments)

#

執(zhí)行函數(shù)

status

=

True

except

Exception

as

e:

#

函數(shù)執(zhí)行失敗

status

=

False

result

=

e

if

callback

is

not

None:

try:

callback(status,

result)

except

Exception

as

e:

pass

#

self.free_list.append(current_thread)

#

event

=

self.q.get()

#

self.free_list.remove(current_thread)

with

self.work_state():

event

=

self.q.get()

else:

self.generate_list.remove(current_thread)

def

close(self):

"""

關閉線程，給傳輸全局非元祖的變量來進行關閉

:return:

"""

for

i

in

range(len(self.generate_list)):

self.q.put(StopEvent)

def

terminate(self):

"""

突然關閉線程

:return:

"""

self.terminal

=

True

while

self.generate_list:

self.q.put(StopEvent)

self.q.empty()

@contextlib.contextmanager

def

work_state(self):

self.free_list.append(threading.currentThread)

try:

yield

finally:

self.free_list.remove(threading.currentThread)def

work(i):

print(i)

return

i

+1

#

返回給回調(diào)函數(shù)def

callback(ret):

print(ret)pool

=

ThreadPool(10)for

item

in

range(50):

pool.run(func=work,

args=(item,),callback=callback)pool.terminate()#

pool.close()python進程multiprocessing是python的多進程管理包，和threading.Thread類似。1、multiprocessing模塊直接從側(cè)面用subprocesses替換線程使用GIL的方式，由于這一點，multiprocessing模塊可以讓程序員在給定的機器上充分的利用CPU。在multiprocessing中，通過創(chuàng)建Process對象生成進程，然后調(diào)用它的start()方法，12345678910from

multiprocessing

import

Process

def

func(name):

print('hello',

name)

if

__name__

==

"__main__":

p

=

Process(target=func,args=('zhangyanlin',))

p.start()

p.join()

#

等待進程執(zhí)行完畢在使用并發(fā)設計的時候最好盡可能的避免共享數(shù)據(jù)，尤其是在使用多進程的時候。如果你真有需要要共享數(shù)據(jù)，multiprocessing提供了兩種方式。（1）multiprocessing，Array,Value數(shù)據(jù)可以用Value或Array存儲在一個共享內(nèi)存地圖里，如下：1234567891011121314151617181920212223from

multiprocessing

import

Array,Value,Process

def

func(a,b):

a.value

=

3.333333333333333

for

i

in

range(len(b)):

b[i]

=

-b[i]

if

__name__

==

"__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))

c.start()

d.start()

c.join()

d.join()

print(num.value)

for

i

in

arr:

print(i)輸出：123.1415927[0,

-1,

-2,

-3,

-4,

-5,

-6,

-7,

-8,

-9]創(chuàng)建num和arr時，“d”和“i”參數(shù)由Array模塊使用的typecodes創(chuàng)建：“d”表示一個雙精度的浮點數(shù)，“i”表示一個有符號的整數(shù)，這些共享對象將被線程安全的處理。Array(‘i’,range(10))中的‘i’參數(shù)：‘c’:ctypes.c_char‘u’:ctypes.c_wchar‘b’:ctypes.c_byte‘B’:ctypes.c_ubyte‘h’:ctypes.c_short

‘H’:ctypes.c_ushort

‘i’:ctypes.c_int‘I’:ctypes.c_uint‘l’:ctypes.c_long,‘L’:ctypes.c_ulong‘f’:ctypes.c_float‘d’:ctypes.c_double（2）multiprocessing，Manager由Manager()返回的manager提供list,dict,Namespace,Lock,RLock,Semaphore,BoundedSemaphore,Condition,Event,Barrier,Queue,ValueandArray類型的支持。123456789101112131415161718from

multiprocessing

import

Process,Managerdef

f(d,l):

d["name"]

=

"zhangyanlin"

d["age"]

=

18

d["Job"]

=

"pythoner"

l.reverse()

if

__name__

==

"__main__":

with

Manager()

as

man:

d

=

man.dict()

l

=

man.list(range(10))

p

=

Process(target=f,args=(d,l))

p.start()

p.join()

print(d)

print(l)輸出：12{0.25:

None,

1:

'1',

'2':

2}[9,

8,

7,

6,

5,

4,

3,

2,

1,

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

multiprocessing

import

Poolimport

time

def

f1(i):

time.sleep(0.5)

print(i)

return

i

+

100

if

__name__

==

"__main__":

pool

=

Pool(5)

for

i

in

range(1,31):

pool.apply(func=f1,args=(i,))

#apply_asyncdef

f1(i):

time.sleep(0.5)

print(i)

return

i

+

100def

f2(arg):

print(arg)

if

__name__

==

"__main__":

pool

=

Pool(5)

for

i

in

range(1,31):

pool.apply_async(func=f1,args=(i,),callback=f2)

pool.close()

pool.join()一個進程池對象可以控制工作進程池的哪些工作可以被提交，它支持超時和回調(diào)的異步結(jié)果，有一個類似map的實現(xiàn)。processes：使用的工作進程的數(shù)量，如果processes是None那么使用os.cpu_count()返回的數(shù)量。initializer：如果initializer是None，那么每一個工作進程在開始的時候會調(diào)用initializer(*initargs)。maxtasksperchild：工作進程退出之前可以完成的任務數(shù)，完成后用一個心的工作進程來替代原進程，來讓閑置的資源被釋放。maxtasksperchild默認是None，意味著只要Pool存在工作進程就會一直存活。context:用在制定工作進程啟動時的上下文，一般使用multiprocessing.Pool()或者一個context對象的Pool()方法來創(chuàng)建一個池，兩種方法都適當?shù)脑O置了context注意：Pool對象的方法只可以被創(chuàng)建pool的進程所調(diào)用。Newinversion3.2:maxtasksperchildNewinversion3.4:context進程池的方法apply(func[,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應被立即完成，否則處理結(jié)果的線程會被阻塞。close()：阻止更多的任務提交到pool，待任務完成后，工作進程會退出。terminate()：不管任務是否完成，立即停止工作進程。在對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_bac