Introduction to pipe and queue for python's multiprocessing multi-process communication_PHP tutorial

Python's multiprocessing provides IPC (Pipe and Queue), making Python multi-process concurrent and more efficient. In this article, we will introduce pipe and queue in detail.

In the past two days, I have been reviewing the multiprocessing module of Python. I have seen the two IPC methods of pipe and queue. What is IPC about? ipc is the inter-process communication mode. Half of the commonly used ones are socket, rpc, pipe and message queue.


Today I’m going to mess with the pipe and queue again.

The code is as follows

#coding:utf-8 代码如下   #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() import multiprocessing import time def proc1(pipe): while True: for i in xrange(10000): print "Send %s"%i pipe.send(i) time.sleep(1) def proc2(pipe): while True: print 'proc2 receives:',pipe.recv() time.sleep(1) def proc3(pipe): while True: print 'proc3 receives:',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()

Not only multiprocessing pipes, but also other pipe implementations are just games between two processes. I give it to you and you receive it, or you come and I receive it. Of course, it can also be made into a duplex state.

In the case of queue, more processes can participate. The usage is similar to some other queues.


Take a look at the documentation on the official website:

multiprocessing.Pipe([duplex])

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

#Two pipe objects. Use these two objects to communicate with each other.


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.

#Maximum number of queues


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.

#Queue size


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.

#Whether it is holed. If it is empty, it returns a True status.


full()

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

#Whether the queue status is full.


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

#Put it into the queue, you can add a timeout.

put_nowait(obj)

Equivalent to put(obj, False).

#There is no traffic here


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 status


get_nowait()

Equivalent to get(False).

#Get the data in the queue without blocking


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.

#Close, save resources of the current process.



I configured the length of the multiprocessing queue to be 3. Then when I put the fourth one, I will find that one is blocked. It is waiting for someone to get the data and drop one. At that time, it can continue to block. enter. If you use put_nowait(), an error will occur immediately if the queue exceeds the limit.


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




Below is a piece of test code. Students can run the demo and get a feel for it.

The code is as follows

#coding:utf-8 代码如下   #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) import os import multiprocessing import time # Write worker def inputQ(queue): while True: info = "Process ID %s: Time: %s"%(os.getpid(),int(time.time())) queue.put(info) time.sleep(1) # Get 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)

Okay, let’s briefly talk about the usage of pipe and queue. In fact, I originally wanted to talk about python pipes, but after searching on Google, I saw multiprocessing pipes. After writing the pipe, I felt that the content of the article was too little, so I added an additional queue. . . http://www.bkjia.com/PHPjc/968077.htmlwww.bkjia.comtruehttp: //www.bkjia.com/PHPjc/968077.htmlTechArticleIntroduction to pipe and queue of python's multiprocessing multi-process communication python's multiprocessing provides IPC (Pipe and Queue), so that Python has multi-process concurrency and is more efficient. Let’s go into details in this article...