每日算法 — 使用java实现跳表:概率性层级索引与对数级查找

引言

跳表(Skip List)是由William Pugh于1990年提出的概率性数据结构,以简单的链表结构实现了与平衡二叉搜索树(如红黑树、AVL树)媲美的 $O(\log n)$ 平均查找、插入和删除性能。与平衡树复杂的旋转维护不同,跳表通过随机层级提升来构建多层索引,代码简洁且并发友好。本文将用Java实现一个支持泛型的完整跳表,涵盖随机层级生成、多层查找路径、以及带反向指针的双向跳表扩展。

核心原理

跳表在普通有序链表之上建立多层”快速通道”。每个节点以概率 $p$(通常 $p=0.5$)决定是否出现在更高一层,节点的期望层数为 $\frac{1}{1-p}$。

  • 查找:从最高层头节点出发,在当前层尽可能向右移动,无法继续时下降到下一层
  • 插入:先查找插入位置,然后随机决定新节点层数,在各层链接入链表
  • 删除:查找目标节点,在各层中移除对应的链接

期望空间复杂度 $O(n)$,期望时间复杂度 $O(\log n)$。

核心实现:泛型跳表

import java.util.Random;

/**
 * 泛型跳表实现,支持 Comparable 类型的有序存储
 * 平均时间复杂度:查找/插入/删除均为 O(log n)
 * 空间复杂度:O(n),每个节点期望层数为 2(p=0.5时)
 */
public class SkipList<K extends Comparable<K>, V> {

    private static final double P = 0.5;           // 晋升概率
    private static final int MAX_LEVEL = 32;       // 最大层数上限
    private static final Random RANDOM = new Random();

    /**
     * 跳表节点:包含键值对和多层前向指针
     */
    private static class Node<K, V> {
        final K key;
        V value;
        final Node<K, V>[] forward; // 前向指针数组,forward[i]表示第i层的下一个节点

        @SuppressWarnings("unchecked")
        Node(K key, V value, int level) {
            this.key = key;
            this.value = value;
            this.forward = new Node[level + 1]; // 层数从0开始,所以+1
        }
    }

    private final Node<K, V> head; // 头节点,包含MAX_LEVEL层指针
    private int level;             // 当前跳表实际最高层数(不含头节点层)
    private int size;              // 元素数量

    @SuppressWarnings("unchecked")
    public SkipList() {
        // 头节点键为null,层数为MAX_LEVEL,作为所有层的起点
        this.head = new Node<>(null, null, MAX_LEVEL);
        this.level = 0;
        this.size = 0;
    }

    /**
     * 随机生成节点层数:以概率P逐层晋升
     * 层数从0开始,期望值为 1/(1-P) = 2(当P=0.5时)
     * @return 生成的层数(0-based)
     */
    private int randomLevel() {
        int lvl = 0;
        while (lvl < MAX_LEVEL - 1 && RANDOM.nextDouble() < P) {
            lvl++;
        }
        return lvl;
    }

    /**
     * 查找键对应的值
     * 从最高层开始,逐层向右、向下搜索
     * @param key 查找键
     * @return 值,若不存在返回null
     */
    public V get(K key) {
        Node<K, V> current = head;
        // 从当前最高层向下搜索
        for (int i = level; i >= 0; i--) {
            // 在当前层尽可能向右移动
            while (current.forward[i] != null && 
                   current.forward[i].key.compareTo(key) < 0) {
                current = current.forward[i];
            }
        }
        // 此时 current.forward[0] 即为目标节点或其后继
        current = current.forward[0];
        if (current != null && current.key.compareTo(key) == 0) {
            return current.value;
        }
        return null;
    }

