【Python】スレッドセーフなキューでタイムアウトを使用する

2026年1月14日

概要

queue.Queue の put() と get() は、デフォルトでは処理が完了するまで「無限に待機（ブロック）」します。

timeout 引数を指定することで、一定時間待っても処理が完了しない場合に例外（Full または Empty）を発生させ、プログラムがフリーズするのを防ぐことができます。

仕様（入出力）

入力: タイムアウト時間（秒数）
出力: 成功時はデータ、失敗時（タイムアウト時）は例外発生
前提: import queue が必要。

メソッドと処理

メソッド・構文	引数例	処理内容
`q.put(item, timeout=秒)`	`timeout=3`	キューにデータを入れます。満杯の場合、指定秒数だけ空きができるのを待ちます。それでも満杯なら `queue.Full` 例外を発生させます。
`q.get(timeout=秒)`	`timeout=3`	キューからデータを取り出します。空の場合、指定秒数だけデータが入るのを待ちます。それでも空なら `queue.Empty` 例外を発生させます。

例外クラスと説明

例外クラス	発生条件	説明
`queue.Full`	`put()` 失敗時	キューの容量制限（`maxsize`）に達しており、タイムアウト時間内に空きができなかった場合に発生します。
`queue.Empty`	`get()` 失敗時	キューにデータがなく、タイムアウト時間内に新しいデータが追加されなかった場合に発生します。

基本の使い方

タイムアウトを指定する場合は、必ず try-except ブロックで例外を捕捉するように記述します。

1. putのタイムアウト（満杯時の処理）

キューのサイズ制限（maxsize）がある場合に発生します。

import queue

# 容量が1つだけのキューを作成
q = queue.Queue(maxsize=1)

# 1つ入れて満杯にする
q.put("Data1")
print("1つ目を追加しました")

try:
    # 満杯状態で2つ目を入れようとする（2秒待機）
    print("2つ目を追加しようとしています...")
    q.put("Data2", timeout=2)
except queue.Full:
    print("タイムアウト：キューが満杯で追加できませんでした。")

2. getのタイムアウト（空の時の処理）

データが来るのを待ち続けるのを防ぎます。

import queue

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

try:
    # データがない状態で取り出そうとする（2秒待機）
    print("データを待っています...")
    item = q.get(timeout=2)
except queue.Empty:
    print("タイムアウト：キューが空で取得できませんでした。")

コード全文

これらを組み合わせたデモ関数です。スレッド間の通信において、データが来ない場合に無限待ちせずにループを抜ける際などによく使われるパターンです。

import queue
import time

def queue_timeout_demo():
    print("--- put(timeout) のデモ ---")
    # maxsize=1 なのですぐに満杯になります
    q_limit = queue.Queue(maxsize=1)
    
    q_limit.put("First Item") # OK
    print("キューの状態: 満杯")
    
    try:
        # 満杯なのでブロックされます。1秒後に諦めます。
        start_time = time.time()
        q_limit.put("Second Item", timeout=1.0)
    except queue.Full:
        elapsed = time.time() - start_time
        print(f"エラー捕捉: queue.Full ({elapsed:.2f}秒経過)")

    print("\n--- get(timeout) のデモ ---")
    # 空のキュー
    q_empty = queue.Queue()
    print("キューの状態: 空")
    
    try:
        # 空なのでブロックされます。1.5秒後に諦めます。
        start_time = time.time()
        data = q_empty.get(timeout=1.5)
    except queue.Empty:
        elapsed = time.time() - start_time
        print(f"エラー捕捉: queue.Empty ({elapsed:.2f}秒経過)")

if __name__ == "__main__":
    queue_timeout_demo()