Как ускорить очередь, сделанную через список, в Python?

Как ускорить очередь, сделанную через список, в Python?

Очередь (queue), реализованная через обычный список с операциями pop(0) и append(), очень неэффективна — удаление первого элемента имеет сложность O(n), так как требует сдвига всех остальных элементов.

Проблема: Медленная реализация через список

# ❌ Неэффективно
class SlowQueue:
    def __init__(self):
        self.items = []
    
    def enqueue(self, item):
        self.items.append(item)  # O(1)
    
    def dequeue(self):
        return self.items.pop(0)  # O(n) — медленно!

# Для каждого dequeue() приходится сдвигать все элементы в памяти

Решение 1: collections.deque (РЕКОМЕНДУЕТСЯ)

deque (double-ended queue) из модуля collections оптимизирована для добавления/удаления с обеих сторон — O(1):

from collections import deque

class FastQueue:
    def __init__(self):
        self.items = deque()
    
    def enqueue(self, item):
        self.items.append(item)  # O(1)
    
    def dequeue(self):
        return self.items.popleft()  # O(1) — быстро!

# Использование
queue = FastQueue()
queue.enqueue(1)
queue.enqueue(2)
queue.enqueue(3)
print(queue.dequeue())  # 1

Решение 2: queue.Queue (для многопоточности)

Если нужна потокобезопасность (несколько потоков читают/пишут):

from queue import Queue

# Создаём очередь потокобезопасную
q = Queue(maxsize=100)  # опционально можно ограничить размер

# Добавление (thread-safe)
q.put(item)

# Получение (thread-safe, блокирует если пуста)
try:
    item = q.get(timeout=1)
except:
    print("Очередь пуста")

# Проверка размера (примерно)
print(q.qsize())

Решение 3: queue.LifoQueue (для стека)

Если нужен стек (LIFO), используй LifoQueue:

from queue import LifoQueue

stack = LifoQueue()
stack.put(1)
stack.put(2)
print(stack.get())  # 2 (последний добавленный)

Сравнение производительности

import time
from collections import deque

# Медленная реализация (список)
def slow_queue(n):
    items = []
    for i in range(n):
        items.append(i)
    for _ in range(n):
        items.pop(0)  # O(n) операция!

# Быстрая реализация (deque)
def fast_queue(n):
    items = deque()
    for i in range(n):
        items.append(i)
    for _ in range(n):
        items.popleft()  # O(1) операция

# Бенчмарк
for n in [1000, 10000, 100000]:
    start = time.time()
    slow_queue(n)
    slow_time = time.time() - start
    
    start = time.time()
    fast_queue(n)
    fast_time = time.time() - start
    
    print(f"n={n}: list={slow_time:.4f}s, deque={fast_time:.6f}s, ускорение={slow_time/fast_time:.0f}x")

# Вывод примерно:
# n=1000: list=0.0005s, deque=0.0001s, ускорение=5x
# n=10000: list=0.0450s, deque=0.0002s, ускорение=225x
# n=100000: list=4.5000s, deque=0.0020s, ускорение=2250x

Альтернативное решение: Кольцевой буфер (ring buffer)

Для фиксированного размера очереди можно использовать кольцевой буфер:

class RingBuffer:
    def __init__(self, size):
        self.size = size
        self.buffer = [None] * size
        self.head = 0
        self.tail = 0
        self.count = 0
    
    def enqueue(self, item):
        if self.count == self.size:
            raise OverflowError("Queue is full")
        
        self.buffer[self.tail] = item
        self.tail = (self.tail + 1) % self.size
        self.count += 1
    
    def dequeue(self):
        if self.count == 0:
            raise IndexError("Queue is empty")
        
        item = self.buffer[self.head]
        self.head = (self.head + 1) % self.size
        self.count -= 1
        return item

# Использование
rb = RingBuffer(5)
rb.enqueue(1)
rb.enqueue(2)
rb.enqueue(3)
print(rb.dequeue())  # 1
print(rb.dequeue())  # 2

Когда использовать какой вариант?

Вывод

• НИКОГДА не используй list.pop(0) для очереди
• ВСЕГДА используй collections.deque для однопоточного кода
• ВСЕГДА используй queue.Queue для многопоточного кода
• deque в 100-1000 раз быстрее, чем список для очередей