Multiprocessing.queues.queue.put

Method Name:

put

Method Signature:

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

Return Value:

NoneType

Parameters:

object – Any Python object can be added to a Queue instance for the consumption of a reader/consumer process. Remember the underlying implementation of this Queue is Unix pipe. Once the message is read using the get() method by a consumer process, it is not available forever in the Queue.

block - Boolean value indicating whether put() should block and wait till a slot is available in the Queue. The default value is True denoting put() should block till a slot is available.

timeout – Number of seconds to wait for before a slot is available for writing data.

Method Overview:

put() adds any Python object into the Queue.
Using the parameters block and timeout the put() operation can wait either indefinitely or for finite number of seconds before a slot is available in the Queue for writing.

Example:

The example creates two producers and one consumer process.
The behavior of the program can be studied by varying the number of producers, consumers, queue size, timeout for put() method, timeout for get() method, sleep time of producers and consumers.

#import the required Python modules

import multiprocessing

import queue

import time

PRODUCER_PROCESS_LIMIT = 4 # Define limit for producer process

PRODUCER_TIMEOUT = 3 # Define timeout for producer process

PRODUCER_SLEEP_QUANTUM = 2 # Define sleep time producer process

CONSUMER_TIMEOUT = 10 # Define timeout for consumer process

# Function for Producer process

def producer0(messageQueue):

try:

for i in range(1, PRODUCER_PROCESS_LIMIT+1):

time.sleep(PRODUCER_SLEEP_QUANTUM)

messageQueue.put("Producer0: Message%d"%i, timeout=PRODUCER_TIMEOUT)

print("From Producer0: Added Message%d"%i)

except queue.Full:

print("From Producer0:Cannot put Message%d into the Queue"%i)

# Function for Producer process

def producer1(messageQueue):

try:

for i in range(1, PRODUCER_PROCESS_LIMIT+1):

time.sleep(PRODUCER_SLEEP_QUANTUM)

messageQueue.put("Producer1: Message%d"%i, timeout=PRODUCER_TIMEOUT)

print("From Producer1: Added Message%d"%i)

except queue.Full:

print("From Producer1:Cannot put Message%d into the Queue"%i)

# Function for Consumer process

def consumer(messageQueue):

try:

count = 0

maximum_consumption = PRODUCER_PROCESS_LIMIT * 2

while count < maximum_consumption:

print("From Consumer: printing %s"%messageQueue.get(timeout=CONSUMER_TIMEOUT))

count = count + 1

except queue.Empty:

print("Consumer:timeout...nothing to read")

if __name__ == "__main__":

multiprocessing.set_start_method = "fork"

messageQueue = multiprocessing.Queue(maxsize=PRODUCER_PROCESS_LIMIT)

# Create producer processes

producerProcess0 = multiprocessing.Process(target=producer0, args=(messageQueue,) )

producerProcess1 = multiprocessing.Process(target=producer1, args=(messageQueue,) )

# Create consumer process

consumerProcess = multiprocessing.Process(target=consumer, args=(messageQueue,) )

# Start the producer and consumer process

producerProcess0.start()

producerProcess1.start()

consumerProcess.start()

# Wait for the producer and consumer processes to complete

producerProcess0.join()

producerProcess1.join()

consumerProcess.join()

Output:

From Producer1: Added Message1

From Producer0: Added Message1

From Consumer: printing Producer1: Message1

From Consumer: printing Producer0: Message1

From Producer0: Added Message2

From Consumer: printing Producer0: Message2

From Producer1: Added Message2

From Consumer: printing Producer1: Message2

From Producer0: Added Message3

From Producer1: Added Message3

From Consumer: printing Producer0: Message3

From Consumer: printing Producer1: Message3

From Producer1: Added Message4

From Producer0: Added Message4

From Consumer: printing Producer0: Message4

From Consumer: printing Producer1: Message4