Pythonのqueueをわかりやすく解説：基本的な使い方から実務での活用シーンまで

プログラミング言語•
Python

はじめに

皆さんはPythonで複数の処理をスムーズに連携させたいと考えることはないでしょうか。

処理のやり取りを安全かつ効率的に行う方法として、Python標準ライブラリのqueueモジュールがよく使われます。

このモジュールを利用すると、複数のタスクを順番に処理したり、並列で動かしているスレッド間でのデータの受け渡しを簡単に行えます。

プログラミングを始めたばかりだと、「queueって一体何をするもの？」「どうやってコードに組み込んだらいいの？」と疑問に思うかもしれませんね。

そこで本記事では、Pythonのqueueモジュールを中心に、初心者の方でもわかりやすい説明を心がけながら、具体的なコード例や実務での活用シーンを交えて紹介していきます。

なお、専門用語はなるべく平易に説明し、実際の現場でも役立つような視点から解説を進めます。

この記事を読むとわかること

  • Pythonのqueueモジュールでできることと、その使い方の基本
  • FIFO（先入れ先出し）やLIFO（後入れ先出し）、優先度付きキューなどの使い分け
  • マルチスレッド処理におけるキュー活用例
  • よくあるエラーや例外への対処方法
  • 実務レベルでのキューの活用シーンとメリット

Pythonのqueueとは何か

最初に、Pythonのqueueモジュールがどのような働きをするものかを簡単に整理してみましょう。

queue (キュー)とは「データを一時的に蓄え、順番に取り出す仕組み」のことを指します。

身近な例でいえば、レジの列を思い浮かべるとわかりやすいかもしれません。

例えばレジで並んでいる人がいれば、先に並んだ人から順番に会計されますよね。

これが「FIFO（先入れ先出し）」というキューの基本的な考え方です。

Pythonでは、このキュー機能を**queueモジュール**として標準で提供しています。

データ構造を自作する必要がなく、手軽に順番制御ができるのが強みです。

queueが活躍する実務シーンの例

実務では、複数の処理を同時に走らせることがよくあります。

例えば、次のような場面でキューは非常に便利です。

  • バックグラウンド処理：サーバーが受け付けたタスクを順番に処理したいとき
  • マルチスレッドでのタスク管理：一つのスレッドが追加したデータを、別のスレッドで処理するとき
  • 一定順序でイベントを処理：ゲームやチャットアプリでのイベントを順番に取り出し、対応する処理を当てはめるとき

特に、マルチスレッドの場面では、データを安全に受け渡しできることが重要です。

キューを使うと、スレッド同士で同じリストや配列を触ってしまい、思わぬエラーが発生するといったリスクを減らせます。

実務での安定稼働や並列処理の安定性を確保するうえで、スレッドセーフなデータ構造であることは大きなメリットですね。

Pythonのqueueモジュール概要

queueモジュールでは、主に以下の三つのクラスが提供されています。

  1. Queue
    • もっとも基本的な先入れ先出し（FIFO）のキュー
  2. LifoQueue
    • 後入れ先出し（LIFO）のキュー
    • スタックのように後に入れたものから先に取り出す
  3. PriorityQueue
    • 優先度を設定して、優先度の高い順に取り出す
    • 同時に入れたタスクのなかでも、重要なものから処理したい場合に使う

これらはいずれもスレッドセーフに設計されています。

つまり、複数のスレッドが同時にデータを入れたり、取り出したりしても安全に動作します。

たとえばQueueクラスを使うだけで、データ競合を意識せずに処理を組み立てられるのがありがたいところです。

queueを使うための基本コード

では、実際に Queue (FIFO) の例を見てみましょう。

import queue

# Queueオブジェクトの生成
task_queue = queue.Queue()

# データをキューに入れる（put）
task_queue.put("タスク1")
task_queue.put("タスク2")
task_queue.put("タスク3")

# データをキューから取り出す（get）
first_task = task_queue.get()
print(first_task)  # タスク1

second_task = task_queue.get()
print(second_task)  # タスク2

third_task = task_queue.get()
print(third_task)  # タスク3

キューを使う際のポイントとしては、以下のような点を押さえておくと良いでしょう。

  • putメソッド：キューにデータを格納する
  • getメソッド：キューからデータを1つ取り出す
  • queue.Queue()のカッコ内に整数を指定すると、キューの最大サイズを設定できる

後から解説しますが、もし指定した上限を超えるデータを入れようとすると、デフォルトの動作ではキューが空きになるまでputでブロック（待機）します。

LifoQueueとPriorityQueueの基本的な使い方

LifoQueue（スタック的な動作）

