Redis Sorted Set 跳表的实现原理和分析

跳表（Skip List）是一种随机化的数据布局，基于有序链表，通过在链表上增长多级索引来提高数据的查找服从。它是由 William Pugh 在 1990 年提出的。
  为什么 Redis 中的 Sorted Set 使用跳跃表

Redis 的有序聚集（Sorted Set）使用跳跃表（Skip List）作为底层实现，主要是因为跳跃表在性能、实现复杂度和灵活性等方面具有显著优势，可以替代平衡树的数据布局。
跳跃表的原理

跳跃表是一种扩展的有序链表，通过维护多个层级的索引来提高查找服从。每个节点包罗一个数据元素和一组指向其他节点的指针，这些指针分布在差别的层级。最底层的链表包罗全部元素，而每一层的节点数目逐渐淘汰。如许，查找操作可以从高层开始，快速跳过不需要的元素，淘汰遍历的节点数，从而提高查找服从。

• 查找过程：从最高层的头节点开始，沿着索引节点向右查找，如果当前节点的下一个节点的值大于或等于查找的目标值，则向下移动到下一层继续查找；否则，向右移动到下一个索引节点。这个过程一直持续到找到目标节点或到达最底层链表。
  
• 插入和删除操作：跳跃表支持高效的动态插入和删除，时间复杂度均为 O(log n)。插入时，首先找到插入位置，然后根据随机函数决定新节点的层数，最后在相应的层中插入节点。
  

跳跃表的优势

• 简单性：跳跃表的实现相对简单，易于理解和维护，而平衡树（如红黑树）的实现较为复杂，轻易出错。
  
• 高效的范围查询：跳跃表在进行范围查询时服从更高，因为它是有序的链表，可以直接遍历后继节点，而平衡树需要中序遍历，复杂度更高。
  
• 灵活性：跳跃表的层数可以根据需要动态调解，顺应差别的查询需求。
  
• 高并发性能：跳跃表的节点可以支持多个线程同时插入或删除节点，而平衡树和哈希表通常需要复杂的并发控制。
  
• 空间服从：跳跃表的空间复杂度为 O(n)，而且通过调解节点的抽取隔断，可以灵活平衡空间和时间复杂度。
  

正是因为有这些优势，Redis 选择使用跳跃表来实现有序聚集，而不是红黑树或其他数据布局。这使得 Redis 在处置惩罚有序聚集时能够高效地支持插入、删除和查找操作，同时保持元素的有序性，非常适合实现如排行榜、范围查询等功能。
为了讲明白跳表的原理，现在我们来模拟一个简单的跳表实现。
在 Java 中模拟实现 Redis 的 Sorted Set 跳表，我们需要定义跳表的数据布局，包括节点和跳表本身。以下是一个简单的实现：