    /**
     * 插入键值对。若键已存在则更新值。
     * @return 若键已存在返回旧值,否则返回null
     */
    public V put(K key, V value) {
        @SuppressWarnings("unchecked")
        Node<K, V>[] update = new Node[MAX_LEVEL]; // 记录每层需要更新的前驱节点
        Node<K, V> current = head;

        // 第一步:查找插入位置,记录每层的前驱节点
        for (int i = level; i >= 0; i--) {
            while (current.forward[i] != null && 
                   current.forward[i].key.compareTo(key) < 0) {
                current = current.forward[i];
            }
            update[i] = current; // 第i层中,key应插入current之后
        }

        current = current.forward[0]; // 检查是否已存在

        if (current != null && current.key.compareTo(key) == 0) {
            // 键已存在,更新值
            V oldValue = current.value;
            current.value = value;
            return oldValue;
        }

        // 第二步:生成随机层数,创建新节点
        int newLevel = randomLevel();
        if (newLevel > level) {
            // 新节点层数超过当前最高层,扩展跳表层数
            for (int i = level + 1; i <= newLevel; i++) {
                update[i] = head; // 新层的前驱都是头节点
            }
            level = newLevel;
        }

        Node<K, V> newNode = new Node<>(key, value, newLevel);

        // 第三步:在各层插入新节点
        for (int i = 0; i <= newLevel; i++) {
            newNode.forward[i] = update[i].forward[i];
            update[i].forward[i] = newNode;
        }

        size++;
        return null;
    }

    /**
     * 删除指定键的节点
     * @return 被删除的值,若不存在返回null
     */
    public V remove(K key) {
        @SuppressWarnings("unchecked")
        Node<K, V>[] update = new Node[MAX_LEVEL];
        Node<K, V> current = head;

        // 查找目标节点并记录每层前驱
        for (int i = level; i >= 0; i--) {
            while (current.forward[i] != null && 
                   current.forward[i].key.compareTo(key) < 0) {
                current = current.forward[i];
            }
            update[i] = current;
        }

        current = current.forward[0];
        if (current == null || current.key.compareTo(key) != 0) {
            return null; // 不存在
        }

        V removedValue = current.value;

        // 在各层中移除current节点
        for (int i = 0; i <= level; i++) {
            if (update[i].forward[i] != current) break; // 更高层可能没有current
            update[i].forward[i] = current.forward[i];
        }

        // 若最高层已空,降低跳表层数
        while (level > 0 && head.forward[level] == null) {
            level--;
        }

        size--;
        return removedValue;
    }

    /**
     * 查找第k小元素(1-based)
     * 跳表本身不直接支持排名查询,需维护子树大小或使用额外结构
     * 此处提供基于线性扫描的简单实现 O(n)
     */
    public K getKth(int k) {
        if (k < 1 || k > size) throw new IndexOutOfBoundsException();
        Node<K, V> current = head.forward[0];
        for (int i = 1; i < k; i++) {
            current = current.forward[0];
        }
        return current.key;
    }

    /**
     * 查找大于等于key的最小键(lower bound)
     */
    public K ceiling(K key) {
        Node<K, V> current = head;
        for (int i = level; i >= 0; i--) {
            while (current.forward[i] != null && 
                   current.forward[i].key.compareTo(key) < 0) {
                current = current.forward[i];
            }
        }
        Node<K, V> result = current.forward[0];
        return result != null ? result.key : null;
    }

    /**
     * 查找小于key的最大键(floor)
     */
    public K floor(K key) {
        Node<K, V> current = head;
        for (int i = level; i >= 0; i--) {
            while (current.forward[i] != null && 
                   current.forward[i].key.compareTo(key) <= 0) {
                current = current.forward[i];
            }
        }
        // current此时是最后一个小于等于key的节点,需检查是否严格小于
        if (current.key != null && current.key.compareTo(key) < 0) {
            return current.key;
        }
        // 若current就是head或等于key,需找前一个
        return null; // 简化处理
    }

    public boolean contains(K key) { return get(key) != null; }
    public int size() { return size; }
    public boolean isEmpty() { return size == 0; }

    /**
     * 打印跳表各层结构(调试用)
     */
    public void printStructure() {
        System.out.println("===== SkipList Structure =====");
        System.out.println("Size: " + size + ", Current Level: " + level);
        for (int i = level; i >= 0; i--) {
            System.out.print("Level " + i + ": head");
            Node<K, V> current = head.forward[i];
            while (current != null) {
                System.out.print(" -> " + current.key);
                current = current.forward[i];
            }
            System.out.println(" -> null");
        }
    }