LifoQueueは後に入れたデータほど先に取り出せるという特徴があります。

import queue

lifo = queue.LifoQueue()
lifo.put("A")
lifo.put("B")
lifo.put("C")

print(lifo.get())  # C
print(lifo.get())  # B
print(lifo.get())  # A

このように、入れた順番と逆の順番で取り出されるので、スタック操作に近いイメージです。

PriorityQueue（優先度に応じて取り出す）

PriorityQueueでは、データを入れるときに優先度を指定して、その大小関係で取り出される順番が変わります。

例えば、数値の小さいものを優先して取り出すように書くと次のようになります。

import queue

pq = queue.PriorityQueue()
pq.put((2, "通常タスク"))
pq.put((1, "重要タスク"))
pq.put((3, "後回しタスク"))

# 優先度の小さい順に取り出し
print(pq.get())  # (1, "重要タスク")
print(pq.get())  # (2, "通常タスク")
print(pq.get())  # (3, "後回しタスク")

ここでは数値が優先度を表す形になっています。

例えば「(1, "重要タスク")」のように、タプルで優先度を先に置いておけば、優先度を数字で表現しやすいです。

数値が小さいほど優先的に取り出される挙動になります。

この使い方は、たとえば「新規機能の開発よりもバグ修正を優先して処理したい」というような実務の場面で便利です。

マルチスレッド処理とqueue

キューが特に威力を発揮するのは、マルチスレッド環境でデータのやりとりを行うときです。

複数のスレッドが同時に操作しても、内部で排他制御（データ競合が起きないための安全策）が行われます。

そのため、スレッド同士で安全にタスクやデータを渡すことができます。

次の例では、1つのスレッドがキューにデータを入れ、別のスレッドがそれを取り出して処理する流れを簡単に示しています。

import queue
import threading
import time

def producer(q):
    for i in range(5):
        item = f"メッセージ{i}"
        q.put(item)
        print(f"producerスレッド：{item}をキューに追加")
        time.sleep(0.5)  # 何らかの処理を想定

def consumer(q):
    while True:
        msg = q.get()
        print(f"consumerスレッド：{msg}を受け取りました")
        q.task_done()  # 処理が完了したことを知らせる

        # このサンプルでは終了条件を明示していないため
        # 無限に待機する形になっています
        # 本来は何らかの終了条件を判断してbreakするのが自然です

q = queue.Queue()
producer_thread = threading.Thread(target=producer, args=(q,))
consumer_thread = threading.Thread(target=consumer, args=(q,), daemon=True)

producer_thread.start()
consumer_thread.start()

producer_thread.join()
# producer_threadが終わったらqueue内の残りを待機する
q.join()  # 全てのtask_done()が呼ばれるまでブロック

コードのポイント

  • producer関数：キューに「メッセージ0」「メッセージ1」…といった文字列を入れていく
  • consumer関数：キューからデータを取り出し、取り出したことを表示する
  • q.task_done()：q.get()で取り出した項目の処理が完了したことを通知
  • q.join()：すべてのtask_done()が呼び出されるまで待機

こうすることで、並列処理中でも明確に「キューにデータがいつまで残っているか」を把握しやすくなり、安全にスレッド間通信を行えます。

queueのブロッキング動作とタイムアウト

キューを使うとき、デフォルトではブロッキング動作が行われます。

つまり、次のような振る舞いになることを意味します。

  • q.put(item) でキューが満杯（最大サイズを指定した場合）になっていると、空きができるまで待機
  • q.get() でキューに何も入っていないと、データが入ってくるまで待機

もし待機ではなく、すぐにタイムアウトしてほしいケースがあれば、put()やget()にtimeout引数を指定できます。

import queue

q = queue.Queue(maxsize=2)

try:
    q.put("A", timeout=1)
    q.put("B", timeout=1)
    # ここは最大サイズを超えるため、1秒待っても空きが出なければエラー
    q.put("C", timeout=1)
except queue.Full:
    print("キューが満杯のため、データを追加できませんでした。")

queue.Emptyやqueue.Fullといった例外も用意されており、エラー処理を細かく制御できます。

実務では、ネットワーク通信中にデータのやり取りが滞るケースもあるので、タイムアウトを設定しておくことでプログラムの停止を回避したいときに役立ちます。

queueのサイズ管理

先ほどの例のようにQueue()へ最大サイズを設定すると、内部のバッファ容量が制限されます。

大規模なデータをやり取りする場合、満杯になると新しいデータが入れられなくなります。

もしキューサイズを無制限に設定してしまうと、想定以上にメモリを消費してしまうリスクもあります。