1. import java.util.Random;
3. public class SkipList {
5.     private static final double P = 0.5; // 节点晋升的概率
6.     private static final int MAX_LEVEL = 16; // 最大层数
7.     private int levelCount; // 当前跳表的层数
8.     private Node head; // 头节点
9.     private int size; // 跳表中元素的数量
10.     private Random random; // 随机数生成器
12.     public SkipList() {
13.         this.levelCount = 0;
14.         this.size = 0;
15.         this.head = new Node(-1, Integer.MIN_VALUE, MAX_LEVEL);
16.         this.random = new Random();
17.     }
19.     // 节点定义
20.     private class Node {
21.         int value;
22.         int key;
23.         Node[] forward; // 向前指针数组
25.         Node(int value, int key, int level) {
26.             this.value = value;
27.             this.key = key;
28.             this.forward = new Node[level + 1];
29.         }
30.     }
32.     // 随机生成节点的层数
33.     private int randomLevel() {
34.         int level = 0;
35.         while (random.nextFloat() < P && level < MAX_LEVEL) {
36.             level++;
37.         }
38.         return level;
39.     }
41.     // 搜索指定值的节点
42.     public Node search(int key) {
43.         Node current = head;
44.         for (int i = levelCount - 1; i >= 0; i--) {
45.             while (current.forward[i] != null && current.forward[i].key < key) {
46.                 current = current.forward[i]; // 沿着当前层的指针移动
47.             }
48.         }
49.         current = current.forward[0]; // 移动到第0层
50.         return current;
51.     }
53.     // 插入节点
54.     public void insert(int key, int value) {
55.         Node[] update = new Node[MAX_LEVEL + 1];
56.         Node current = head;
58.         // 查找插入位置
59.         for (int i = levelCount - 1; i >= 0; i--) {
60.             while (current.forward[i] != null && current.forward[i].key < key) {
61.                 current = current.forward[i];
62.             }
63.             update[i] = current;
64.         }
66.         int level = randomLevel(); // 随机生成层数
67.         if (level > levelCount) {
68.             levelCount = level;
69.             for (int i = levelCount - 1; i >= 0; i--) {
70.                 update[i] = head;
71.             }
72.         }
74.         current = new Node(value, key, level);
75.         for (int i = 0; i < level; i++) {
76.             current.forward[i] = update[i].forward[i];
77.             update[i].forward[i] = current;
78.         }
79.         size++;
80.     }
82.     // 删除节点
83.     public void delete(int key) {
84.         Node[] update = new Node[MAX_LEVEL + 1];
85.         Node current = head;
87.         // 查找要删除的节点
88.         for (int i = levelCount - 1; i >= 0; i--) {
89.             while (current.forward[i] != null && current.forward[i].key < key) {
90.                 current = current.forward[i];
91.             }
92.             update[i] = current;
93.         }
95.         current = current.forward[0];
96.         if (current != null && current.key == key) {
97.             for (int i = 0; i < levelCount; i++) {
98.                 if (update[i].forward[i] != current) {
99.                     break;
100.                 }
101.                 update[i].forward[i] = current.forward[i];
102.             }
103.             size--;
104.             while (levelCount > 0 && head.forward[levelCount - 1] == null) {
105.                 levelCount--;
106.             }
107.         }
108.     }
110.     // 打印跳表
111.     public void printList() {
112.         Node current = head.forward[0];
113.         while (current != null) {
114.             System.out.println(current.key + ":" + current.value);
115.             current = current.forward[0];
116.         }
117.     }
119.     // 主函数测试跳表
120.     public static void main(String[] args) {
121.         SkipList list = new SkipList();
122.         list.insert(3, 100);
123.         list.insert(6, 300);
124.         list.insert(4, 400);
125.         list.insert(7, 700);
126.         list.insert(5, 500);
127.         list.printList();
128.         list.delete(3);
129.         list.printList();
130.     }
131. }

复制代码

下面，V 哥来表明一下：

• 跳表节点：每个节点包罗一个值、一个键和一个向前指针数组。向前指针数组存储了指向同一层下一个节点的引用。
  
• 随机层数：每个节点的层数是根据预设的概率 P 随机天生的，如许可以模拟出差别高度的索引层。
  
• 搜索操作：从最高层开始，沿着当前层的指针移动，直到找到插入点或到达底层。
  
• 插入操作：首先找到插入位置，然后根据随机天生的层数创建新节点，并更新每一层的指针。
  
• 删除操作：找到要删除的节点，然后逐层更新指针，最后淘汰跳表的层数。
  
• 打印跳表：从底层的头节点开始，遍历打印每个节点的键和值。
  

从这个简化版的跳表实现可以看到跳表的基本操作。可以帮助我们理解跳表实现的原理。
如安在 Java 中实现跳表的并发操作？

在 Java 中实现跳表的并发操作需要思量线程安全问题。可以通过以下方法来实现：

• 使用同步块：在每个公共方法中使用 synchronized 关键字来确保同一时间只有一个线程可以执行方法。
  
• 使用锁：使用 ReentrantLock 或其他并发锁来控制对跳表的并发访问。
  
• 使用原子引用：使用 AtomicReference 或 AtomicReferenceArray 来确保节点的原子更新。
  
• 使用并发聚集：使用 ConcurrentHashMap 等并发聚集作为辅助工具来实现线程安全的跳表。
  

