{"id":314,"date":"2024-01-28T10:45:38","date_gmt":"2024-01-28T10:45:38","guid":{"rendered":"https:\/\/learnpython.elegantwallp.com\/?p=314"},"modified":"2024-01-28T10:45:39","modified_gmt":"2024-01-28T10:45:39","slug":"python-thread-safe-queue","status":"publish","type":"post","link":"https:\/\/learnpython.elegantwallp.com\/2024\/01\/28\/python-thread-safe-queue\/","title":{"rendered":"Python Thread-safe Queue"},"content":{"rendered":"\n

Summary<\/strong>: in this tutorial, you\u2019ll learn how to use a Python thread-safe queue to exchange data safely between multiple threads.<\/p>\n\n\n\n

Introduction to the Python thread-safe queue<\/h2>\n\n\n\n

The built-in\u00a0queue<\/code>\u00a0module allows you to exchange data safely between\u00a0multiple threads. The\u00a0Queue<\/code>\u00a0class in the\u00a0queue<\/code>\u00a0module implements all required locking semantics.<\/p>\n\n\n\n

Creating a new queue<\/h3>\n\n\n\n

To create a new queue, you import the Queue class from the queue module:<\/p>\n\n\n\n

from queue import Queue<\/code>Code language: Python (python)<\/small><\/code><\/pre>\n\n\n\n
and use the\u00a0Queue<\/code>\u00a0constructor as follows:<\/p>\n\n\n\n
queue = Queue()<\/code>Code language: Python (python)<\/small><\/code><\/pre>\n\n\n\n
To create a queue with a size limit, you can use the\u00a0maxsize<\/code>\u00a0parameter. For example, the following creates a queue that can store up to 10 items:<\/p>\n\n\n\n
queue = Queue(maxsize=10)<\/code>Code language: Python (python)<\/small><\/code><\/pre>\n\n\n\n
Adding an item to the queue<\/h3>\n\n\n\n
To add an item to the queue, you use the\u00a0put()<\/code>\u00a0method like this:<\/p>\n\n\n\n
queue.add(item)<\/code>Code language: Python (python)<\/small><\/code><\/pre>\n\n\n\n
Once the queue is full, you won\u2019t be able to add an item to it. Also, the call to the put()<\/code> method will block until the queue has space available.<\/p>\n\n\n\n
If you don\u2019t want the\u00a0put()<\/code>\u00a0method to block if the queue is full, you can set the\u00a0block<\/code>\u00a0argument to\u00a0False<\/code>:<\/p>\n\n\n\n
queue.put(item, block=False)<\/code>Code language: Python (python)<\/small><\/code><\/pre>\n\n\n\n
In this case, the\u00a0put()<\/code>\u00a0method will raise the\u00a0queue.Full<\/code>\u00a0exception if the queue is full:<\/p>\n\n\n\n
try: queue.put(item, block=False) except queue.Full as e: # handle exceptoin<\/em><\/code>Code language: Python (python)<\/small><\/code><\/pre>\n\n\n\n
To add an item to a sized limited queue and block with a timeout, you can use the\u00a0timeout<\/code>\u00a0parameter like this:<\/p>\n\n\n\n
try: queue.put(item, timeout=3) except queue.Full as e: # handle exceptoin<\/em><\/code>Code language: Python (python)<\/small><\/code><\/pre>\n\n\n\n
Getting an item from the queue<\/h3>\n\n\n\n
To get an item from the queue, you can use the\u00a0get()<\/code>\u00a0method:<\/p>\n\n\n\n
item = queue.get()<\/code>Code language: Python (python)<\/small><\/code><\/pre>\n\n\n\n
The get()<\/code> method will block until an item is available for retrieval from the queue.<\/p>\n\n\n\n
To get an item from the queue without blocking, you can set the block parameter to\u00a0False<\/code>:<\/p>\n\n\n\n
try: queue.get(block=False) except queue.Empty: # handle exception<\/em><\/code>Code language: Python (python)<\/small><\/code><\/pre>\n\n\n\n
To get an item from the queue and block it with a time limit, you can use the\u00a0get()<\/code>\u00a0method with a timeout:<\/p>\n\n\n\n
try: item = queue.get(timeout=10) except queue.Empty: # ...<\/em><\/code>Code language: Python (python)<\/small><\/code><\/pre>\n\n\n\n
Getting the size of the queue<\/h3>\n\n\n\n
The\u00a0qsize()<\/code>\u00a0method returns the number of items in the queue:<\/p>\n\n\n\n
size = queue.size()<\/code>Code language: Python (python)<\/small><\/code><\/pre>\n\n\n\n
Also, the empty()<\/code> method returns True<\/code> if the queue is empty or False otherwise. On the other hand, the full()<\/code> method returns True if the queue is full or False<\/code> otherwise.<\/p>\n\n\n\n
Marking a task as completed<\/h3>\n\n\n\n
An item that you add to the queue represents a unit of work or a task.<\/p>\n\n\n\n
When a thread calls the get()<\/code> method to get the item from the queue, it may need to process it before the task is considered completed.<\/p>\n\n\n\n
Once completed, the thread may call the\u00a0task_done()<\/code>\u00a0method of the queue to indicate that it has processed the task completely:<\/p>\n\n\n\n
item = queue.get() # process the item<\/em> # ...<\/em> # mark the item as completed<\/em> queue.task_done()<\/code>Code language: Python (python)<\/small><\/code><\/pre>\n\n\n\n
Waiting for all tasks on the queue to be completed<\/h3>\n\n\n\n
To wait for all tasks on the queue to be completed, you can call the\u00a0join()<\/code>\u00a0method on the queue object:<\/p>\n\n\n\n
queue.join()<\/code>Code language: Python (python)<\/small><\/code><\/pre>\n\n\n\n
Python thread-safe queue example<\/h2>\n\n\n\n
The following example illustrates how to use the thread-safe queue to exchange data between two threads:<\/p>\n\n\n\n
import time from queue import Empty, Queue from threading import Thread def producer(queue): for i in range(1, 6): print(f'Inserting item {i} into the queue') time.sleep(1) queue.put(i) def consumer(queue): while True: try: item = queue.get() except Empty: continue else: print(f'Processing item {item}') time.sleep(2) queue.task_done() def main(): queue = Queue() # create a producer thread and start it<\/em> producer_thread = Thread( target=producer, args=(queue,) ) producer_thread.start() # create a consumer thread and start it<\/em> consumer_thread = Thread( target=consumer, args=(queue,), daemon=True ) consumer_thread.start() # wait for all tasks to be added to the queue<\/em> producer_thread.join() # wait for all tasks on the queue to be completed<\/em> queue.join() if __name__ == '__main__': main()<\/code>Code language: Python (python)<\/small><\/code><\/pre>\n\n\n\n
How it works.<\/p>\n\n\n\n
First, define the\u00a0producer()<\/code>\u00a0function that adds numbers from 1 to 11 to the queue. It delays one second in each iteration:<\/p>\n\n\n\n
def producer(queue): for i in range(1, 6): print(f'Inserting item {i} into the queue') time.sleep(1) queue.put(i)<\/code>Code language: Python (python)<\/small><\/code><\/pre>\n\n\n\n
Second, define the\u00a0consumer()<\/code>\u00a0function that gets an item from the queue and processes it. It delays two seconds after processing each item on the queue:<\/p>\n\n\n\n
def consumer(queue): while True: try: item = queue.get() except Empty: continue else: print(f'Processing item {item}') time.sleep(2) queue.task_done()<\/code>Code language: Python (python)<\/small><\/code><\/pre>\n\n\n\n
The queue.task_done()<\/code> indicates that the function has processed the item on the queue.<\/p>\n\n\n\n
Third, define the\u00a0main()<\/code>\u00a0function that creates two threads, one thread adds a number to the queue every second while another thread processes an item on the queue every two seconds:<\/p>\n\n\n\n
def main(): queue = Queue() # create a producer thread and start it<\/em> producer_thread = Thread( target=producer, args=(queue,) ) producer_thread.start() # create a consumer thread and start it<\/em> consumer_thread = Thread( target=consumer, args=(queue,), daemon=True ) consumer_thread.start() # wait for all tasks to be added to the queue<\/em> producer_thread.join() # wait for all tasks on the queue to be completed<\/em> queue.join()<\/code>Code language: Python (python)<\/small><\/code><\/pre>\n\n\n\n
Output:<\/p>\n\n\n\n
Inserting item 1 into the queue Inserting item 2 into the queue Processing item 1 Inserting item 3 into the queue Processing item 2 Inserting item 4 into the queue Inserting item 5 into the queue Processing item 3 Processing item 4 Processing item 5<\/code>Code language: Python (python)<\/small><\/code><\/pre>\n\n\n\n
The following are steps in the main()<\/code> function:<\/p>\n\n\n\n
    \n
  1. Create a new queue by calling the Queue()<\/code> constructor<\/li>\n\n\n\n
  2. Create a new thread called producer_thread<\/code> and start it immediately<\/li>\n\n\n\n
  3. Create a\u00a0daemon thread\u00a0called\u00a0consumer_thread<\/code>\u00a0and start it immediately.<\/li>\n\n\n\n
  4. Wait for all the numbers to be added to the queue using the join()<\/code> method of the thread.<\/li>\n\n\n\n
  5. Wait for all the tasks on the queue to be completed by calling the join()<\/code> method of the queue.<\/li>\n<\/ol>\n\n\n\n
    The producer adds a number to the queue every second, and the consumer process a number from the queue every two seconds. It also displays the numbers on the queue every second.<\/p>\n","protected":false},"excerpt":{"rendered":"
    Summary: in this tutorial, you\u2019ll learn how to use a Python thread-safe queue to exchange data safely between multiple threads. Introduction to the Python thread-safe queue The built-in\u00a0queue\u00a0module allows you to exchange data safely between\u00a0multiple threads. The\u00a0Queue\u00a0class in the\u00a0queue\u00a0module implements all required locking semantics. Creating a new queue To create a new queue, you import […]<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[41],"tags":[],"class_list":["post-314","post","type-post","status-publish","format-standard","hentry","category-1-python-concurrency"],"_links":{"self":[{"href":"https:\/\/learnpython.elegantwallp.com\/wp-json\/wp\/v2\/posts\/314","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/learnpython.elegantwallp.com\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/learnpython.elegantwallp.com\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/learnpython.elegantwallp.com\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/learnpython.elegantwallp.com\/wp-json\/wp\/v2\/comments?post=314"}],"version-history":[{"count":1,"href":"https:\/\/learnpython.elegantwallp.com\/wp-json\/wp\/v2\/posts\/314\/revisions"}],"predecessor-version":[{"id":315,"href":"https:\/\/learnpython.elegantwallp.com\/wp-json\/wp\/v2\/posts\/314\/revisions\/315"}],"wp:attachment":[{"href":"https:\/\/learnpython.elegantwallp.com\/wp-json\/wp\/v2\/media?parent=314"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/learnpython.elegantwallp.com\/wp-json\/wp\/v2\/categories?post=314"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/learnpython.elegantwallp.com\/wp-json\/wp\/v2\/tags?post=314"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}