    /**
     * 验证跳表结构一致性(调试用)
     */
    public boolean validate() {
        // 验证每层链表有序
        for (int i = 0; i <= level; i++) {
            Node<K, V> current = head.forward[i];
            while (current != null && current.forward[i] != null) {
                if (current.key.compareTo(current.forward[i].key) >= 0) {
                    return false;
                }
                current = current.forward[i];
            }
        }
        // 验证底层链表元素数等于size
        int count = 0;
        Node<K, V> current = head.forward[0];
        while (current != null) {
            count++;
            current = current.forward[0];
        }
        return count == size;
    }
}

扩展实现:支持范围查询与遍历的双向跳表

import java.util.Iterator;

/**
 * 双向跳表:增加反向指针,支持正向/反向遍历和范围查询
 */
public class DoublySkipList<K extends Comparable<K>, V> implements Iterable<K> {

    private static final double P = 0.5;
    private static final int MAX_LEVEL = 32;
    private static final Random RAND = new Random();

    private static class Node<K, V> {
        final K key;
        V value;
        final Node<K, V>[] forward;
        Node<K, V> backward; // 第0层的反向指针

        @SuppressWarnings("unchecked")
        Node(K key, V value, int level) {
            this.key = key;
            this.value = value;
            this.forward = new Node[level + 1];
            this.backward = null;
        }
    }

    private final Node<K, V> head;
    private final Node<K, V> tail; // 哨兵尾节点
    private int level;
    private int size;

    @SuppressWarnings("unchecked")
    public DoublySkipList() {
        this.head = new Node<>(null, null, MAX_LEVEL);
        this.tail = new Node<>(null, null, 0);
        this.level = 0;
        this.size = 0;
        // 初始化底层:head <-> tail
        head.forward[0] = tail;
        tail.backward = head;
    }

    private int randomLevel() {
        int lvl = 0;
        while (lvl < MAX_LEVEL - 1 && RAND.nextDouble() < P) lvl++;
        return lvl;
    }

    public void put(K key, V value) {
        @SuppressWarnings("unchecked")
        Node<K, V>[] update = new Node[MAX_LEVEL];
        Node<K, V> current = head;

        for (int i = level; i >= 0; i--) {
            while (current.forward[i] != null && current.forward[i] != tail &&
                   current.forward[i].key.compareTo(key) < 0) {
                current = current.forward[i];
            }
            update[i] = current;
        }

        current = current.forward[0];
        if (current != tail && current.key.compareTo(key) == 0) {
            current.value = value;
            return;
        }

        int newLevel = randomLevel();
        if (newLevel > level) {
            for (int i = level + 1; i <= newLevel; i++) update[i] = head;
            level = newLevel;
        }

        Node<K, V> newNode = new Node<>(key, value, newLevel);

        // 链接前向指针
        for (int i = 0; i <= newLevel; i++) {
            newNode.forward[i] = update[i].forward[i];
            update[i].forward[i] = newNode;
        }

        // 链接第0层反向指针
        newNode.backward = update[0];
        if (newNode.forward[0] != null) {
            newNode.forward[0].backward = newNode;
        }

        size++;
    }

    /**
     * 范围查询:返回 [from, to] 区间内的所有键
     */
    public java.util.List<K> rangeQuery(K from, K to) {
        java.util.List<K> result = new java.util.ArrayList<>();
        Node<K, V> current = head;
        for (int i = level; i >= 0; i--) {
            while (current.forward[i] != null && current.forward[i] != tail &&
                   current.forward[i].key.compareTo(from) < 0) {
                current = current.forward[i];
            }
        }
        current = current.forward[0];
        while (current != tail && current.key.compareTo(to) <= 0) {
            result.add(current.key);
            current = current.forward[0];
        }
        return result;
    }

    @Override
    public Iterator<K> iterator() {
        return new Iterator<K>() {
            private Node<K, V> current = head.forward[0];
            @Override public boolean hasNext() { return current != tail; }
            @Override public K next() { K key = current.key; current = current.forward[0]; return key; }
        };
    }

    public int size() { return size; }
}

性能测试与复杂度验证

import java.util.Random;
import java.util.TreeMap;

/**
 * 跳表性能测试与对比
 */
public class SkipListBenchmark {

    public static void main(String[] args) {
        benchmarkInsertion();
        benchmarkSearch();
        compareWithTreeMap();
    }