以下是一个使用 synchronized 关键字实现线程安全的跳表的示例：

1. import java.util.Random;
3. public class ConcurrentSkipList {
5.     private static final double P = 0.5; // 节点晋升的概率
6.     private static final int MAX_LEVEL = 16; // 最大层数
7.     private int levelCount; // 当前跳表的层数
8.     private Node head; // 头节点
9.     private int size; // 跳表中元素的数量
10.     private Random random; // 随机数生成器
12.     public ConcurrentSkipList() {
13.         this.levelCount = 0;
14.         this.size = 0;
15.         this.head = new Node(-1, Integer.MIN_VALUE, MAX_LEVEL);
16.         this.random = new Random();
17.     }
19.     private class Node {
20.         int value;
21.         int key;
22.         Node[] forward; // 向前指针数组
24.         Node(int value, int key, int level) {
25.             this.value = value;
26.             this.key = key;
27.             this.forward = new Node[level + 1];
28.         }
29.     }
31.     private int randomLevel() {
32.         int level = 0;
33.         while (random.nextFloat() < P && level < MAX_LEVEL) {
34.             level++;
35.         }
36.         return level;
37.     }
39.     public synchronized Node search(int key) {
40.         Node current = head;
41.         for (int i = levelCount - 1; i >= 0; i--) {
42.             while (current.forward[i] != null && current.forward[i].key < key) {
43.                 current = current.forward[i];
44.             }
45.         }
46.         current = current.forward[0];
47.         return current;
48.     }
50.     public synchronized void insert(int key, int value) {
51.         Node[] update = new Node[MAX_LEVEL + 1];
52.         Node current = head;
54.         for (int i = levelCount - 1; i >= 0; i--) {
55.             while (current.forward[i] != null && current.forward[i].key < key) {
56.                 current = current.forward[i];
57.             }
58.             update[i] = current;
59.         }
61.         int level = randomLevel();
62.         if (level > levelCount) {
63.             for (int i = levelCount; i < level; i++) {
64.                 update[i] = head;
65.             }
66.             levelCount = level;
67.         }
69.         current = new Node(value, key, level);
70.         for (int i = 0; i < level; i++) {
71.             current.forward[i] = update[i].forward[i];
72.             update[i].forward[i] = current;
73.         }
74.         size++;
75.     }
77.     public synchronized void delete(int key) {
78.         Node[] update = new Node[MAX_LEVEL + 1];
79.         Node current = head;
81.         for (int i = levelCount - 1; i >= 0; i--) {
82.             while (current.forward[i] != null && current.forward[i].key < key) {
83.                 current = current.forward[i];
84.             }
85.             update[i] = current;
86.         }
88.         current = current.forward[0];
89.         if (current != null && current.key == key) {
90.             for (int i = 0; i < levelCount; i++) {
91.                 if (update[i].forward[i] != current) {
92.                     break;
93.                 }
94.                 update[i].forward[i] = current.forward[i];
95.             }
96.             size--;
97.             while (levelCount > 0 && head.forward[levelCount - 1] == null) {
98.                 levelCount--;
99.             }
100.         }
101.     }
103.     public synchronized void printList() {
104.         Node current = head.forward[0];
105.         while (current != null) {
106.             System.out.println(current.key + ":" + current.value);
107.             current = current.forward[0];
108.         }
109.     }
111.     public static void main(String[] args) {
112.         ConcurrentSkipList list = new ConcurrentSkipList();
113.         list.insert(3, 100);
114.         list.insert(6, 300);
115.         list.insert(4, 400);
116.         list.insert(7, 700);
117.         list.insert(5, 500);
118.         list.printList();
119.         list.delete(3);
120.         list.printList();
121.     }
122. }

复制代码

在这个示例中，我们使用了 synchronized 关键字来确保 search、insert 和 delete 方法是线程安全的。这会锁定当前对象，确保同一时间只有一个线程可以执行这些方法。
请注意，虽然 synchronized 可以提供线程安全，但它也可能导致性能瓶颈，特殊是在高并发环境下。在现实应用中，可以思量使用更细粒度的锁，如 ReentrantLock，大概使用原子引用和其他并发工具来提高性能。
使用 ReentrantLock 的实现方式

