栈与队列

栈（Stack）

栈是一种后进先出（LIFO, Last In First Out）的数据结构，只允许在一端进行插入和删除操作。这一端称为栈顶，另一端称为栈底。

基本操作

• push：向栈顶添加元素
• pop：弹出栈顶元素
• top/peek：查看栈顶元素
• isEmpty：判断栈是否为空

class Stack:
    def __init__(self):
        self._data = []

    def push(self, val):
        self._data.append(val)

    def pop(self):
        if self.is_empty():
            raise IndexError("pop from empty stack")
        return self._data.pop()

    def top(self):
        if self.is_empty():
            raise IndexError("top from empty stack")
        return self._data[-1]

    def is_empty(self):
        return len(self._data) == 0

    def size(self):
        return len(self._data)

括号匹配

栈的经典应用——检验括号是否匹配。

def is_valid(s: str) -> bool:
    stack = []
    mapping = {')': '(', ']': '[', '}': '{'}

    for ch in s:
        if ch in '([{':
            stack.append(ch)
        else:
            if not stack or stack[-1] != mapping[ch]:
                return False
            stack.pop()

    return len(stack) == 0

# 测试
print(is_valid("()[]{}"))  # True
print(is_valid("([)]"))    # False

最小栈

设计一个支持 push、pop、top 操作，并能在 O(1) 时间内获取最小元素的栈。

class MinStack:
    def __init__(self):
        self._stack = []
        self._min_stack = []  # 辅助栈保存最小值

    def push(self, val):
        self._stack.append(val)
        if not self._min_stack or val <= self._min_stack[-1]:
            self._min_stack.append(val)

    def pop(self):
        if not self._stack:
            return None
        val = self._stack.pop()
        if val == self._min_stack[-1]:
            self._min_stack.pop()
        return val

    def top(self):
        return self._stack[-1] if self._stack else None

    def get_min(self):
        return self._min_stack[-1] if self._min_stack else None

栈与表达式

栈可以用于中缀表达式转后缀表达式，以及后缀表达式求值。

def infix_to_postfix(expr: str) -> str:
    """中缀转后缀"""
    precedence = {'+': 1, '-': 1, '*': 2, '/': 2, '^': 3}
    stack = []
    result = []

    for ch in expr.replace(' ', ''):
        if ch.isalnum():
            result.append(ch)
        elif ch == '(':
            stack.append(ch)
        elif ch == ')':
            while stack and stack[-1] != '(':
                result.append(stack.pop())
            stack.pop()
        else:
            while (stack and stack[-1] != '(' and
                   precedence.get(stack[-1], 0) >= precedence[ch]):
                result.append(stack.pop())
            stack.append(ch)

    while stack:
        result.append(stack.pop())

    return ''.join(result)

def eval_postfix(expr: str) -> int:
    """后缀表达式求值"""
    stack = []
    for ch in expr:
        if ch.isdigit():
            stack.append(int(ch))
        else:
            b, a = stack.pop(), stack.pop()
            if ch == '+': stack.append(a + b)
            elif ch == '-': stack.append(a - b)
            elif ch == '*': stack.append(a * b)
            elif ch == '/': stack.append(int(a / b))
    return stack[0]

print(infix_to_postfix("a+b*c"))     # abc*+
print(infix_to_postfix("(a+b)*c"))   # ab+c*
print(eval_postfix("123*+"))         # 1+2*3 = 7

---

队列（Queue）

队列是一种先进先出（FIFO, First In First Out）的数据结构，一端插入（队尾），另一端删除（队首）。

基本操作

• enqueue：向队尾添加元素
• dequeue：从队首移除元素
• front：查看队首元素
• isEmpty：判断队列是否为空

from collections import deque

class Queue:
    def __init__(self):
        self._data = deque()

    def enqueue(self, val):
        self._data.append(val)

    def dequeue(self):
        if self.is_empty():
            raise IndexError("dequeue from empty queue")
        return self._data.popleft()

    def front(self):
        if self.is_empty():
            raise IndexError("front from empty queue")
        return self._data[0]

    def is_empty(self):
        return len(self._data) == 0

    def size(self):
        return len(self._data)

循环队列

循环队列使用固定大小的数组，通过 head 和 tail 指针实现队列的循环利用，避免普通队列出队时整体平移的开销。

