python的multiprocessing多進(jìn)程通信的pipe和queue介紹
來源:易賢網(wǎng) 閱讀:1787 次 日期:2015-03-17 10:11:13
溫馨提示:易賢網(wǎng)小編為您整理了“python的multiprocessing多進(jìn)程通信的pipe和queue介紹”,方便廣大網(wǎng)友查閱!

python的multiprocessing提供了IPC(Pipe和Queue),使Python多進(jìn)程并發(fā),效率上更高。本文我們就來詳細(xì)介紹一下pipe和queue。

這兩天溫故了python的multiprocessing多進(jìn)程模塊,看到的pipe和queue這兩種ipc方式,啥事ipc? ipc就是進(jìn)程間的通信模式,常用的一半是socke,rpc,pipe和消息隊(duì)列等。

今個(gè)就再把pipe和queue搞搞。

代碼如下:

#coding:utf-8

import multiprocessing

import time

def proc1(pipe):

while True:

for i in xrange(10000):

print "發(fā)送 %s"%i

pipe.send(i)

time.sleep(1)

def proc2(pipe):

while True:

print 'proc2 接收:',pipe.recv()

time.sleep(1)

def proc3(pipe):

while True:

print 'proc3 接收:',pipe.recv()

time.sleep(1)

# Build a pipe

pipe = multiprocessing.Pipe()

print pipe

# Pass an end of the pipe to process 1

p1 = multiprocessing.Process(target=proc1, args=(pipe[0],))

# Pass the other end of the pipe to process 2

p2 = multiprocessing.Process(target=proc2, args=(pipe[1],))

p1.start()

p2.start()

p1.join()

p2.join()

名單1

python的multiprocessing多進(jìn)程通信的pipe和queue介紹

不只是multiprocessing的pipe,包括其他的pipe實(shí)現(xiàn),都只是兩個(gè)進(jìn)程之間的游玩,我給你,你來接收 或者是你來,我接收。 當(dāng)然也可以做成雙工的狀態(tài)。

queue的話,可以有更多的進(jìn)程參與進(jìn)來。用法和一些別的queue差不多。

看下官網(wǎng)的文檔:

multiprocessing.Pipe([duplex])

Returns a pair (conn1, conn2) of Connection objects representing the ends of a pipe.

#兩個(gè)pipe對(duì)象。用這兩個(gè)對(duì)象,來互相的交流。

If duplex is True (the default) then the pipe is bidirectional. If duplex is False then the pipe is unidirectional: conn1 can only be used for receiving messages and conn2 can only be used for sending messages.

class multiprocessing.Queue([maxsize])

Returns a process shared queue implemented using a pipe and a few locks/semaphores. When a process first puts an item on the queue a feeder thread is started which transfers objects from a buffer into the pipe.

#隊(duì)列的最大數(shù)

The usual Queue.Empty and Queue.Full exceptions from the standard library’s Queue module are raised to signal timeouts.

Queue implements all the methods of Queue.Queue except for task_done() and join().

qsize()

Return the approximate size of the queue. Because of multithreading/multiprocessing semantics, this number is not reliable.

#隊(duì)列的大小

Note that this may raise NotImplementedError on Unix platforms like Mac OS X where sem_getvalue() is not implemented.

empty()

Return True if the queue is empty, False otherwise. Because of multithreading/multiprocessing semantics, this is not reliable.

#是否孔了。 如果是空的,他回返回一個(gè)True 的狀態(tài)。

full()

Return True if the queue is full, False otherwise. Because of multithreading/multiprocessing semantics, this is not reliable.

#隊(duì)列的狀態(tài)是否滿了。

put(obj[, block[, timeout]])

Put obj into the queue. If the optional argument block is True (the default) and timeout is None (the default), block if necessary until a free slot is available. If timeout is a positive number, it blocks at most timeout seconds and raises the Queue.Full exception if no free slot was available within that time. Otherwise (block is False), put an item on the queue if a free slot is immediately available, else raise the Queue.Full exception (timeout is ignored in that case).

#塞入隊(duì)列,可以加超時(shí)的時(shí)間。

put_nowait(obj)

Equivalent to put(obj, False).

#這里是不堵塞的

get([block[, timeout]])

Remove and return an item from the queue. If optional args block is True (the default) and timeout is None (the default), block if necessary until an item is available. If timeout is a positive number, it blocks at most timeout seconds and raises the Queue.Empty exception if no item was available within that time. Otherwise (block is False), return an item if one is immediately available, else raise the Queue.Empty exception (timeout is ignored in that case).

