В чём разница между queue и deque в collections?

Разница между queue и deque в collections

В Python модуль collections предоставляет deque, а модуль queue — класс Queue. Это разные структуры данных для разных задач.

deque (Double-Ended Queue)

deque — это структура данных из модуля collections, которая позволяет добавлять и удалять элементы с обоих концов очень быстро.

from collections import deque

# Создание
d = deque([1, 2, 3, 4, 5])
print(d)  # deque([1, 2, 3, 4, 5])

# Добавить в конец — O(1)
d.append(6)
print(d)  # deque([1, 2, 3, 4, 5, 6])

# Добавить в начало — O(1)
d.appendleft(0)
print(d)  # deque([0, 1, 2, 3, 4, 5, 6])

# Удалить с конца — O(1)
d.pop()  # Вернёт 6
print(d)  # deque([0, 1, 2, 3, 4, 5])

# Удалить с начала — O(1)
d.popleft()  # Вернёт 0
print(d)  # deque([1, 2, 3, 4, 5])

# Обращение (поворот)
d.rotate(2)  # Сдвигает на 2 позиции вправо
print(d)  # deque([4, 5, 1, 2, 3])

Операции deque:

Когда использовать deque:

• Скользящее окно
• Структуры данных, требующие быстрые операции с обоих концов
• Реализация стека и очереди

# Пример: скользящее окно (moving average)
from collections import deque

def moving_average(values, window_size=3):
    d = deque(maxlen=window_size)
    
    for value in values:
        d.append(value)
        if len(d) == window_size:
            yield sum(d) / window_size

numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9]
for avg in moving_average(numbers, 3):
    print(avg)  # 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0

Queue (из модуля queue)

Queue — это потокобезопасная очередь из модуля queue. Это синхронизированная структура данных для многопоточных приложений.

from queue import Queue

# Создание
q = Queue(maxsize=5)

# Добавить элемент
q.put(1)
q.put(2)
q.put(3)

# Получить элемент (FIFO)
print(q.get())  # 1
print(q.get())  # 2
print(q.get())  # 3

# Проверка пустоты
print(q.empty())  # True

# Размер (примерно, не гарантирован в многопоточности)
print(q.qsize())  # 0

Особенности Queue:

Когда использовать Queue:

• Многопоточные приложения
• Производитель-потребитель паттерн
• Обработка задач в отдельных потоках

import threading
from queue import Queue

def producer(q):
    for i in range(5):
        q.put(i)
        print(f"Produced: {i}")

def consumer(q):
    while True:
        item = q.get()
        if item is None:
            break
        print(f"Consumed: {item}")
        q.task_done()

q = Queue()

# Запускаем потоки
t1 = threading.Thread(target=producer, args=(q,))
t2 = threading.Thread(target=consumer, args=(q,))

t1.start()
t2.start()

t1.join()
q.put(None)  # Сигнал к завершению
t2.join()

Прямое сравнение

Практические примеры

1. Очередь задач с deque (однопоточный)

from collections import deque
import time

class TaskQueue:
    def __init__(self):
        self.tasks = deque()
    
    def add_task(self, task):
        self.tasks.append(task)
    
    def process_all(self):
        while self.tasks:
            task = self.tasks.popleft()
            task()
            time.sleep(0.1)

def task1():
    print("Task 1")

def task2():
    print("Task 2")

q = TaskQueue()
q.add_task(task1)
q.add_task(task2)
q.process_all()

2. BFS (поиск в ширину) с deque

from collections import deque

def bfs(graph, start):
    visited = set()
    queue = deque([start])
    visited.add(start)
    
    while queue:
        node = queue.popleft()
        print(node)
        
        for neighbor in graph[node]:
            if neighbor not in visited:
                queue.append(neighbor)
                visited.add(neighbor)

graph = {
    'A': ['B', 'C'],
    'B': ['D'],
    'C': ['E'],
    'D': [],
    'E': []
}

bfs(graph, 'A')  # A, B, C, D, E

3. Многопоточная очередь с Queue

import threading
from queue import Queue
import time

def worker(q, worker_id):
    while True:
        item = q.get()
        if item is None:
            break
        print(f"Worker {worker_id}: processing {item}")
        time.sleep(1)
        q.task_done()

q = Queue()
threads = []

# Создаём 3 работника
for i in range(3):
    t = threading.Thread(target=worker, args=(q, i))
    t.start()
    threads.append(t)

# Добавляем задачи
for i in range(10):
    q.put(f"Task-{i}")

# Ждём завершения
q.join()

# Останавливаем работников
for _ in threads:
    q.put(None)

for t in threads:
    t.join()

4. Очередь приоритета с Queue

from queue import PriorityQueue
import threading

pq = PriorityQueue()

# Элементы с приоритетом (priority, item)
pq.put((1, 'Low priority'))
pq.put((0, 'High priority'))
pq.put((2, 'Low priority 2'))

while not pq.empty():
    priority, item = pq.get()
    print(f"Priority {priority}: {item}")

Производительность

from collections import deque
from queue import Queue
import time

# Тест deque (однопоточный)
start = time.time()
d = deque()
for i in range(100000):
    d.append(i)
while d:
    d.popleft()
print(f"deque time: {time.time() - start:.3f}s")  # ~0.01s

# Тест Queue (однопоточный)
start = time.time()
q = Queue()
for i in range(100000):
    q.put(i)
while not q.empty():
    q.get()
print(f"Queue time: {time.time() - start:.3f}s")  # ~0.5s

deque быстрее в 50 раз! Queue медленнее из-за использования locks для потокобезопасности.

LifoQueue и другие варианты

from queue import Queue, LifoQueue, PriorityQueue

# FIFO (обычная очередь)
fifo = Queue()
fifo.put(1); fifo.put(2)
print(fifo.get())  # 1

# LIFO (стек)
lifo = LifoQueue()
lifo.put(1); lifo.put(2)
print(lifo.get())  # 2

# С приоритетом
pq = PriorityQueue()
pq.put((1, 'first')); pq.put((0, 'zero'))
print(pq.get())  # (0, 'zero')

Заключение

• deque: для быстрых операций с обоих концов очереди в однопоточном коде
• Queue: для безопасной обработки задач в многопоточных приложениях

Выбирай deque если работаешь в одном потоке и нужна скорость. Выбирай Queue если работаешь с несколькими потоками и нужна синхронизация.