class CircularQueue:
    def __init__(self, k: int):
        self.k = k
        self._data = [None] * k
        self._head = 0
        self._tail = 0
        self._count = 0

    def enqueue(self, val) -> bool:
        if self.is_full():
            return False
        self._data[self._tail] = val
        self._tail = (self._tail + 1) % self.k
        self._count += 1
        return True

    def dequeue(self):
        if self.is_empty():
            return None
        val = self._data[self._head]
        self._data[self._head] = None
        self._head = (self._head + 1) % self.k
        self._count -= 1
        return val

    def front(self):
        if self.is_empty():
            return None
        return self._data[self._head]

    def rear(self):
        if self.is_empty():
            return None
        return self._data[(self._tail - 1) % self.k]

    def is_empty(self):
        return self._count == 0

    def is_full(self):
        return self._count == self.k

生产者-消费者问题

队列在线程同步中广泛应用，以下是使用 queue 模块模拟生产者-消费者问题：

import threading
import queue
import time

def producer(q, num_items):
    for i in range(num_items):
        item = f"item-{i}"
        q.put(item)
        print(f"生产者: 生产了 {item}")
        time.sleep(0.1)

def consumer(q, num_items):
    for _ in range(num_items):
        item = q.get()
        print(f"消费者: 消费了 {item}")
        time.sleep(0.15)
        q.task_done()

q = queue.Queue(maxsize=5)
threads = []
threads.append(threading.Thread(target=producer, args=(q, 5)))
threads.append(threading.Thread(target=consumer, args=(q, 5)))

for t in threads:
    t.start()
for t in threads:
    t.join()

print("所有任务完成")

---

单调栈（Monotonic Stack）

单调栈是一种特殊的栈，栈内元素保持单调递增或递减。常用于 O(n) 时间内解决「下一个更大元素」等问题。

框架

def monotonic_stack(nums: list[int], mode: str = 'inc') -> list[int]:
    """
    单调栈通用框架
    mode: 'inc' 升序, 'dec' 降序
    返回每个元素对应的"下一个更大/更小元素"的索引
    """
    n = len(nums)
    res = [-1] * n
    stack = []  # 存放下标

    for i in range(n):
        # 升序栈：遇到更大元素时弹出并更新结果
        # 降序栈：遇到更小元素时弹出并更新结果
        while (stack and
               (mode == 'inc' and nums[i] > nums[stack[-1]]) or
               (mode == 'dec' and nums[i] < nums[stack[-1]])):
            idx = stack.pop()
            res[idx] = nums[i]

        stack.append(i)

    return res

示例：每日温度（LeetCode 739）

给定气温数组，求每一天需要等多少天才能等到更高的温度。

def daily_temperatures(temps: list[int]) -> list[int]:
    """单调递减栈：存放还未遇到更高温度的日期"""
    n = len(temps)
    res = [0] * n
    stack = []  # 存放下标，栈中温度单调递减

    for i in range(n):
        while stack and temps[i] > temps[stack[-1]]:
            prev = stack.pop()
            res[prev] = i - prev
        stack.append(i)

    return res

# 测试
print(daily_temperatures([73, 74, 75, 71, 69, 72, 76, 73]))
# 输出: [1, 1, 4, 2, 1, 1, 0, 0]

示例：柱状图中的最大矩形（LeetCode 84）

def largest_rectangle_area(heights: list[int]) -> int:
    """单调递增栈 + 哨兵技巧"""
    if not heights:
        return 0

    # 前后加 0 作为哨兵，简化边界处理
    h = [0] + heights + [0]
    stack = [0]  # 栈中存放下标，高度单调递增
    max_area = 0

    for i in range(1, len(h)):
        while h[i] < h[stack[-1]]:
            height = h[stack.pop()]
            width = i - stack[-1] - 1
            max_area = max(max_area, height * width)
        stack.append(i)

    return max_area

print(largest_rectangle_area([2, 1, 5, 6, 2, 3]))  # 10

---

优先队列（Priority Queue）

优先队列是一种每个元素都有优先级，优先级最高的元素最先出队的队列。通常用堆实现。

堆的基本实现

import heapq

class MinHeap:
    def __init__(self):
        self._data = []

    def push(self, val):
        heapq.heappush(self._data, val)

    def pop(self):
        return heapq.heappop(self._data)

    def peek(self):
        return self._data[0] if self._data else None

    def is_empty(self):
        return len(self._data) == 0

    def size(self):
        return len(self._data)


