進程

Python中的多線程無法利用多核優勢 , 所以如果我們想要充分地使用多核CPU的資源 , 那么就只能靠多進程了

multiprocessing模塊中提供了Process , Queue , Pipe , Lock , RLock , Event , Condition等組件 , 與threading模塊有很多相似之處

1.創建進程

from multiprocessing importProcessimporttimedeffunc(name):

time.sleep(2)print('hello',name)if __name__ == '__main__':

p= Process(target=func,args=('derek',))

p.start()#p.join()

print('end...')

View Code

2.進程間通訊

(1)Queue

不同進程間內存是不共享的，要想實現兩個進程間的數據交換。進程間通信有兩種主要形式 , 隊列和管道

from multiprocessing import Process, Queue #Queue是進程排列

deff(test):

test.put('22') #通過創建的子進程往隊列添加數據，實線父子進程交互

if __name__ == '__main__':

q= Queue() #父進程

q.put("11")

p= Process(target=f, args=(q,)) #子進程

p.start()

p.join()print("取到：",q.get_nowait())print("取到：",q.get_nowait())#父進程在創建子進程的時候就把q克隆一份給子進程#通過pickle序列化、反序列化，來達到兩個進程之間的交互

結果：

取到：11取到：22

Queue

(2)Pipe(管道)

The?Pipe()?function returns a pair of connection objects connected by a pipe which by default is duplex (two-way).

from multiprocessing importProcess, Pipedeff(conn):

conn.send('11')

conn.send('22')print("from parent:",conn.recv())print("from parent:", conn.recv())

conn.close()if __name__ == '__main__':

parent_conn, child_conn= Pipe() #生成管道實例，可以互相send()和recv()

p= Process(target=f, args=(child_conn,))

p.start()print(parent_conn.recv()) #prints "11"

print(parent_conn.recv()) #prints "22"

parent_conn.send("33") #parent 發消息給 child

parent_conn.send("44")

p.join()

Pipe

3.Manager

進程之間是相互獨立的 ,Queue和pipe只是實現了數據交互，并沒實現數據共享，Manager可以實現進程間數據共享 。

Manager還支持進程中的很多操作 , 比如Condition , Lock , Namespace , Queue , RLock , Semaphore等

from multiprocessing importProcess, Managerimportosdeff(d, l):

d[os.getpid()]=os.getpid()

l.append(os.getpid())print(l)if __name__ == '__main__':

with Manager() as manager:

d= manager.dict() #{} #生成一個字典，可在多個進程間共享和傳遞

l= manager.list(range(5)) #生成一個列表，可在多個進程間共享和傳遞

p_list =[]for i in range(2):

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

p.start()

p_list.append(p)for res in p_list: #等待結果

res.join()print(d)print(l)

View Code

4.lock

from multiprocessing importProcess, Lockdeff(l, i):#l.acquire()

print('hello world', i)#l.release()

if __name__ == '__main__':

lock=Lock()for num in range(100):

Process(target=f, args=(lock, num)).start() #要把lock傳到函數的參數l

#lock防止在屏幕上打印的時候會亂

lock

5.進程池

進程池內部維護一個進程序列，當使用時，則去進程池中獲取一個進程，如果進程池序列中沒有可供使用的進程，那么程序就會等待，直到進程池中有可用進程為止。

進程池中有以下幾個主要方法：

apply：從進程池里取一個進程并執行

apply_async：apply的異步版本

terminate:立刻關閉線程池

join：主進程等待所有子進程執行完畢，必須在close或terminate之后

close：等待所有進程結束后，才關閉線程池

from multiprocessing importProcess, PoolimporttimeimportosdefFoo(i):

time.sleep(2)print("in process",os.getpid())return i + 100

defBar(arg):print('-->exec done:', arg,os.getpid())if __name__ == '__main__': #多進程，必須加這一句(windows系統)

pool = Pool(processes=3) #允許進程池同時放入3個進程

print("主進程",os.getpid())for i in range(10):

pool.apply_async(func=Foo, args=(i,), callback=Bar) #callback=回調，執行完Foo(),接著執行Bar()

#pool.apply(func=Foo, args=(i,)) #串行

print('end')

pool.close()

pool.join()#進程池中進程執行完畢后再關閉，如果注釋，那么程序直接關閉。必須先close(),再join()

Pool

協程

1.簡介

協程(Coroutine) : 是單線程下的并發 , 又稱微線程 , 纖程 . 協程是一種用戶態的輕量級線程 , 即協程有用戶自己控制調度

協程擁有自己的寄存器上下文和棧。協程調度切換時，將寄存器上下文和棧保存到其他地方，在切回來的時候，恢復先前保存的寄存器上下文和棧。

協程能保留上一次調用時的狀態(即所有局部狀態的一個特定組合)，每次過程重入時，就相當于進入上一次調用的狀態

使用協程的優缺點

優點 :

協程的切換開銷更小 , 屬于程序級別的切換 , 更加輕量級

單線程內就可以實現并發的效果 , 最大限度利用CPU

缺點 :

協程的本質是單線程下 , 無法利用多核 , 可以是一個程序開啟多個進程 , 每個進程內開啟多個線程 , 每個線程內開啟協程

協程指的是單個線程 , 因而一旦協程出現阻塞 將會阻塞整個線程

2.Greenlet

greenlet是一個用C實現的協程模塊，相比與python自帶的yield，它可以使你在任意函數之間隨意切換，而不需把這個函數先聲明為generator

手動切換

