计算机基础

06. 搜索与哈希

掌握二分查找、哈希表、布隆过滤器与跳表：从 O(log n) 到 O(1) 的搜索效率进化，理解各类数据结构的权衡与设计。

spcent 5 分钟阅读 2024 字

1.1 基本思想

在已排序数组中，每次将搜索范围减半。

def binary_search(arr, target):
    """
    返回 target 的索引，不存在返回 -1
    循环不变量：target 在 [left, right] 范围内
    """
    left, right = 0, len(arr) - 1
    while left <= right:
        mid = left + (right - left) // 2  # 防溢出
        if arr[mid] == target:
            return mid
        elif arr[mid] < target:
            left = mid + 1
        else:
            right = mid - 1
    return -1

# 找左边界（第一个 ≥ target 的位置）
def lower_bound(arr, target):
    left, right = 0, len(arr)
    while left < right:
        mid = left + (right - left) // 2
        if arr[mid] < target:
            left = mid + 1
        else:
            right = mid
    return left

# 找右边界（第一个 > target 的位置）
def upper_bound(arr, target):
    left, right = 0, len(arr)
    while left < right:
        mid = left + (right - left) // 2
        if arr[mid] <= target:
            left = mid + 1
        else:
            right = mid
    return left

1.2 二分查找的扩展应用

# 旋转排序数组中找最小值
def find_min_rotated(arr):
    left, right = 0, len(arr) - 1
    while left < right:
        mid = left + (right - left) // 2
        if arr[mid] > arr[right]:
            left = mid + 1
        else:
            right = mid
    return arr[left]

# 在旋转数组中搜索
def search_rotated(arr, target):
    left, right = 0, len(arr) - 1
    while left <= right:
        mid = left + (right - left) // 2
        if arr[mid] == target:
            return mid
        # 判断哪一半有序
        if arr[left] <= arr[mid]:  # 左半有序
            if arr[left] <= target < arr[mid]:
                right = mid - 1
            else:
                left = mid + 1
        else:  # 右半有序
            if arr[mid] < target <= arr[right]:
                left = mid + 1
            else:
                right = mid - 1
    return -1

二分查找的精髓：不是只用于查找，而是用于求解单调性问题的最优解（如能力值验证）。

2. 哈希表（Hash Table）

2.1 核心概念

哈希表通过哈希函数将键映射到数组索引，实现 O(1) 的平均查找。

键(key) → 哈希函数 → 索引(index)
  "apple" → hash("apple") → 3
  "banana" → hash("banana") → 7

2.2 冲突解决

链地址法（Separate Chaining）

class HashTable:
    def __init__(self, capacity=16):
        self.capacity = capacity
        self.size = 0
        self.buckets = [[] for _ in range(capacity)]
        self.load_factor_threshold = 0.75

    def _hash(self, key):
        return hash(key) % self.capacity

    def put(self, key, value):
        idx = self._hash(key)
        bucket = self.buckets[idx]
        for i, (k, v) in enumerate(bucket):
            if k == key:
                bucket[i] = (key, value)  # 更新
                return
        bucket.append((key, value))
        self.size += 1
        if self.size / self.capacity > self.load_factor_threshold:
            self._resize()

    def get(self, key):
        idx = self._hash(key)
        for k, v in self.buckets[idx]:
            if k == key:
                return v
        raise KeyError(key)

    def delete(self, key):
        idx = self._hash(key)
        bucket = self.buckets[idx]
        for i, (k, v) in enumerate(bucket):
            if k == key:
                del bucket[i]
                self.size -= 1
                return v
        raise KeyError(key)

    def _resize(self):
        old_buckets = self.buckets
        self.capacity *= 2
        self.size = 0
        self.buckets = [[] for _ in range(self.capacity)]
        for bucket in old_buckets:
            for k, v in bucket:
                self.put(k, v)

开放寻址法（Open Addressing）

class OpenAddressingHash:
    """线性探测"""
    def __init__(self, capacity=16):
        self.capacity = capacity
        self.size = 0
        self.keys = [None] * capacity
        self.values = [None] * capacity
        self.DELETED = object()  # 哨兵标记

    def _probe(self, key, i):
        return (hash(key) + i) % self.capacity

    def put(self, key, value):
        for i in range(self.capacity):
            idx = self._probe(key, i)
            if self.keys[idx] in (None, self.DELETED):
                self.keys[idx] = key
                self.values[idx] = value
                self.size += 1
                return
            if self.keys[idx] == key:
                self.values[idx] = value
                return
        raise RuntimeError("哈希表已满")

    def get(self, key):
        for i in range(self.capacity):
            idx = self._probe(key, i)
            if self.keys[idx] is None:
                raise KeyError(key)
            if self.keys[idx] == key:
                return self.values[idx]
        raise KeyError(key)

2.3 哈希函数设计

方法说明适用
除留余数法hash(key) % m通用，m 取质数
乘法哈希floor(m × (A × key mod 1))均匀分布
MurmurHash非加密型，速度快、散列性好现代哈希表实现
SHA-256加密型，计算慢安全场景

3. 布隆过滤器（Bloom Filter）

3.1 原理

空间效率极高的概率型数据结构，用于判断"元素可能在集合中"或"一定不在集合中"。

位数组 m 个位 + k 个哈希函数