キューサイズの使いどころ

  • メモリ節約：無制限にキューにため込むとメモリが不足する恐れがある
  • 処理負荷の制御：一定以上の処理が同時に溜まらないようにすることで、処理リソースを守る

現場では、やり取りするデータ量を想定しながら、適切に最大サイズを決めることが大切ですね。

queueの中身を確認する方法

開発時に「今、キューの中にどれぐらいデータが入っているかを確認したい」という場面はよくあります。

そんなときは、以下のメソッドで確認できます。

  • q.qsize()：キューの要素数を返す
  • q.empty()：キューが空ならTrue、そうでなければFalse
  • q.full()：キューが満杯ならTrue、そうでなければFalse

ただし、マルチスレッド環境では、一瞬前の状態とは異なる結果になりうるので、あくまで目安として把握しておくのが良いでしょう。

タスクの終了を知らせるための機能

マルチスレッドでキューを使っていると、すべてのタスクが完了したらアプリケーションを終了させたい場面が出てきます。

そんなときに便利なのが、先ほどの例にもあったtask_done()とjoin()です。

task_done()

  • get()で取得したタスクの処理が終わったことを明示する

join()

  • すべてのタスクが完了するまでブロックする

この仕組みがあるおかげで、「プロデューサ側がキューに投入したタスクが、全て処理されるまで待ちたい」という制御がシンプルに書けます。

以下はもう少し具体的なコード例です。

import queue
import threading
import time

def worker(q, worker_id):
    while True:
        try:
            task = q.get(timeout=1)
        except queue.Empty:
            # 一定時間データが来なければスレッド終了
            break

        print(f"worker{worker_id}: {task}を処理中")
        time.sleep(0.2)  # 処理にかかる時間を想定
        q.task_done()

task_q = queue.Queue()
tasks = [f"タスク{i}" for i in range(10)]
for t in tasks:
    task_q.put(t)

threads = []
for i in range(3):
    t = threading.Thread(target=worker, args=(task_q, i))
    threads.append(t)
    t.start()

task_q.join()
print("全タスク処理が完了しました。")

上記では、以下のような流れになっています。

  1. キューに10個のタスクを投入
  2. 3つのスレッドが並行で作業を行う
  3. worker関数でq.get(timeout=1)を使い、タイムアウトしたら終了
  4. すべてのタスクに対してtask_done()が呼ばれると、最後のtask_q.join()が解除
  5. 「全タスク処理が完了しました。」が表示される

こういった仕組みは、一定量のタスクを並行処理させつつ、確実に最後まで実行したい場合に有効です。

queueを使った並列処理のメリットとデメリット

メリット

  • スレッドセーフにデータを扱えるため、競合バグを起こしにくい
  • FIFOやLIFO、優先度付きなど、多様なデータ処理パターンをシンプルなコードで実現
  • タスクの終了を待機してから次のステップに進むなど、制御が行いやすい

デメリット

  • 適切なキューサイズを設定しないと、メモリ不足や処理詰まりになる
  • スレッド数が多すぎると、管理が複雑になりやすい
  • リアルタイム性能が必要なケース（例えば大量のデータが瞬間的に来る場面）では、より高度な最適化が必要

ただ初心者の方がPythonで並列処理を触り始めるときには、queueモジュールでの学習がとても良いステップになると思います。

queueと他のデータ構造との違い

「そもそもPythonではリストや辞書など、他にもいろいろなデータ構造があるけど、なぜキューなの？」と疑問に感じる方もいるかもしれません。

実際、リストでも要素の追加や削除はできますよね。

しかし、リストは複数のスレッドから同時に書き込みや読み取りを行うと、データ競合を起こしやすいです。

データ競合とは、一方のスレッドが途中の状態のリストにアクセスしてしまい、データ不整合が起こる状況です。

このような不具合は、システムが大規模化すると予想外のタイミングで発生し、バグの特定が非常に難しくなります。

一方で、queueモジュールのキューは内部でロック処理を行っているため、複数スレッドが同時に読み書きしても原則として安全を保てるのです。

実務レベルでの安定稼働には欠かせない仕組みというわけですね。

イベント駆動型プログラムとqueue

GUIアプリやチャットアプリなど、ユーザーからのイベントを順番に処理したい場面でもキューがよく使われます。

例えば、画面上でボタンをクリックしたり、キーボード操作が入ったり、ネットワークからメッセージが届いたりと、イベントは多方向から飛んできます。

これらをそのまま一つのリストに入れてしまうと、どれを先に処理すべきかが曖昧になるかもしれません。

