Join us and get access to thousands of tutorials and a community of expert Pythonistas.
This lesson is for members only. Join us and get access to thousands of tutorials and a community of expert Pythonistas.
In this lesson, you’ll finish creating your message queue and analyze the output of the program. If you download the sample code, you can get your own copy of 13-queue.py:
To learn more, you can also check out the documentation for threading.Event, the queue module, and super.
00:00 Let’s scroll up to the top of the output
00:05
and we see a bunch of producing messages printed here. So, that tells us that we no longer have to have one message produced at a time. We actually have more than one messages put on the queue before the consumer picks one off. The first one the consumer got was 76, which is the first one that was put on by the producer.
00:35
And we’re seeing queue size is 10. This is coming from within the .get_message() method. And so whenever the consumer thread reaches this point, it will print that, so we’re seeing queue size at these various times.
00:56 And as the consumer picks off these messages, the queue size starts to decrease. And notice how now most of the print statements are coming from the consumer.
01:07
The reason why the consumer is slower than the producer is because of our call to time.sleep() in the consumer thread. The producer doesn’t sleep in the beginning, so that’s why we see a bunch of producing messages here.
01:25
But in the end, through the power of leveraging this queue package, the consumer ends up picking off the messages in order with no deadlock and with no explicit assignment of any locks.
01:47
and we can see the queue size decreasing, and then the output is the same. And just for fun, let’s go ahead and change this maxsize to 0.
02:00 We’ll save that and run the program again. It looks like that maybe somewhere around 20,000 was the capacity on my machine. It may be different for your computer.
02:15
But we see there that roughly 20,000 messages were put on the queue and the consumer is slowly picking them off as it sleeps for a random interval between 0 and 1 second.
02:29
And this will go on and on until the queue size is 0 and then it will print a huge list, but we’re not going to wait for that.
02:38
We can, though, change maybe maxsize to 20,
02:47 and we should see a rush of 20 messages on the queue and then a slow one-by-one pick off from the consumer as it goes through. And then we get the output that is in order and good.
03:06
So, beautiful! Let’s look at our code and summarize what we’ve done. You’ve imported the queue package. You inherited that Queue class into your Pipeline.
03:20
You got rid of a bunch of code, got rid of that FINISH flag. You used the .put() and the .get() methods that come from the Queue class.
03:30
You learned about an Event, which is basically a global flag that any thread that is passed an Event has access to. And then to really fire off the events of your pipeline, you said event.set(), and then these threads started doing their thing.
03:50
You got a rush of messages from the producer and then the consumer picked them off one at a time in the order in which they were placed on the queue by the producer.
04:02 Congratulations on writing this awesome program. A lot of this code is actually print statements that we’re using to learn, so when you delete those, it’s going to look even cleaner, and you can actually use this to build your own message queues.
Bartosz Zaczyński RP Team on Aug. 31, 2020
In fact, it’s not always +2. The maxsize parameter determines the queue’s maximum capacity, while the total number of elements pushed through it will vary depending on how quickly they’re produced and consumed. When you comment out the time.sleep() line from the consumer, you’ll see many more elements in the output.
Angad on Dec. 10, 2021
Can you please explain what is the exit condition for the program? Why 12 numbers?
John McB on Jan. 30, 2025
I think your explanation around 03:40 is incorrect. I may be wrong but that’s how you learn.
Instead of event.set() starting things, it looks like it stops the producer from adding more things to the queue.
The producer function will add things to the queue until while not event.is_set() breaks.
You can test this by putting a small number into the time.sleep(...) call. If I use time.sleep(0.0005) on my laptop then the final pipeline lists are different lengths each time I run it, len=3 then 16 etc.
In reply to Angad’s question the exit condition is the not pipeline.empty() condition in the while loop of the consumer function. Once the consumer has emptied the Queue then the program exits.
Feedback appreciated as I may have misunderstood this - threading is hard.
Become a Member to join the conversation.
Pradeep Kumar on Aug. 29, 2020
It was a great tutorial, I am just curious to know, why it is giving +2 results, for example: maxsize = 10 , it is producing 12 outcomes and for maxsize = 20 , it is producing 22 outcomes. I am not sure why it is like that?