關於python multiprocessing進程通訊的pipe和queue方式

這兩天溫故了python 的multiprocessing多進程模塊，看到的pipe和queue這兩種ipc方式，啥事ipc？ ipc就是進程間的通訊模式，經常使用的一半是socke，rpc，pipe和消息隊列等。 

今個就再把pipe和queue搞搞。

#coding:utf-8
import multiprocessing
import time

def proc1(pipe):
    while True:
        for i in xrange(10000):
            print "發送 %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()
#原文： http://rfyiamcool.blog.51cto.com/1030776/1549857

不僅是multiprocessing的pipe，包括其餘的pipe實現，都只是兩個進程之間的遊玩，我給你，你來接收 或者是你來，我接收。 固然也能夠作成雙工的狀態。  

queue的話，能夠有更多的進程參與進來。用法和一些別的queue差很少。

看下官網的文檔:

multiprocessing.Pipe([duplex])

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

#兩個pipe對象。用這兩個對象，來互相的交流。 

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.

#隊列的最大數

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.

#隊列的大小

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.

#是否孔了。 若是是空的，他回返回一個True 的狀態。 

full()

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

#隊列的狀態是否滿了。 

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).

#塞入隊列，能夠加超時的時間。 

#文章原文: http://rfyiamcool.blog.51cto.com/1030776/1549857 

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).

#獲取狀態

get_nowait()

Equivalent to get(False).

#不堵塞的get隊列裏面的數據

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.

#關閉，省當前進程的資源。 

 

我配置了multiprocessing隊里長度是3個，而後當我放入的是第四個的時候， 會發現一隻的堵塞，他是在等待，有人把數據get掉一個，那個時候 他才能繼續的塞入 。若是用put_nowait()的話，隊列超出會立馬會一個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)

下面是一段測試的代碼，同窗們能夠跑跑demo，感覺下。 

#coding:utf-8
import os
import multiprocessing
import time
# 寫入 worker
def inputQ(queue):
    while True:
        info = "進程號 %s : 時間: %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)

好了，簡單講講了 pipe和queue的用法。 其實我今個原本想扯扯python pipe的，結果google一搜，看到了multiprocessing的pipe。寫完了pipe後，感受文章的內容太少了，因此我才額外的增長了queue的。。。 

還有：  你們中秋節快樂 ~