そこでFIFOのキューを使えば、イベントを受け取った順に処理する流れをきちんと担保できるのです。

特に、チャットアプリやリアルタイム通信などでは、同時に複数のユーザーアクションが来ることも珍しくありません。

そうした場合にキューを導入することで、処理順序の混乱を避けやすくなります。

キューを使ったタスクスケジューリングの一例

もう少し実践的な例として、サーバー上で一定間隔ごとにタスクを処理する「スケジューラ」を考えてみましょう。

例えば、ユーザーへの通知メールをまとめて送るタスクがあるとします。

このとき、通知メールの内容（ユーザーIDやメール本文など）をキューにためておき、一定時間ごとにメール送信用のスレッドを起動してまとめて処理するといった仕組みが考えられます。

import queue
import threading
import time

mail_queue = queue.Queue()

def add_mail_task(user_id, content):
    mail_queue.put((user_id, content))

def mail_sender():
    while True:
        try:
            user_id, content = mail_queue.get(timeout=5)
            # ここでメール送信処理をすると仮定
            print(f"メール送信：ユーザー{user_id}に'{content}'を送りました")
            mail_queue.task_done()
        except queue.Empty:
            # 5秒間何もなければ一度ループを抜ける
            print("キューにメールタスクがありませんでした")
            break

# メールタスクを登録
add_mail_task(101, "新しいメッセージがあります")
add_mail_task(102, "パスワードが変更されました")
add_mail_task(103, "サーバーメンテナンスのお知らせ")

sender_thread = threading.Thread(target=mail_sender)
sender_thread.start()

sender_thread.join()
print("メール送信タスク処理終了")

このような仕組みなら、必要に応じてタスクをどんどんキューに放り込み、ある程度溜まったら送信処理を走らせるといった流れを作りやすいです。

さらに、必要ならスケジューラで定期的にmail_senderを起動するよう設定すれば、定期処理も自動化できます。

ここでもキューを介しておけば、複数のスレッドや関数が同時にメールタスクを追加しても安全に運用できるのがメリットです。

queueを使う際によくあるエラーと対処

queueを使っていると、次のようなエラーに遭遇することがあります。

queue.Empty

q.get_nowait() など、非ブロッキングモードで取得しようとしたときに、キューが空だと出る

queue.Full

キューが満杯の場合に、非ブロッキングモードで put_nowait() などを使うと出る

デッドロック

複数のスレッドがお互いに待ち状態になってしまい、プログラムが進まない状況

エラーが起きたときには、まず**try-exceptで例外処理**を組み込みましょう。

デッドロックを避けるためには、以下のような点に気を配ります。

  • ブロッキングメソッド（putやget）を使う場合は、timeoutを設定する
  • タスクが完了したら、必ずtask_done()を呼ぶ
  • join()の使いどころに注意し、無制限に待つことがないようにする

こうしたポイントを押さえれば、queueを使った並列処理は比較的スムーズに実現できます。

実務での活用例：ログの集約

例えば、サーバーのログを集約する仕組みにもqueueはよく使われます。

複数のスレッドやプロセスが同時に動いていて、それぞれが独自にログを書き込むと、ファイルアクセスが衝突したり、ログの順番がバラバラになったりする可能性があります。

そこで、すべてのログメッセージをキューに集め、ログ専用のスレッドがキューから取り出してファイルに書き込むようにすれば、ファイルアクセスを一元管理できます。

これにより、ログの順序が乱れず、ファイルへのアクセスも衝突しにくくなります。

実務での活用例：ワーカーキュー

マイクロサービスなどで、サーバー間で非同期処理を分担するケースがあります。

その際、キューを使ってタスクをワーカーに振り分けるワーカーキューのアーキテクチャを組むことが多いです。

Pythonだけでなく、全体のシステム設計としてRabbitMQやRedis Pub/Subなどを使うこともありますが、ローカルのスレッドレベルではqueueモジュールを基本にして同じ概念を実現できます。

あらゆる並列処理や非同期処理の入り口として、Pythonのqueueは理解しておくと役立つでしょう。

Pythonのqueueはシンプルに書ける反面、実際の大規模アプリケーションでは、ネットワーク上でキューを管理する場合も多いです。まずはqueueを使った並列処理をしっかり身につけると応用が効きます。

queueを使うときの設計ポイント

初心者のうちから意識しておくと良い設計ポイントをいくつかまとめておきます。

  1. キューサイズの上限を設定する
    • キューが無制限だと、メモリを圧迫し続ける可能性がある
  2. タイムアウト付きのgetやputを活用する
    • 必要以上にスレッドが待たされる事態を防ぐ
  3. スレッドの数を最適化する
    • スレッドを増やすほど処理が並列化されるわけではなく、逆に管理が大変になることもある
  4. モジュールやクラス単位で整理する
    • producerとconsumerなど、役割の違う処理は分割し、可読性を高める

