计算机基础

06. 搜索与哈希

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

spcent 6 分钟阅读 2695 字

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）中等高（预留空间）

6. 一致性哈希与分布式系统

6.1 传统哈希的问题

# 经典哈希：server = hash(key) % N
# 当服务器数量 N 变化时，几乎全部 key 都会重新映射
# 原来在 server0 的 key，加一台机器后可能跑到 server1/server2

6.2 一致性哈希原理

将服务器和 key 映射到同一个环形哈希空间（0 ~ 2³²-1），每个 key 顺时针找到最近的 server：

   0            2^32-1
   |              |
   v              v
   +--- serverA --+
   |              |
   key1  → 顺时针方向  → serverB
   |              |
   +--- serverB --+
   |              |
   +--- serverC --+

虚拟节点：每个物理节点对应多个虚拟节点，解决数据倾斜问题。

import hashlib
import bisect

class ConsistentHash:
    def __init__(self, replicas=150):
        self.replicas = replicas  # 每个物理节点虚拟节点数
        self.ring = []            # 排序后的哈希环
        self.nodes = {}           # hash → 物理节点
        self.sorted_keys = []
    
    def _hash(self, key):
        return int(hashlib.md5(key.encode()).hexdigest(), 16)
    
    def add_node(self, node):
        for i in range(self.replicas):
            key = f"{node}#{i}"
            h = self._hash(key)
            self.nodes[h] = node
            bisect.insort(self.sorted_keys, h)
    
    def remove_node(self, node):
        for i in range(self.replicas):
            h = self._hash(f"{node}#{i}")
            self.sorted_keys.remove(h)
            del self.nodes[h]
    
    def get_node(self, key):
        if not self.sorted_keys:
            return None
        h = self._hash(key)
        idx = bisect.bisect_right(self.sorted_keys, h)
        if idx == len(self.sorted_keys):
            idx = 0
        return self.nodes[self.sorted_keys[idx]]

# 使用
ch = ConsistentHash(replicas=150)
ch.add_node("server1")
ch.add_node("server2")
ch.add_node("server3")

# 查看 key 被路由到哪个节点
print(ch.get_node("user:10001"))   # server2
print(ch.get_node("user:10002"))   # server1
场景变化影响
增加 1 台服务器 (N→N+1)约 1/(N+1) 的数据需要迁移
减少 1 台服务器 (N→N-1)约 1/N 的数据需要迁移
传统哈希 N→N+1几乎全部数据需要迁移

应用场景：分布式缓存分片（Redis Cluster）、分布式存储、负载均衡。

7. 搜索算法进阶

7.1 插值搜索

对均匀分布的有序数组，插值搜索利用值的分布预估位置，平均时间接近 O(log log n)：

def interpolation_search(arr, target):
    left, right = 0, len(arr) - 1
    while left <= right and arr[left] <= target <= arr[right]:
        # 根据目标值在范围内的比例估算位置
        pos = left + int((right - left) * (target - arr[left]) / (arr[right] - arr[left]))
        if arr[pos] == target:
            return pos
        elif arr[pos] < target:
            left = pos + 1
        else:
            right = pos - 1
    return -1

7.2 指数搜索（无界数组搜索）

当数组长度未知时，先以指数步长确定范围，再二分搜索：

def exponential_search(arr, target):
    bound = 1
    while bound < len(arr) and arr[bound] < target:
        bound *= 2
    # 在 [bound/2, min(bound, len-1)] 范围内二分
    left, right = bound // 2, min(bound, len(arr) - 1)
    while left <= right:
        mid = (left + right) // 2
        if arr[mid] == target:
            return mid
        elif arr[mid] < target:
            left = mid + 1
        else:
            right = mid - 1
    return -1

数据结构选择总结

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

参考文章

继续阅读

探索更多技术文章

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

全部文章 返回首页

「计算机基础」更多文章

  1. 46. 排队论与容量估算：利特尔法则与尾延迟
  2. 45. 编译器优化与中间表示：SSA、内联与循环优化
  3. 44. 并发模型对比：Actor、CSP 与数据并行