from greenlet importgreenletdeftest1():print(12)

gr2.switch()#到這里切換到gr2，執行test2()

print(34)

gr2.switch()#切換到上次gr2運行的位置

deftest2():print(56)

gr1.switch()#切換到上次gr1運行的位置

print(78)

gr1= greenlet(test1) #啟動一個協程gr1

gr2 = greenlet(test2) #啟動一個協程gr2

gr1.switch()#開始運行gr1

greenlet

3.Gevent

Gevent 是一個第三方庫，可以輕松通過gevent實現并發同步或異步編程，在gevent中用到的主要模式是Greenlet, 它是以C擴展模塊形式接入Python的輕量級協程。

(1)IO阻塞自動切換

importgeventdeffoo():print('Running in foo')

gevent.sleep(2)print('阻塞時間最長，最后運行')defbar():print('running in bar')

gevent.sleep(1)print('foo()還在阻塞，這里第二個運行')deffunc3():print("running in func3")

gevent.sleep(0)print("其它兩個還在IO阻塞先運行")#創建協程實例

gevent.joinall([

gevent.spawn(foo),#生成，

gevent.spawn(bar),

gevent.spawn(func3),

])#遇到IO自動切換

結果：

Runninginfoo

runninginbar

runninginfunc3

其它兩個還在IO阻塞先運行

foo()還在阻塞，這里第二個運行

阻塞時間最長，最后運行

Process finished with exit code 0

View Code

由于切換是在IO操作時自動完成，所以gevent需要修改Python自帶的一些標準庫，這一過程在啟動時通過monkey patch完成：

(2)爬蟲例子：

from urllib importrequestimportgevent,timefrom gevent importmonkey

monkey.patch_all()#作用：把當前程序的所有的io操作給我單獨的做上標記

deff(url):print('GET: %s' %url)

resp=request.urlopen(url)

data=resp.read()print('%d bytes received from %s.' %(len(data), url))#同步需要的時間

urls = ['https://www.python.org/','https://www.yahoo.com/','https://github.com/']

time_start=time.time()for url inurls:

f(url)print("同步cost",time.time() -time_start)#下面是異步花費的時間

async_time_start =time.time()

gevent.joinall([

gevent.spawn(f,'https://www.python.org/'),

gevent.spawn(f,'https://www.yahoo.com/'),

gevent.spawn(f,'https://github.com/'),

])print("異步cost",time.time() -async_time_start)

結果：

GET: https://www.python.org/

48954 bytes received from https://www.python.org/.

GET: https://www.yahoo.com/

491871 bytes received from https://www.yahoo.com/.

GET: https://github.com/

51595 bytes received from https://github.com/.

同步cost4.928282260894775GET: https://www.python.org/GET: https://www.yahoo.com/GET: https://github.com/

48954 bytes received from https://www.python.org/.494958 bytes received from https://www.yahoo.com/.51599 bytes received from https://github.com/.

異步cost1.4920852184295654

IO多路復用

詳解：http://www.cnblogs.com/alex3714/articles/5876749.html

selectors模塊

selectors基于select模塊實現IO多路復用，調用語句selectors.DefaultSelector()，特點是根據平臺自動選擇最佳IO多路復用機制，調用順序：epoll > poll > select

做一個socket servers

importselectorsimportsocket

sel= selectors.DefaultSelector() #根據平臺自動選擇最佳IO多路復用機制

defaccept(sock, mask):

conn, addr= sock.accept() #Should be ready

#print('accepted', conn, 'from', addr,mask)

conn.setblocking(False) #設置為非阻塞IO

sel.register(conn, selectors.EVENT_READ, read)#新連接注冊read回調函數

#將conn和read函數注冊到一起，當conn有變化時執行read函數

defread(conn, mask):

data= conn.recv(1024) #Should be ready

ifdata:print('echoing', repr(data), 'to', conn)

conn.send(data)#Hope it won't block

else:print('closing', conn)

sel.unregister(conn)

conn.close()

sock=socket.socket()

sock.bind(('localhost', 9999))

sock.listen(100)

sock.setblocking(False)#設置為非阻塞IO

sel.register(sock, selectors.EVENT_READ, accept)#將sock和accept函數注冊到一起，當sock有變化時執行accept函數

whileTrue:

events= sel.select() #默認阻塞，有活動連接就返回活動的連接列表，監聽[(key1,mask1),(key2),(mask2)]

for key, mask inevents:

callback= key.data #accept #1 key.data就是accept # 2 key.data就是read

callback(key.fileobj, mask) #key.fileobj= 文件句柄

#1 key.fileobj就是sock # 2 key.fileobj就是conn

server

client

importsocketimportsys

messages= [ b'This is the message.',

b'It will be sent',

b'in parts.',

]

server_address= ('localhost', 9999)#Create a TCP/IP socket

socks = [ socket.socket(socket.AF_INET, socket.SOCK_STREAM) for i in range(5)]print(socks)#Connect the socket to the port where the server is listening

print('connecting to %s port %s' %server_address)for s insocks:

s.connect(server_address)for message inmessages:#Send messages on both sockets

for s insocks:print('%s: sending "%s"' %(s.getsockname(), message) )

s.send(message)#Read responses on both sockets

for s insocks:

data= s.recv(1024)print( '%s: received "%s"' %(s.getsockname(), data) )if notdata:print( 'closing socket', s.getsockname() )

mutlti conn socket client

總結

以上是生活随笔為你收集整理的python异步和进程_12.python进程\协程\异步IO的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。