Queue ใน Python คืออะไร?
Queue เป็น Linear Data Structure ทำงานแบบ First In First Out (FIFO) Element ที่เข้าไปก่อนจะถูก Process ก่อน (เหมือนคิวรอรถบัส)
ตัวอย่างในชีวิตจริง
คิวรอรถบัส: คนมาก่อนอยู่หน้า คนตามมาต่อท้าย รถมา คนหน้าขึ้นก่อน (First In First Out)
การ Implement Queue ใน Python
วิธีที่ 1: ใช้ List (ไม่แนะนำ - ช้า O(n) สำหรับ pop(0))
queue = []
queue.append(1)
queue.append(2)
queue.append(3)
print("Initial queue:", queue) # [1, 2, 3]
print("Element popped from the queue")
print(queue.pop(0)) # 1 - O(n) ต้อง Shift ทุก Element
print(queue.pop(0)) # 2
print("Queue after popping:", queue) # [3]
# Pop from empty -> IndexError
# queue.pop(0)
วิธีที่ 2: ใช้ queue.Queue (Thread-Safe, Built-in)
from queue import Queue
q = Queue(maxsize=3) # maxsize=0 = unlimited
q.put(1)
q.put(2)
q.put(3)
print("Is queue full:", q.full()) # True
print("Element popped from the queue")
print(q.get()) # 1
print(q.get()) # 2
print("Number of elements in queue:", q.qsize()) # 1
print("Is queue empty:", q.empty()) # False
Methods ของ Queue
| Method | Description |
|---|---|
| put(item) | Enqueue, block ถ้า Full |
| put_nowait(item) | Enqueue, raise Exception ถ้า Full |
| get() | Dequeue, block ถ้า Empty |
| get_nowait() | Dequeue, raise Exception ถ้า Empty |
| full() | True ถ้า Full |
| empty() | True ถ้า Empty |
| qsize() | จำนวน Elements |
วิธีที่ 3: ใช้ collections.deque (แนะนำ - เร็วที่สุด O(1))
from collections import deque
queue = deque()
queue.append(1)
queue.append(2)
queue.append(3)
print("Initial queue:", queue) # deque([1, 2, 3])
print("Element popped from the queue")
print(queue.popleft()) # 1 - O(1)!
print(queue.popleft()) # 2
print("Queue after popping:", queue) # deque([3])
# popleft from empty -> IndexError
# queue.popleft()
ทำไม deque ดีกว่า List สำหรับ Queue?
append()(Enqueue) = O(1)popleft()(Dequeue) = O(1)- List:
pop(0)= O(n) ต้อง Shift Elements ทั้งหมด
Queue Class แบบ Custom
class Queue:
def __init__(self):
self._items = deque()
def enqueue(self, item):
self._items.append(item)
def dequeue(self):
if self.is_empty():
raise IndexError("Dequeue from empty queue")
return self._items.popleft()
def peek(self):
if self.is_empty():
raise IndexError("Peek from empty queue")
return self._items[0]
def is_empty(self):
return len(self._items) == 0
def size(self):
return len(self._items)
def __str__(self):
return str(list(self._items))
# ใช้งาน
q = Queue()
q.enqueue("A")
q.enqueue("B")
print(q.peek()) # A
print(q.dequeue()) # A
print(q) # ['B']
ประยุกต์ใช้: BFS (Breadth-First Search)
from collections import deque
def bfs(graph, start):
visited = set()
queue = deque([start])
visited.add(start)
result = []
while queue:
vertex = queue.popleft()
result.append(vertex)
for neighbor in graph[vertex]:
if neighbor not in visited:
visited.add(neighbor)
queue.append(neighbor)
return result
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']
Priority Queue (คิวลำดับความสำคัญ)
import heapq
# heapq เป็น Min-Heap
pq = []
heapq.heappush(pq, (2, "Medium"))
heapq.heappush(pq, (1, "High"))
heapq.heappush(pq, (3, "Low"))
while pq:
priority, task = heapq.heappop(pq)
print(f"Priority {priority}: {task}")
# Priority 1: High
# Priority 2: Medium
# Priority 3: Low