#獲取狀態(tài)

get_nowait()

Equivalent to get(False).

#不堵塞的get隊(duì)列里面的數(shù)據(jù)

Queue has a few additional methods not found in Queue.Queue. These methods are usually unnecessary for most code:

close()

Indicate that no more data will be put on this queue by the current process. The background thread will quit once it has flushed all buffered data to the pipe. This is called automatically when the queue is garbage collected.

#關(guān)閉,省當(dāng)前進(jìn)程的資源。

我配置了multiprocessing隊(duì)里長度是3個(gè),然后當(dāng)我放入的是第四個(gè)的時(shí)候, 會(huì)發(fā)現(xiàn)一只的堵塞,他是在等待,有人把數(shù)據(jù)get掉一個(gè),那個(gè)時(shí)候 他才能繼續(xù)的塞入 。如果用put_nowait()的話,隊(duì)列超出會(huì)立馬會(huì)一個(gè)error的。

/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/multiprocessing/queues.pyc in put_nowait(self, obj)

/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/multiprocessing/queues.pyc in put(self, obj, block, timeout)

wKioL1QMjmuR30vjAAOEnmz0ElE220.jpg

名單

下面是一段測(cè)試的代碼,同學(xué)們可以跑跑demo,感受下。

代碼如下:

#coding:utf-8

import os

import multiprocessing

import time

# 寫入 worker

def inputQ(queue):

while True:

info = "進(jìn)程號(hào) %s : 時(shí)間: %s"%(os.getpid(),int(time.time()))

queue.put(info)

time.sleep(1)

# 獲取 worker

def outputQ(queue,lock):

while True:

info = queue.get()

# lock.acquire()

print (str(os.getpid()) + '(get):' + info)

# lock.release()

time.sleep(1)

#===================

# Main

record1 = [] # store input processes

record2 = [] # store output processes

lock = multiprocessing.Lock() # To prevent messy print

queue = multiprocessing.Queue(3)

# input processes

for i in range(10):

process = multiprocessing.Process(target=inputQ,args=(queue,))

process.start()

record1.append(process)

# output processes

for i in range(10):

process = multiprocessing.Process(target=outputQ,args=(queue,lock))

process.start()

record2.append(process)

wKiom1QMigmjsp8LAANICa7A4dI435.jpg

名單2

好了,簡單講講了 pipe和queue的用法。 其實(shí)我今個(gè)本來想扯扯python pipe的,結(jié)果google一搜,看到了multiprocessing的pipe。寫完了pipe后,感覺文章的內(nèi)容太少了,所以我才額外的增加了queue的。。。

更多信息請(qǐng)查看IT技術(shù)專欄

更多信息請(qǐng)查看網(wǎng)絡(luò)編程

扫描二维码关注易贤网公众号

易賢網(wǎng)手機(jī)網(wǎng)站地址:python的multiprocessing多進(jìn)程通信的pipe和queue介紹
由于各方面情況的不斷調(diào)整與變化,易賢網(wǎng)提供的所有考試信息和咨詢回復(fù)僅供參考,敬請(qǐng)考生以權(quán)威部門公布的正式信息和咨詢?yōu)闇?zhǔn)!

2025國考·省考課程試聽報(bào)名

  • 報(bào)班類型
  • 姓名
  • 手機(jī)號(hào)
  • 驗(yàn)證碼

【备考推荐】2025年事业单位联考系统精讲课,免费试看,配套资料包邮到家

關(guān)于我們 | 聯(lián)系我們 | 人才招聘 | 網(wǎng)站聲明 | 網(wǎng)站幫助 | 非正式的簡要咨詢 | 簡要咨詢須知 | 新媒體/短視頻平臺(tái) | 手機(jī)站點(diǎn) | 投訴建議
工業(yè)和信息化部備案號(hào):滇ICP備2023014141號(hào)-1 云南省教育廳備案號(hào):云教ICP備0901021 滇公網(wǎng)安備53010202001879號(hào) 人力資源服務(wù)許可證:(云)人服證字(2023)第0102001523號(hào)
云南網(wǎng)警備案專用圖標(biāo)
聯(lián)系電話:0871-65099533/13759567129 獲取招聘考試信息及咨詢關(guān)注公眾號(hào):hfpxwx
咨詢QQ:1093837350(9:00—18:00)版權(quán)所有:易賢網(wǎng)
云南網(wǎng)警報(bào)警專用圖標(biāo)