使用 ReentrantLock 实现跳表的并发操作可以提供比 synchronized 更细粒度的锁定，从而提高并发性能。ReentrantLock 允许您在差别的方法中锁定和解锁，乃至可以在差别的类中使用同一个锁对象。
以下是使用 ReentrantLock 实现线程安全的跳表的示例：

1. import java.util.Random;
2. import java.util.concurrent.locks.Lock;
3. import java.util.concurrent.locks.ReentrantLock;
5. public class ConcurrentSkipList {
7.     private static final double P = 0.5; // 节点晋升的概率
8.     private static final int MAX_LEVEL = 16; // 最大层数
9.     private final Lock lock = new ReentrantLock(); // 创建一个 ReentrantLock 对象
10.     private int levelCount; // 当前跳表的层数
11.     private Node head; // 头节点
12.     private int size; // 跳表中元素的数量
13.     private Random random; // 随机数生成器
15.     public ConcurrentSkipList() {
16.         this.levelCount = 0;
17.         this.size = 0;
18.         this.head = new Node(Integer.MIN_VALUE, MAX_LEVEL);
19.         this.random = new Random();
20.     }
22.     private class Node {
23.         int value;
24.         int key;
25.         Node[] forward; // 向前指针数组
27.         Node(int key, int level) {
28.             this.key = key;
29.             this.forward = new Node[level + 1];
30.         }
31.     }
33.     private int randomLevel() {
34.         int level = 0;
35.         while (random.nextFloat() < P && level < MAX_LEVEL) {
36.             level++;
37.         }
38.         return level;
39.     }
41.     public void search(int key) {
42.         lock.lock(); // 加锁
43.         try {
44.             Node current = head;
45.             for (int i = levelCount - 1; i >= 0; i--) {
46.                 while (current.forward[i] != null && current.forward[i].key < key) {
47.                     current = current.forward[i];
48.                 }
49.             }
50.             current = current.forward[0];
51.             // ... 处理找到的节点
52.         } finally {
53.             lock.unlock(); // 释放锁
54.         }
55.     }
57.     public void insert(int key, int value) {
58.         lock.lock(); // 加锁
59.         try {
60.             Node[] update = new Node[MAX_LEVEL + 1];
61.             Node current = head;
62.             for (int i = levelCount - 1; i >= 0; i--) {
63.                 while (current.forward[i] != null && current.forward[i].key < key) {
64.                     current = current.forward[i];
65.                 }
66.                 update[i] = current;
67.             }
69.             int level = randomLevel();
70.             if (level > levelCount) {
71.                 levelCount = level;
72.             }
74.             current = new Node(value, key, level);
75.             for (int i = 0; i < level; i++) {
76.                 current.forward[i] = update[i].forward[i];
77.                 update[i].forward[i] = current;
78.             }
79.             size++;
80.         } finally {
81.             lock.unlock(); // 释放锁
82.         }
83.     }
85.     public void delete(int key) {
86.         lock.lock(); // 加锁
87.         try {
88.             Node[] update = new Node[MAX_LEVEL + 1];
89.             Node current = head;
90.             for (int i = levelCount - 1; i >= 1; i--) {
91.                 while (current.forward[i] != null && current.forward[i].key < key) {
92.                     current = current.forward[i];
93.                 }
94.                 update[i] = current;
95.             }
97.             current = current.forward[0];
98.             if (current != null && current.key == key) {
99.                 for (int i = 0; i < levelCount; i++) {
100.                     if (update[i].forward[i] != current) {
101.                         break;
102.                     }
103.                     update[i].forward[i] = current.forward[i];
104.                 }
105.                 size--;
106.                 while (levelCount > 0 && head.forward[levelCount - 1] == null) {
107.                     levelCount--;
108.                 }
109.             }
110.         } finally {
111.             lock.unlock(); // 释放锁
112.         }
113.     }
115.     public void printList() {
116.         lock.lock(); // 加锁
117.         try {
118.             Node current = head.forward[0];
119.             while (current != null) {
120.                 System.out.println(current.key + ":" + current.value);
121.                 current = current.forward[0];
122.             }
123.         } finally {
124.             lock.unlock(); // 释放锁
125.         }
126.     }
128.     public static void main(String[] args) {
129.         ConcurrentSkipList list = new ConcurrentSkipList();
130.         list.insert(3, 100);
131.         list.insert(6, 300);
132.         list.insert(4, 400);
133.         list.insert(7, 700);
134.         list.insert(5, 500);
135.         list.printList();
136.         list.delete(3);
137.         list.printList();
138.     }
139. }