class MaxHeap:
    """通过取负值实现最大堆"""
    def __init__(self):
        self._data = []

    def push(self, val):
        heapq.heappush(self._data, -val)

    def pop(self):
        return -heapq.heappop(self._data)

    def peek(self):
        return -self._data[0] if self._data else None

    def is_empty(self):
        return len(self._data) == 0

Top-K 问题

def top_k(nums: list[int], k: int) -> list[int]:
    """找出第 k 大的元素（使用最小堆）"""
    if k <= 0:
        return []

    heap = MinHeap()
    for num in nums:
        heap.push(num)
        if heap.size() > k:
            heap.pop()

    result = []
    while not heap.is_empty():
        result.append(heap.pop())
    return result

# 使用 heapq 模块更简洁
def top_k_heapq(nums: list[int], k: int) -> list[int]:
    return heapq.nlargest(k, nums)

print(top_k_heapq([3, 2, 1, 5, 6, 4], 2))  # [6, 5]

合并 K 个有序链表

import heapq

class ListNode:
    def __init__(self, val=0, next=None):
        self.val = val
        self.next = next

def merge_k_lists(lists: list[ListNode]) -> ListNode:
    """使用最小堆合并 K 个有序链表"""
    dummy = ListNode(0)
    cur = dummy
    heap = []

    # 初始化：把每个链表的第一个节点加入堆
    for i, node in enumerate(lists):
        if node:
            heapq.heappush(heap, (node.val, i, node))

    while heap:
        val, i, node = heapq.heappop(heap)
        cur.next = node
        cur = cur.next
        if node.next:
            heapq.heappush(heap, (node.next.val, i, node.next))

    return dummy.next

滑动窗口中位数（LeetCode 480）

import heapq

class SlidingWindowMedian:
    def __init__(self):
        self._max_heap = []  # 较小的一半（用负值模拟最大堆）
        self._min_heap = []  # 较大的一半
        self._delayed = {}   # 延迟删除标记
        self._k = 0
        self._left_count = 0

    def _clean(self, heap):
        """清理已标记删除的元素"""
        while heap:
            val = -heap[0] if heap is self._max_heap else heap[0]
            if val in self._delayed and self._delayed[val] > 0:
                heapq.heappop(heap)
                self._delayed[val] -= 1
                if self._delayed[val] == 0:
                    del self._delayed[val]
            else:
                break

    def _rebalance(self):
        """平衡两个堆的元素数量"""
        if self._left_count > (self._k + 1) // 2:
            # 左侧多了，移到右侧
            val = -heapq.heappop(self._max_heap)
            self._clean(self._max_heap)
            self._left_count -= 1
            heapq.heappush(self._min_heap, val)
        elif self._left_count < self._k // 2:
            # 右侧多了，移到左侧
            val = heapq.heappop(self._min_heap)
            self._clean(self._min_heap)
            self._left_count += 1
            heapq.heappush(self._max_heap, -val)

    def _peek(self, heap):
        self._clean(heap)
        val = -heap[0] if heap is self._max_heap else heap[0]
        return val

    def median_sliding_window(self, nums: list[int], k: int) -> list[float]:
        self._k = k
        self._left_count = 0
        result = []

        for i, num in enumerate(nums):
            # 加入新元素
            if not self._max_heap or num <= self._peek(self._max_heap):
                heapq.heappush(self._max_heap, -num)
                self._left_count += 1
            else:
                heapq.heappush(self._min_heap, num)

            self._rebalance()

            # 达到窗口大小，取中位数
            if i >= k - 1:
                if k % 2 == 1:
                    median = self._peek(self._max_heap)
                else:
                    median = (self._peek(self._max_heap) + self._peek(self._min_heap)) / 2
                result.append(median)

                # 移除左侧第一个元素
                to_remove = nums[i - k + 1]
                self._delayed[to_remove] = self._delayed.get(to_remove, 0) + 1
                if to_remove <= self._peek(self._max_heap):
                    self._left_count -= 1
                self._clean(self._max_heap)
                self._clean(self._min_heap)
                self._rebalance()

        return result

---

总结对比

数据结构	操作复杂度	特性	典型应用
栈	O(1) push/pop	LIFO	括号匹配、函数调用栈、表达式求值
队列	O(1) enqueue/dequeue	FIFO	任务调度、BFS、消息队列
单调栈	均摊 O(1)	维护单调性	下一个更大元素、柱状图最大矩形
优先队列	O(log n) push/pop	按优先级出队	Top-K、中位数、合并有序序列

[[返回 算法首页|../index]]