添加元素：用 k 个哈希函数计算 k 个位置，全部置 1
查询元素：k 个位置全为 1 → "可能存在"；任一位置为 0 → "一定不存在"

3.2 Python 实现

import hashlib
import math

class BloomFilter:
    def __init__(self, expected_items, false_positive_rate=0.01):
        """
        expected_items: 预期元素数量
        false_positive_rate: 可接受的误判率
        """
        self.size = self._optimal_size(expected_items, false_positive_rate)
        self.hash_count = self._optimal_hash_count(self.size, expected_items)
        self.bit_array = [0] * self.size
        self.item_count = 0

    def _optimal_size(self, n, p):
        return int(-n * math.log(p) / (math.log(2) ** 2))

    def _optimal_hash_count(self, m, n):
        return int(m / n * math.log(2))

    def _hashes(self, item):
        """生成 k 个哈希值（使用双哈希模拟）"""
        item = str(item).encode('utf-8')
        h1 = int(hashlib.md5(item).hexdigest(), 16)
        h2 = int(hashlib.sha256(item).hexdigest(), 16)
        for i in range(self.hash_count):
            yield (h1 + i * h2) % self.size

    def add(self, item):
        for idx in self._hashes(item):
            self.bit_array[idx] = 1
        self.item_count += 1

    def __contains__(self, item):
        return all(self.bit_array[idx] for idx in self._hashes(item))

# 示例
bf = BloomFilter(expected_items=100000, false_positive_rate=0.01)
for i in range(10000):
    bf.add(f"item_{i}")

print("item_5" in bf)       # True
print("item_99999" in bf)   # False
print("xxx" in bf)          # 大概率 False，小概率 True（误判）

3.3 应用场景

场景说明
URL 过滤器判断 URL 是否在黑名单中
缓存穿透防护Redis 前判断 key 是否可能有效
数据库查询优化先过滤肯定不存在的记录
爬虫去重海量 URL 去重

4. 跳表（Skip List）

4.1 原理

概率性平衡的有序数据结构，利用多层索引加速搜索。

Level 3:     1 ---------------------------→ 9
Level 2:     1 -----------→ 5 -----------→ 9
Level 1:     1 ---→ 3 ---→ 5 ---→ 7 ---→ 9
Level 0:  → 1 → 2 → 3 → 4 → 5 → 6 → 7 → 8 → 9  (原始有序链表)

搜索 6：
  Level 3: 1 → 9 (6 < 9, 下降)
  Level 2: 1 → 5 → 9 (6 < 9, 下降)
  Level 1: 5 → 7 (6 < 7, 下降)
  Level 0: 5 → 6 ✓

Redis 的 ZSet、Java ConcurrentSkipListMap 内部使用跳表。

4.2 Python 实现

import random

class SkipListNode:
    def __init__(self, val, level):
        self.val = val
        # forwards[i] 表示第 i 层的下一个节点
        self.forwards = [None] * level

class SkipList:
    MAX_LEVEL = 16
    P = 0.5  # 晋升概率

    def __init__(self):
        self.head = SkipListNode(None, self.MAX_LEVEL)
        self.level = 1

    def _random_level(self):
        level = 1
        while random.random() < self.P and level < self.MAX_LEVEL:
            level += 1
        return level

    def search(self, target):
        curr = self.head
        for i in range(self.level - 1, -1, -1):
            while curr.forwards[i] and curr.forwards[i].val < target:
                curr = curr.forwards[i]
        curr = curr.forwards[0]
        return curr is not None and curr.val == target

    def insert(self, val):
        update = [None] * self.MAX_LEVEL
        curr = self.head
        for i in range(self.level - 1, -1, -1):
            while curr.forwards[i] and curr.forwards[i].val < val:
                curr = curr.forwards[i]
            update[i] = curr

        new_level = self._random_level()
        if new_level > self.level:
            for i in range(self.level, new_level):
                update[i] = self.head
            self.level = new_level

        new_node = SkipListNode(val, new_level)
        for i in range(new_level):
            new_node.forwards[i] = update[i].forwards[i]
            update[i].forwards[i] = new_node

    def delete(self, val):
        update = [None] * self.MAX_LEVEL
        curr = self.head
        for i in range(self.level - 1, -1, -1):
            while curr.forwards[i] and curr.forwards[i].val < val:
                curr = curr.forwards[i]
            update[i] = curr

        target = curr.forwards[0]
        if not target or target.val != val:
            return False

        for i in range(self.level):
            if update[i].forwards[i] != target:
                break
            update[i].forwards[i] = target.forwards[i]

        while self.level > 1 and not self.head.forwards[self.level - 1]:
            self.level -= 1
        return True

4.3 跳表 vs 平衡树 vs 哈希表

特性跳表红黑树哈希表
有序性✅✅❌
范围查询✅ O(log n + k)✅ O(log n + k)❌
实现复杂度简单复杂中等
并发友好度高（锁粒度小）低中等
空间占用中等（~1.33n）中等高（预留空间）

5. 数据结构选择总结

查找场景：
  有序数据 + 频繁查找 → 二分查找 / 跳表
  无序数据 + 单次查找 → 哈希表
  海量数据 + 去重判断 → 布隆过滤器（允许误判）
  有序数据 + 范围查询 → 跳表 / B+ 树

参考文章

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「计算机基础」更多文章

  1. 16. 数据链路层
  2. 15. 网络层与路由
  3. 14. 网络模型与协议