こうした点を押さえておけば、チーム開発でも他の人がコードを読みやすくなり、バグのリスクを下げられます。

代表的なメソッド一覧

メソッド概要
put(item)データをキューに追加（ブロッキング）
put(item, False)非ブロッキングでキューに追加
get()キューからデータを取得（ブロッキング）
get(False)非ブロッキングでキューからデータを取得
qsize()キューの現在の要素数を返す
empty()キューが空かどうかを返す（True/False）
full()キューが満杯かどうかを返す（True/False）
task_done()get()したタスクの処理が完了したことを通知
join()すべてのタスクの処理完了まで待機

上の表にあるように、ブロッキングと非ブロッキングを使い分けることで柔軟な制御が可能です。

テストコードでの利用例

プログラムの単体テストや結合テストを書くときにも、キューは重宝します。

例えば、ある関数が非同期でメッセージを生成するコードをテストしたい場合、実際にはそのメッセージをキューに入れさせておいて、テスト側でキューをチェックするといった手法が使えます。

これにより、**「本当に必要な数のメッセージが生成されたか」**を検証しやすくなるのです。

import queue
import unittest

def async_process(q):
    for i in range(3):
        q.put(f"データ{i}")

class TestAsyncProcess(unittest.TestCase):
    def test_async_process(self):
        q = queue.Queue()
        async_process(q)

        # 予想どおり3件データが入っているかチェック
        self.assertEqual(q.qsize(), 3)

        # 取り出してみる
        for i in range(3):
            item = q.get()
            self.assertIn("データ", item)

if __name__ == "__main__":
    unittest.main()

こうすることで、非同期部分であっても動作の結果を検証しやすくなります。

queueとともに理解しておきたいライブラリ

Pythonで並列処理・並行処理を扱う場合、multiprocessing や concurrent.futures とあわせて学ぶと理解が深まることが多いです。

  • multiprocessing：マルチプロセスで並列に処理するためのライブラリ（CPUリソースをフルに活用したいとき）
  • concurrent.futures：スレッドプールやプロセスプールを抽象的に扱う仕組み

これらの仕組みでも、同様の考え方（キューのように安全にデータをやりとりする手法）が登場します。

ただ、まずは**queueモジュール**で基本の概念を掴んでおくと、後々スムーズにステップアップしやすいでしょう。

実際の大規模システムでは、threadingよりもmultiprocessingを使う場合もありますが、queueモジュールの考え方は同じです。まずはqueueで入門してみるのが一番わかりやすいかもしれません。

まとめ

ここまで、Pythonのqueueモジュールを中心に、基本的な使い方や実務での活用シーンを解説してきました。

キューの最大の特徴は、先入れ先出しや優先度などの仕組みを簡単に実装しつつ、スレッドセーフに動作させられる点です。

実務でも、タスクの振り分けや並列処理において非常に役立ちます。

初心者のうちは「スレッド間で安全にデータを渡す」ことがどれだけ難しいかピンと来ないかもしれません。

しかし、キューを使えば難しいロック機構や排他制御を自力で書く必要がほとんどなくなるのです。

以下のポイントを意識してぜひ活用してみてください。

  • FIFO、LIFO、PriorityQueueなど、用途に応じた選択をする
  • ブロッキング動作とタイムアウトの仕組みを理解しておく
  • task_done()とjoin()を使ってタスクの終了を管理する
  • 最大サイズやスレッド数などを適切に設定し、デッドロックやリソース不足を防ぐ

これらを踏まえて、シンプルな例題から試してみるのがおすすめです。

プログラムが複雑になるほど、こうしたキューの仕組みがより大きな恩恵をもたらしてくれるでしょう。

皆さんの開発において、Pythonのqueueモジュールがスレッドセーフな並列処理を効率的に実現する有用なツールとして活躍することを願っています。

Pythonをマスターしよう

この記事で学んだPythonの知識をさらに伸ばしませんか？
Udemyには、現場ですぐ使えるスキルを身につけられる実践的な講座が揃っています。

Pythonの人気講座ランキングを見る筆者おすすめ No.1 講座！「現役シリコンバレーエンジニアが教えるPython 3 入門 + 応用 +アメリカのシリコンバレー流コードスタイル」
プログラミング言語の新着記事をチェック

【PHP】implodeとは？配列を文字列に結合する方法を初心者向けにわかりやすく解説