Skip to content

Python Deque:collections.dequeによる高速な両端キュー

Updated on

Pythonのリストをキューやスタックとして使うのは自然に見えます -- 左側からpopし始めるまでは。list.pop(0)の呼び出しはO(n)操作です。残りのすべての要素が1つ前にシフトする必要があるためです。数百万のアイテムを処理するループでは、この隠れたコストが高速なパイプラインを遅いものに変えます。同じ問題はlist.insert(0, x)にも影響し、これもO(n)です。シーケンスの両端に触れるあらゆるワークロードにおいて、プレーンなリストはパフォーマンスの罠になります。

Pythonのcollections.deque(両端キュー、「デック」と発音)は、両端からO(1)のappendとpop操作を提供することでこれを解決します。固定長ブロックの双方向リンクリストとして実装されており、インデックスアクセスもサポートしながら各端で定数時間のパフォーマンスを提供します。このガイドでは、dequeについて知る必要があるすべてをカバーします:作成、コア操作、制限付きキュー、スレッドセーフティ、実用的なパターン。

📚

Dequeの作成

from collections import deque
 
d = deque([1, 2, 3, 4, 5])
print(d)  # deque([1, 2, 3, 4, 5])
 
letters = deque("hello")
print(letters)  # deque(['h', 'e', 'l', 'l', 'o'])
 
d = deque()
 
d = deque(maxlen=3)
d.append(1)
d.append(2)
d.append(3)
d.append(4)  # 右に追加すると左端のアイテムが削除される
print(d)  # deque([2, 3, 4], maxlen=3)

コア操作

Dequeはシーケンスの両端に対する対称的な操作を提供します。

from collections import deque
 
d = deque([2, 3, 4])
 
d.append(5)
d.appendleft(1)
print(d)  # deque([1, 2, 3, 4, 5])
 
right = d.pop()
left = d.popleft()
print(right)  # 5
print(left)   # 1
 
d = deque([3, 4, 5])
d.extend([6, 7, 8])
d.extendleft([2, 1, 0])    # 注意:逆順になる
print(d)  # deque([0, 1, 2, 3, 4, 5, 6, 7, 8])

extendleftはアイテムを1つずつ処理し、各アイテムを位置0に挿入することに注意してください。結果の順序は入力イテラブルと比較して逆になります。

その他の便利なメソッド

from collections import deque
 
d = deque([1, 2, 3, 2, 4, 2, 5])
 
print(d.count(2))   # 3
print(d.index(3))   # 2
d.remove(2)
d.insert(1, 99)
d.reverse()
d.clear()

rotate()による回転

rotate()メソッドはdequeをn歩右に回転します。負のnは左に回転します。

from collections import deque
 
d = deque([1, 2, 3, 4, 5])
 
d.rotate(2)
print(d)       # deque([4, 5, 1, 2, 3])
 
d.rotate(-3)
print(d)       # deque([2, 3, 4, 5, 1])

maxlenによる制限付きDeque

制限付きdequeは最も実用的な機能の1つです。maxlenを設定することで、容量に達したときに最も古いアイテムを自動的に排出する固定サイズのバッファを作成します。

スライディングウィンドウ / 移動平均

from collections import deque
 
def moving_average(data, window_size):
    """Calculate moving average using a bounded deque."""
    window = deque(maxlen=window_size)
    averages = []
    for value in data:
        window.append(value)
        if len(window) == window_size:
            averages.append(sum(window) / window_size)
    return averages
 
prices = [100, 102, 104, 103, 105, 107, 106, 108, 110, 109]
print(moving_average(prices, 3))

ファイルの末尾(最後のN行)

from collections import deque
 
def tail(filename, n=10):
    """Return the last n lines of a file."""
    with open(filename) as f:
        return deque(f, maxlen=n)

Dequeをキュー(FIFO)として使う

from collections import deque
 
queue = deque()
queue.append("first")
queue.append("second")
queue.append("third")
 
print(queue.popleft())    # first
print(queue.popleft())    # second
print(queue.popleft())    # third

ここがdequeがリストに対して真に輝くところです。リストではlist.pop(0)はO(n)。dequeではpopleft()はO(1)です。

Dequeをスタック(LIFO)として使う

from collections import deque
 
stack = deque()
stack.append("first")
stack.append("second")
stack.append("third")
 
print(stack.pop())    # third
print(stack.pop())    # second
print(stack.pop())    # first

パフォーマンス比較:Deque vs リスト

操作dequelist備考
append(x)(右)O(1)O(1)償却両方とも右端追加は高速
appendleft(x) / insert(0, x)O(1)O(n)Dequeが圧倒的に速い
pop()(右)O(1)O(1)償却両方とも右端取り出しは高速
popleft() / pop(0)O(1)O(n)Dequeが圧倒的に速い
extend(iterable)O(k)O(k)償却k = イテラブルの長さ
rotate(k)O(k)N/Aリストに相当するものなし
インデックスアクセス d[i]O(n)O(1)ランダムアクセスはリストが勝つ
len()O(1)O(1)両方とも長さを内部追跡
in(所属テスト)O(n)O(n)両方とも線形スキャンが必要

ベンチマーク

from collections import deque
import time
 
def benchmark_left_operations(n):
    """Compare popleft/appendleft performance."""
    data = list(range(n))
    start = time.perf_counter()
    while data:
        data.pop(0)
    list_time = time.perf_counter() - start
 
    data = deque(range(n))
    start = time.perf_counter()
    while data:
        data.popleft()
    deque_time = time.perf_counter() - start
 
    print(f"n={n:>8,}")
    print(f"  list.pop(0):     {list_time:.4f}s")
    print(f"  deque.popleft(): {deque_time:.4f}s")
    print(f"  Speedup: {list_time / deque_time:.1f}x")
 
