Какую структуру данных будешь использовать для реализации очереди на Python?

Структуры данных для реализации очереди в Python

Рекомендуемый выбор: collections.deque

Для реализации очереди (FIFO - First In, First Out) в Python следует использовать collections.deque - это оптимальный выбор из стандартной библиотеки.

from collections import deque

# Создание очереди
queue = deque()

# Добавление элемента в конец (enqueue)
queue.append(1)
queue.append(2)
queue.append(3)

# Удаление элемента с начала (dequeue)
element = queue.popleft()  # 1
element = queue.popleft()  # 2

Почему deque, а не list?

Сложность операций:

Для очереди критичны операции с концами, поэтому deque намного эффективнее.

Практическая реализация с deque

from collections import deque
from typing import Optional

class Queue:
    """Простая реализация очереди"""
    
    def __init__(self, maxsize: int = 0):
        self._queue = deque()
        self._maxsize = maxsize
    
    def put(self, item) -> None:
        """Добавить элемент в очередь"""
        if self._maxsize > 0 and len(self._queue) >= self._maxsize:
            raise OverflowError("Queue is full")
        self._queue.append(item)
    
    def get(self):
        """Получить элемент из очереди"""
        if not self._queue:
            raise IndexError("Queue is empty")
        return self._queue.popleft()
    
    def empty(self) -> bool:
        """Пуста ли очередь"""
        return len(self._queue) == 0
    
    def size(self) -> int:
        """Размер очереди"""
        return len(self._queue)

# Использование
q = Queue(maxsize=5)
q.put("task1")
q.put("task2")
q.put("task3")

print(q.get())  # task1
print(q.get())  # task2
print(q.size()) # 1

Альтернатива: queue.Queue для многопоточности

Если нужна потокобезопасная очередь для многопоточных приложений:

import queue
import threading

# Создание потокобезопасной очереди
thread_queue = queue.Queue(maxsize=10)

def producer():
    for i in range(5):
        thread_queue.put(f"item-{i}")
        print(f"Добавлен item-{i}")

def consumer():
    while True:
        item = thread_queue.get()
        print(f"Обработан {item}")
        thread_queue.task_done()

# Запуск в разных потоках
t1 = threading.Thread(target=producer)
t2 = threading.Thread(target=consumer, daemon=True)

t1.start()
t2.start()
t1.join()

Очередь с приоритетом: queue.PriorityQueue

import queue
import heapq

# Очередь с приоритетом
pq = queue.PriorityQueue()

# Элементы с приоритетом (меньше = выше приоритет)
pq.put((1, "high priority"))
pq.put((3, "low priority"))
pq.put((2, "medium priority"))

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

# Output:
# Priority 1: high priority
# Priority 2: medium priority
# Priority 3: low priority

Двусторонняя очередь с deque

from collections import deque

# Deque позволяет работать с обоих концов
dq = deque([1, 2, 3])

# Добавление
dq.append(4)       # Добавить справа: [1, 2, 3, 4]
dq.appendleft(0)   # Добавить слева: [0, 1, 2, 3, 4]

# Удаление
dq.pop()           # Удалить справа: [0, 1, 2, 3]
dq.popleft()       # Удалить слева: [1, 2, 3]

# Вращение
dq.rotate(1)       # Сдвинуть вправо: [3, 1, 2]

Сравнение подходов

Вывод

Рекомендация: Используй collections.deque для большинства случаев. Это стандартная, оптимальная по производительности реализация очереди. Если нужна многопоточность - переходи на queue.Queue.