PythonのQueueモジュールの使い方

PythonのQueueモジュールは、マルチスレッドプログラミングのような、複数のスレッド間でデータのやり取りをする際に有効な標準ライブラリです。Pythonでは、以下のようなキューが提供されています。

• queue.Queue：FIFO方式
• queue.LifoQueue：LIFO方式
• queue.PriorityQueue：優先度付き

本記事では、それぞれの動作と使い方について紹介します。
(サンプルスクリプトはPython3.7.6)

１．queue.Queue（FIFO方式）

基本的なキューで、最初に入れたデータが最初に取り出されます。
以下のサンプルスクリプトでは、キューに3つのデータを追加し、その後にデータの取り出しと、キューの中身の確認をしています。

sample1.py

import queue

## Queueの生成
q = queue.Queue(maxsize=3)

## データの追加
q.put(10)
q.put(20)
q.put(30)

## データの取り出し
print(f"empty: {q.empty()}, full: {q.full()}, size: {q.qsize()}")
print(f"get: {q.get()}")
print(f"empty: {q.empty()}, full: {q.full()}, size: {q.qsize()}")
print(f"get: {q.get()}")
print(f"empty: {q.empty()}, full: {q.full()}, size: {q.qsize()}")
print(f"get: {q.get()}")
print(f"empty: {q.empty()}, full: {q.full()}, size: {q.qsize()}")

empty: False, full: True, size: 3
get: 10
empty: False, full: False, size: 2
get: 20
empty: False, full: False, size: 1
get: 30
empty: True, full: False, size: 0

• queue.Queue(maxsize=0)
  maxsizeで設定した整数値を上限とするキューを生成します。デフォルトは0で、0以下の値を設定した場合、キューのサイズは無限大になります。

• put(data, block=True, timeout=None)
  キューにデータを追加します。
  ・data：キューに追加するデータ
  ・block：キューが上限に達したとき、Trueの場合は待機、Falseの場合はqueue.Full例外が発生します。
  ・timeout：block=Trueの場合、待機時間(秒)後に例外が発生します。

• get(block=True, timeout=None)
  キューからデータを取り出します。
  ・block：キューが上限に達したとき、Trueの場合は待機、Falseの場合はqueue.Empty例外が発生します。
  ・timeout：block=Trueの場合、待機時間(秒)後に例外が発生します。

• empty()
  キューが空かどうかを確認します。空の場合はTrueを返し、そうでなければFalseを返します。

• full()
  キューが上限に達しているかどうかを確認します。

• qsize()
  キューに格納されているデータの数を返します。

２．queue.LifoQueue（LIFO方式）

スタックのように、最後に入れたデータが最初に取り出されます。
以下のサンプルスクリプトでは、queue.Queueと同様に、キューに3つのデータを追加し、その後にデータの取り出しと、キューの中身の確認をしています。

sample2.py

import queue

## Queueの生成
q = queue.LifoQueue(maxsize=3)

## データの追加
q.put(10)
q.put(20)
q.put(30)

## データの取り出し
print(f"empty: {q.empty()}, full: {q.full()}, size: {q.qsize()}")
print(f"get: {q.get()}")
print(f"empty: {q.empty()}, full: {q.full()}, size: {q.qsize()}")
print(f"get: {q.get()}")
print(f"empty: {q.empty()}, full: {q.full()}, size: {q.qsize()}")
print(f"get: {q.get()}")
print(f"empty: {q.empty()}, full: {q.full()}, size: {q.qsize()}")

empty: False, full: True, size: 3
get: 30
empty: False, full: False, size: 2
get: 20
empty: False, full: False, size: 1
get: 10
empty: True, full: False, size: 0

• queue.LifoQueue(maxsize=0)
  maxsizeで設定した整数値を上限とするキューを生成します。デフォルトは0で、0以下の値を設定した場合、キューのサイズは無限大になります。

３．queue.PriorityQueue（優先度付き）

優先度に基づいてデータが取り出されます。データはタプルの形で管理することが多く、1番目の要素が優先度で、2番目の要素が実際のデータです。

sample3.py

import queue

## Queueの生成
q = queue.PriorityQueue(maxsize=4)

## データの追加
q.put((2, 20))
q.put((1, 10))
q.put((3, 30))
q.put((2, 40))

## データの取り出し
while not q.empty():
    priority, item = q.get()
    print(f"priority: {priority}, item: {item}")

priority: 1, item: 10
priority: 2, item: 20
priority: 2, item: 40
priority: 3, item: 30

• queue.PriorityQueue(maxsize=0)
  maxsizeで設定した整数値を上限とするキューを生成します。デフォルトは0で、0以下の値を設定した場合、キューのサイズは無限大になります。

• put((priority, data), block=True, timeout=None)
  データに優先度をつけてキューに追加します。取り出す際は、優先度が高い(priorityの値が小さい)ものから順に取り出します。仮に、優先度の同じデータが複数ある場合、データを追加(put)した順に従って取り出されます。

４．使用例

Queueモジュールは、複数のスレッドが同時にアクセスしても正しく動作する（スレッドセーフ）ので、非同期的にデータを追加したり、取り出したりする際に有効的です。以下のサンプルスクリプトは、queue.Queueを用いた例です。producerスレッドでキューにデータを追加し、consumerスレッドでデータを取り出します。処理を終了させるために、Noneをキューに追加して、consumerスレッドに終了シグナルを送っています。

sample4.py

import queue
import threading
import time

# キューを作成
q = queue.Queue()

# データを追加するスレッド
def producer():
    for i in range(5):
        print(f"producer : {i} をキューに追加")
        q.put(i)  # キューへデータを追加
        time.sleep(1)

# データを取り出すスレッド
def consumer():
    while True:
        data = q.get()  # キューからデータを取り出し
        if data is None:  # 終了シグナルを受け取ったらスレッドを終了する
            break
        print(f"consumer : {data} を取り出し")
        time.sleep(2)

# スレッドの作成
producer_thread = threading.Thread(target=producer)
consumer_thread = threading.Thread(target=consumer)

# スレッドの開始
producer_thread.start()
consumer_thread.start()

# スレッドの終了を待つ
producer_thread.join()

# 終了シグナルを送る
q.put(None)  # consumerスレッドを終了させるためにNoneを追加

# スレッドの終了を待つ
consumer_thread.join()
print("全ての処理が終了")

producer : 0 をキューに追加
consumer : 0 を取り出し
producer : 1 をキューに追加
consumer : 1 を取り出し
producer : 2 をキューに追加
producer : 3 をキューに追加
consumer : 2 を取り出し
producer : 4 をキューに追加
consumer : 3 を取り出し
consumer : 4 を取り出し
全ての処理が終了

５．まとめ

PythonのQueueモジュールは、スレッド間での安全で効率的なデータのやり取りを実現し、非同期処理や並行処理の設計において効果的なツールとなります。実際のシステムに応用することで、パフォーマンスを最大化し、複雑なタスクをスムーズに処理することができるようになります。本記事を参考にして、このツールを活用してみてください。