复制代码

在这个示例中，我们使用了 ReentrantLock 对象来控制对跳表的并发访问。每个公共方法在执行之前都会调用 lock.lock() 加锁，并在执行完毕后（包括正常执行和异常退出）调用 lock.unlock() 释放锁。
使用 ReentrantLock 的利益是它提供了比 synchronized 更灵活的锁定机制，例如：

• 可制止的锁定：ReentrantLock 允许线程在尝试获取锁时被制止。
  
• 尝试非阻塞锁定：ReentrantLock 提供了 tryLock() 方法，允许线程尝试获取锁而不无穷等待。
  
• 超时获取锁：ReentrantLock 还提供了 tryLock(long timeout, TimeUnit unit) 方法，允许线程在指定时间内等待锁。
  
• 公平锁：ReentrantLock 可以选择是否使用公平锁，公平锁会按照线程哀求锁的顺序来分配锁。
  
• 多个条件变量：ReentrantLock 可以与多个 Condition 对象配合使用，而 synchronized 只能与一个条件变量配合使用。
  

理解了以上代码实现原理后，我们再来理解 Redis Sorted Set 就不难了。
Redis 的 Sorted Set 是一种包罗元素和关联分数的数据布局，它能够根据分数对元素进行排序。Sorted Set 在 Redis 中的内部实现可以是跳跃表（Skip List）和字典（Hash Table）的组合，大概是压缩列表（Zip List），具体使用哪种实现取决于 Sorted Set 的巨细和元素的长度。
跳跃表 + 字典实现

当 Sorted Set 的元素数目较多大概元素长度较长时，Redis 使用跳跃表和字典来实现 Sorted Set。跳跃表是一种概率平衡的数据布局，它通过多级索引来提高搜索服从，类似于二分查找。字典则用于快速查找和更新操作。
跳跃表的每个节点包罗以下信息：

• 元素（member）
  
• 分数（score）
  
• 后退指针（backward）
  
• 多层前进指针（forward），每一层都是一个有序链表
  

字典则存储了元素到分数的映射，以便快速访问。
压缩列表实现

当 Sorted Set 的元素数目较少（默认小于128个），而且全部元素的长度都小于64字节时，Redis 使用压缩列表来存储 Sorted Set。压缩列表是一种连续内存块的顺序存储布局，它将全部的元素和分数紧凑地存储在一起，以节省内存空间。
应用场景

Sorted Set 常用于以下场景：

• 排行榜：例如游戏中的玩家分数排名。
  
• 范围查询：例如获取一定分数范围内的用户。
  
• 任务调度：例如按照任务的优先级执行。
  
• 实时排名：例如股票价格的实时排名。
  

代码分析

在 Redis 源码中，Sorted Set 的实现主要在 t_zset.c 文件中。插入操作（zaddCommand）会根据 Sorted Set 的编码类型（跳跃表或压缩列表）来执行差别的逻辑。如果是跳跃表编码，那么插入操作会涉及到字典的更新和跳跃表节点的添加。如果是压缩列表编码，则会检查是否需要转换为跳跃表编码。
总结

Sorted Set 是 Redis 提供的一种强盛的有序数据布局，它结合了跳跃表和字典的优点，提供了高效的插入、删除、更新和范围查询操作。通过合理的使用 Sorted Set，可以有用地办理许多现实问题。如何以上内容对你有帮助，恳请点赞转发，V 哥在此感谢各位兄弟的支持。88，洗洗睡了。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。