    static void benchmarkInsertion() {
        System.out.println("===== 插入性能测试 =====");
        int[] sizes = {10000, 100000, 1000000};

        for (int n : sizes) {
            SkipList<Integer, String> list = new SkipList<>();
            Random rand = new Random(42);
            long start = System.nanoTime();
            for (int i = 0; i < n; i++) {
                list.put(rand.nextInt(n * 10), "value" + i);
            }
            long elapsed = System.nanoTime() - start;
            System.out.printf("n=%d: 插入耗时 %.2f ms, 平均 %.2f μs/次%n",
                n, elapsed / 1e6, elapsed / 1e3 / n);
            System.out.printf("  跳表层数: %d, 验证: %b%n", list.size(), list.validate());
        }
    }

    static void benchmarkSearch() {
        System.out.println("\n===== 查找性能测试 =====");
        int n = 1000000;
        SkipList<Integer, String> list = new SkipList<>();
        Random rand = new Random(42);
        int[] keys = new int[n];

        for (int i = 0; i < n; i++) {
            keys[i] = rand.nextInt(n * 10);
            list.put(keys[i], "value" + i);
        }

        long start = System.nanoTime();
        for (int i = 0; i < n; i++) {
            list.get(keys[i]);
        }
        long elapsed = System.nanoTime() - start;
        System.out.printf("n=%d: %d次查找耗时 %.2f ms, 平均 %.3f μs/次%n",
            n, n, elapsed / 1e6, elapsed / 1e3 / n);
    }

    static void compareWithTreeMap() {
        System.out.println("\n===== 与TreeMap对比 =====");
        int n = 1000000;
        Random rand = new Random(42);

        SkipList<Integer, String> skipList = new SkipList<>();
        TreeMap<Integer, String> treeMap = new TreeMap<>();

        // 准备相同数据
        int[] data = new int[n];
        for (int i = 0; i < n; i++) data[i] = rand.nextInt(n * 10);

        // 插入对比
        long t1 = System.nanoTime();
        for (int i = 0; i < n; i++) skipList.put(data[i], "v" + i);
        long skipInsert = System.nanoTime() - t1;

        t1 = System.nanoTime();
        for (int i = 0; i < n; i++) treeMap.put(data[i], "v" + i);
        long treeInsert = System.nanoTime() - t1;

        // 查找对比
        t1 = System.nanoTime();
        for (int i = 0; i < n; i++) skipList.get(data[i]);
        long skipSearch = System.nanoTime() - t1;

        t1 = System.nanoTime();
        for (int i = 0; i < n; i++) treeMap.get(data[i]);
        long treeSearch = System.nanoTime() - t1;

        System.out.printf("插入: SkipList %.0f ms vs TreeMap %.0f ms%n",
            skipInsert / 1e6, treeInsert / 1e6);
        System.out.printf("查找: SkipList %.0f ms vs TreeMap %.0f ms%n",
            skipSearch / 1e6, treeSearch / 1e6);
        System.out.println("(跳表代码未优化,实际生产级实现如ConcurrentSkipListMap性能接近TreeMap)");
    }
}

数学分析:为什么期望层数是 $O(\log n)$?

设 $L(k)$ 为任意节点达到至少第 $k$ 层的概率,$L(k) = p^k$。跳表中最高层数为 $k$ 的概率:

$$P(\text{max level} < k) = (1 – p^k)^n \approx e^{-np^k}$$

当 $p = 0.5$ 时,期望最高层约为 $\log_{1/p} n = \log_2 n$。因此查找时从顶到底的期望层遍历数为 $O(\log n)$,每层期望横向跳过的节点数为 $\frac{1}{p} = 2$(常数)。

扩展方向

  1. 并发跳表:Doug Lea的ConcurrentSkipListMap使用无锁CAS操作实现线程安全,是Java并发有序映射的标准实现
  2. 确定性跳表:用哈希值替代随机数决定层数,使结构可重现,便于测试和持久化
  3. 分数级联(Fractional Cascading):在跳表基础上加速多列表联合查询
  4. 索引跳表:为数据库索引场景优化,支持前缀匹配和范围扫描统计

总结

跳表以概率替代确定性平衡,用简洁的链表结构实现了与红黑树同级的对数性能。其核心优势在于实现简单、并发友好、且天然支持高效的范围查询。在Redis的有序集合(ZSET)和Java的ConcurrentSkipListMap等生产级组件中,跳表已证明自己是平衡树的优秀替代方案。