benchmark_left_operations(100_000)

スレッドセーフティ

Pythonのdequeはスレッドセーフなappend()appendleft()pop()popleft()操作を提供します。これらのメソッドはCPythonでアトミックです。GILで保護された単一のバイトコード命令内で実行されるためです。

ブロッキング動作が必要なプロダクションシステムには、代わりにqueue.Queueを使用してください。

実用的なパターン

BFSグラフ走査

from collections import deque
 
def bfs(graph, start):
    """Breadth-first search using deque as queue."""
    visited = set()
    queue = deque([start])
    order = []
    while queue:
        node = queue.popleft()
        if node not in visited:
            visited.add(node)
            order.append(node)
            queue.extend(
                neighbor for neighbor in graph[node]
                if neighbor not in visited
            )
    return order
 
graph = {
    'A': ['B', 'C'],
    'B': ['A', 'D', 'E'],
    'C': ['A', 'F'],
    'D': ['B'],
    'E': ['B', 'F'],
    'F': ['C', 'E']
}
 
print(bfs(graph, 'A'))  # ['A', 'B', 'C', 'D', 'E', 'F']

スライディングウィンドウの最大値

from collections import deque
 
def sliding_window_max(nums, k):
    """Find maximum in each sliding window of size k."""
    result = []
    window = deque()
    for i, num in enumerate(nums):
        while window and window[0] < i - k + 1:
            window.popleft()
        while window and nums[window[-1]] < num:
            window.pop()
        window.append(i)
        if i >= k - 1:
            result.append(nums[window[0]])
    return result
 
nums = [1, 3, -1, -3, 5, 3, 6, 7]
print(sliding_window_max(nums, 3))
# [3, 3, 5, 5, 6, 7]

元に戻す履歴

制限付きdequeは、最も古いアクションを自動的に忘れる元に戻す履歴として優れています:

from collections import deque
 
class TextEditor:
    def __init__(self, max_undo=50):
        self.content = ""
        self._undo_stack = deque(maxlen=max_undo)
 
    def type_text(self, text):
        self._undo_stack.append(self.content)
        self.content += text
 
    def delete_last(self, n=1):
        self._undo_stack.append(self.content)
        self.content = self.content[:-n]
 
    def undo(self):
        if self._undo_stack:
            self.content = self._undo_stack.pop()
 
editor = TextEditor(max_undo=10)
editor.type_text("Hello")
editor.type_text(" World")
print(editor.content)   # Hello World
editor.undo()
print(editor.content)   # Hello

Jupyterでデータ構造のパフォーマンスをテスト

dequeとリストのベンチマークや、特定のワークロードに合わせたmaxlenの調整を行う際、インタラクティブなノートブックは非常に価値があります。RunCell (opens in a new tab)はAI搭載のJupyter環境を提供し、データ構造操作のプロファイリング、パフォーマンスの違いの可視化、最適化のための知的な提案を受けることができます。

FAQ

Pythonのdequeとは?

deque(両端キュー)はPythonのcollectionsモジュールのデータ構造で、両端からO(1)のappendとpop操作をサポートします。前方からの挿入や削除にO(n)かかるリストとは異なり、dequeはどちらの端でも定数時間のパフォーマンスを提供します。

dequeとlistのどちらを使うべき?

シーケンスの両端で高速な挿入や削除が必要な場合、例えばキュー(FIFO)の実装、スライディングウィンドウの維持、BFS走査の実行にはdequeを使用してください。インデックスによる高速なランダムアクセスが必要な場合はlistを使用してください(リストはO(1)、dequeはO(n))。

Python dequeはスレッドセーフ?

はい、append()appendleft()pop()popleft()メソッドはCPythonでスレッドセーフです。GILで保護された単一のバイトコード命令内で完了するため、これらの操作はアトミックです。ただし、複合操作はアトミックではないため、ブロッキングセマンティクスが必要なプロデューサー-コンシューマーパターンにはqueue.Queueを使用してください。

dequeのmaxlenは何をする?

maxlenパラメータは固定最大サイズの制限付きdequeを作成します。dequeが満杯で新しいアイテムが追加されると、反対側のアイテムが自動的に破棄されます。これはローリングウィンドウ、最近のアクティビティログ、元に戻す履歴の実装に便利です。

dequeとqueue.Queueの比較は?

collections.dequeは速度のために設計されています。queue.Queueは内部的にdequeをラップしますが、ロックや条件変数などのスレッド同期プリミティブを追加します。シングルスレッドコードにはdequeを使用してください。マルチスレッドのプロデューサー-コンシューマーシナリオでブロッキングget()put()操作が必要な場合はqueue.Queueを使用してください。

まとめ

Pythonのcollections.dequeは標準ライブラリで最も有用でありながら過小評価されているデータ構造の1つです。リストの左側操作のO(n)ペナルティを排除し、maxlenによる組み込みの制限付きバッファリングを提供し、スレッドセーフなアトミック操作を提供し、キュー、スタック、スライディングウィンドウ、グラフ走査アルゴリズムの基盤として機能します。

重要な判断は明確です:コードがシーケンスの両端に触れる場合はdequeを使用してください。インデックスによる高速なランダムアクセスが必要な場合はlistを使用してください。ほとんどのキューとスタックのパターンでは、dequeがデフォルトの選択であるべきです。

📚