Python | データ構造強化：キュー

Python Python
スポンサーリンク

概要（キューは「先に入れたものが先に出る」安全な順序制御の基本構造）

キューは、FIFO（First-In, First-Out）で要素を管理するデータ構造です。タスクの順序制御、イベントの順序処理、スレッド間の安全な受け渡しに向いています。Pythonでは「標準の list をキューに使うのは非推奨（先頭取り出しが遅い）」で、代わりに collections.deque（高速な両端操作）か queue.Queue（スレッドセーフ）が定番です。用途に応じて「速さ優先（deque）」「安全優先（queue.Queue）」を使い分けます。

基本の使い方（ここが重要）

deque で軽量・高速なキュー（単スレッドや簡易用途向け）

from collections import deque

q = deque()            # 空のキュー
q.append("A")          # 入れる（enqueue）
q.append("B")
x = q.popleft()        # 取り出す（dequeue）→ 'A'
print(x, list(q))      # A ['B']
Python

deque は先頭（popleft）・末尾（append）がどちらも平均 O(1) で高速です。list の pop(0) は O(n) なので避けてください。

queue.Queue でスレッドセーフなキュー（並行処理向け）

import queue

q = queue.Queue(maxsize=3)  # 上限を付けられる（省略可）
q.put("A")                  # 追加（ブロッキング）
q.put("B")
x = q.get()                 # 取り出し（ブロッキング）→ 'A'
q.task_done()               # 取り出したタスクの完了通知
q.join()                    # すべてのタスクが完了するまで待つ（任意）
Python

Queue は複数スレッドからの安全な put/get を提供します。get/put はデフォルトで「待つ（ブロックする）」動作なので、非ブロッキングやタイムアウトを使い分けるのが重要です。

q.get(block=False)            # 空なら例外 queue.Empty
q.put("X", timeout=0.5)       # 0.5秒だけ待って満杯なら例外 queue.Full
Python

重要ポイントの深掘り（選び方・ブロッキング・エラー・バリエーション）

どれを使うか（状況別の最適解）

  • 単スレッドで高速に回したい: deque（append/popleft）。
  • スレッド間で安全に受け渡し: queue.Queue（put/get、task_done/join）。
  • 最後に入れたものを先に出したい: queue.LifoQueue（スタック的動作）。
  • 優先度付きで取り出したい: queue.PriorityQueue（小さい値ほど優先）。
  • 軽量なスレッドセーフキュー（タスク完了機能不要）: queue.SimpleQueue。

ブロッキング動作を理解して「止まらない」設計に

  • 満杯で put が止まる: maxsize を適切に設定し、消費側を十分に回す。必要なら timeout や block=False を使用。
  • 空で get が止まる: producer（投入側）の生存を監視し、終了時は「センチネル（例：None）」を入れて消費側を終了させる。
import queue, threading, time

END = object()
q = queue.Queue()

def producer():
    for i in range(5):
        q.put(i)
    q.put(END)  # 終了シグナル

def consumer():
    while True:
        x = q.get()
        if x is END:
            break
        print("consume:", x)
        q.task_done()

threading.Thread(target=producer).start()
threading.Thread(target=consumer).start()
Python

よくある例外と対処

  • queue.Empty: 非ブロッキング get やタイムアウトで「空」。再試行するか待機戦略を設計。
  • queue.Full: 非ブロッキング put やタイムアウトで「満杯」。バッファ拡張か流量制御を検討。
  • デッドロック: producer/consumer の片方が停止しているのに片方が待ち続ける。センチネルで終了合図、join/task_done の対応関係を正しく保つ。

実務での使いどころ（スケジューリング・並行処理・優先度制御）

タスクスケジューリング（順番保証とバックプレッシャー）

「先に入れたものから必ず処理」の保証が必要なワークキューは queue.Queue を使い、maxsize でバックプレッシャー（投入側を適度に待たせる）をかけると暴走を防げます。

import queue, threading

q = queue.Queue(maxsize=100)

def worker():
    while True:
        task = q.get()
        if task is None:      # 終了シグナル
            break
        task()                # 実行
        q.task_done()

threads = [threading.Thread(target=worker, daemon=True) for _ in range(4)]
for t in threads: t.start()

# 仕事を投入
for _ in range(1000):
    q.put(lambda: None)       # ダミー
# 終了を知らせる
for _ in threads: q.put(None)
q.join()
Python

優先度付き処理（緊急タスクを先に）

小さい値ほど先に取り出されます。タプルで「優先度、シーケンス、実体」を入れると安定動作します。

import queue
pq = queue.PriorityQueue()
pq.put((0, 100, "urgent"))    # 優先度0（最優先）
pq.put((10, 101, "normal"))
print(pq.get())               # (0, 100, 'urgent')
Python

LIFO（後入れ先出し）でキャッシュ的処理

最近入ったものを先に処理したい（例：深さ優先探索）なら LifoQueue。

import queue
stack = queue.LifoQueue()
stack.put("A")
stack.put("B")
print(stack.get())            # 'B'
Python

注意点と最適化（list を避ける・センチネル・計測）

list をキューに使わない理由（性能）

list の pop(0) は前詰めが必要で O(n)。データ量が増えると急激に遅くなります。deque の popleft は O(1)。

センチネルで安全終了

キューの消費ループは「いつ終わるか」を外から判断しにくいので、必ず「終了合図（センチネル）」を流すか、キュー空＋producer 終了の条件を設計します。

実際に計測してボトルネックを見つける

put/get 自体がボトルネックになるケースは多くありません。処理本体が重いことが多いので、キューのサイズ・ワーカー数・待ち戦略を調整し、計測で決めるのが王道です。

例題で身につける（定番から一歩先まで）

例題1：deque で行列の順次処理

from collections import deque

q = deque(["task1", "task2", "task3"])
while q:
    t = q.popleft()
    print("do:", t)
Python

例題2：queue.Queue でプロデューサ／コンシューマ

import queue, threading, time

q = queue.Queue()

def producer():
    for i in range(5):
        q.put(f"T{i}")
        time.sleep(0.1)
    q.put(None)  # 終了

def consumer():
    while True:
        x = q.get()
        if x is None:
            break
        print("handle:", x)
        q.task_done()

threading.Thread(target=producer).start()
threading.Thread(target=consumer).start()
q.join()
Python

例題3：PriorityQueue で緊急度順に処理

import queue

pq = queue.PriorityQueue()
pq.put((2, "normal"))
pq.put((0, "urgent"))
pq.put((1, "soon"))

while not pq.empty():
    prio, msg = pq.get()
    print(prio, msg)
Python

例題4：非ブロッキングで空・満杯を扱う

import queue

q = queue.Queue(maxsize=1)
q.put("A")
try:
    q.put("B", block=False)  # 満杯なので例外
except queue.Full:
    print("full")

try:
    q.get()
    q.get(block=False)       # 空なので例外
except queue.Empty:
    print("empty")
Python

まとめ

キューは「順序保証」と「安全な受け渡し」のための標準手段です。単スレッド・軽量用途は deque、並行処理は queue.Queue を使う。ブロッキングの理解（空で get が止まる、満杯で put が止まる）とセンチネル設計、例外（Empty/Full）への対処、maxsize によるバックプレッシャーが重要。用途に応じて LifoQueue や PriorityQueue、SimpleQueue を選び、list の pop(0) は避ける。まずは小さなプロデューサ／コンシューマで手を動かし、停止条件と終了合図を「コードで明示」する癖をつけると、実務でも壊れない並行処理が書けます。

スポンサーリンク
タイトルとURLをコピーしました