01｜动态数组：按需分配的vector为什么要二倍扩容？

作者: 黄清昊

你好，我是微扰君。今天我们进入第一章基础数据结构的学习。

计算机程序一直以来最根本的作用就是处理数据。即使在早期的计算机中，计算就已经不仅仅是几个数字之间的加减乘除那么简单了，经常需要处理大量线性存储的数据，一个很好的例子就是向量乘法。显然，我们需要找到一种合适的方式在计算机中存储这些信息，并能让我们可以快速地进行向量运算。

再举一个更工程化的例子。假设有个需求，我们希望只借助内存实现一个简易的银行账户管理系统，每个账号里包括两个基本信息：账户ID及余额。用户首次开户的时候，会被分配一个账户ID；系统要支持用户通过ID快速查询余额，也可以存款/取款改变自己的余额。

你可能会觉得这有什么难的？用数组就可以解决。建立一个整型动态数组，每来一个用户就给存到数组的某个位置，用该位置的数组下标来当用户的ID就行。查询起来更快，数组大小是动态的，也不用考虑用户数量超过容量上限的问题。

但是，基于下标随机访问数组元素为什么这么高效？动态数组是怎么做到看起来可以有无限容量？扩容机制的时间复杂度是多少，是不是会带来额外的内存浪费呢？不知道你有没有思考过这些问题。

今天，我们就带着这些问题一起学习第一种序列式容器vector，也就是动态数组。相信你学完之后，对这些问题的理解就深刻啦。

数组和内存

讲解动态数组的实现之前，首先要回顾一下数组是什么，不过为了和动态数组区分开来，我们常常也称之为静态数组。可以这样定义：静态数组是由相同类型的元素线性排列的数据结构，在计算机上会分配一段连续的内存，对元素进行顺序存储。

其中有三个关键词，相同类型、连续内存、顺序存储。之所以这样设计，本质就是为了能做到基于下标，对数组进行O(1)时间复杂度的快速随机访问。

存储数组时，会事先分配一段连续的内存空间，将数组元素依次存入内存。因为数组元素的类型都是一样的，所以每个元素占用的空间大小也是一样的，这样我们就很容易用“数组的开始地址+index*元素大小”的计算方式，快速定位到指定索引位置的元素，这也是数组基于下标随机访问的复杂度为O(1)的原因。

为什么要事先分配一段内存呢？答案也很简单，因为内存空间并不是无限的。一段程序里可能有很多地方都需要分配内存，我们必然要为分配的连续内存寻找一个边界。

事先确定数组大小并分配相应的内存给数组，就是告诉程序，这块地方已经是某个数组的地盘了，就不要再来使用了。同样，访问该数组的时候，下标也不应该超过地盘的范围，在大部分语言里这样的非法操作都会引起越界的错误，但在一些没有越界保护实现的语言（比如C语言）中，这就是一个很大的问题，需要开发者非常谨慎。有时这甚至会成为软件被黑客攻击的漏洞。

静态数组的特性

当然，在内存里这样的顺序存储也不是没有代价，这直接导致了数组的插入和删除会低效很多，平均的复杂度是O(N)。因为数组，和集合不同，元素在数组中的位置是我们关心的。

在长度为N的数组中，要在下标为T的位置插入数据时，原数组中下标为T到N-1的元素都需要向后顺移一位，这需要遍历数组中共计N-T个元素，当然，如果希望插入到数组的末端，只需要做插入而不需要做任何移动操作。但同样，如果我们希望将新元素插入到数组最开始的位置，就要将原数组所有元素都向后移动了，这需要移动共计N个元素。

所以平均而言，数组的插入操作的时间复杂度为O(N)。删除操作基于类似的原因，复杂度当然也是O(N)。

总的来说，静态数组的特性就是数组中元素的个数是事先确定的，每个元素都有对应的索引，第一个元素的索引为0。因为每个元素在内存占用的空间是一样的，我们可以基于首元素的地址和目标元素的下标，直接定位到目标元素的位置。

动态数组

很显然，使用静态数组的时候需要事先指定空间大小，这并不是很让人们满意。因为静态数组的使用者分配完内存之后，数组空间就不再能扩展（或收缩）了。比如在开头的简易银行系统中，确定固定的数组大小带来的风险就是：当用户数不断上涨直至超过数组容量范围时，我们的系统就不能继续工作了。

唯一的解决方案只能是重新申请一个更大的数组。这个过程，如果自己手动实现，有相当多琐碎的操作，至少包括配置一整块更大的连续内存空间、将元素逐一拷贝至新空间，以及释放原本的空间。

而动态数组的意义就在于将这些繁琐的细节封装起来，给用户良好使用体验的同时，也兼顾效率。这就是为什么我们在大部分业务开发场景下，更多地采用动态数组容器，而不是原生的静态数组。

STL的Vector就是这样一种经过严格测试和实战检验的动态数组容器，我们下面来分析一下它的原理和实现。其他高级语言的动态数组容器的实现思路其实也是类似的，比如Java中的ArrayList等（后续我们也会用Java中的实现来讲解）。你搞清楚一个，其他的就很好理解了。

动态数组源码分析

当然，STL的源码涉及了许多高深的C++技巧，我们并不会展开讨论，会对源码做一些简单的调整方便你理解原理。这里也给你一个看源码的小建议，不要死抠细节。我个人看源码比较喜欢自顶而下的方式，先从大的模块暴露的方法和接口看起，而不是上来就开始研究小模块的实现细节。

比如学习STL源码中vector实现的时候，你会经常发现allocator相关的方法，如果你揪着它不放一路溯源，会发现allocator的底层实现也非常复杂，但是，大多数时候我们不需要这么做，只要理解清楚了allocator的哪些方法用来申请内存、哪些方法用于释放内存，具体实现细节暂时当作黑箱，把精力集中在当下要搞清楚的问题上就可以了。

首先，来看vector在内存中的表示。它有两个指标，大小和容量。

大小，表示现在存了多少数据。存数据的部分其实和静态数组是一样的，都是一段连续的内存空间顺序排列这若干类型相同、大小一致的元素，但不同的地方在于，数组的大小是可以动态调整的。

我们知道，向计算机申请空间连续的内存空间是一个成本比较高的操作，不只需要扫描出堆区内存的空闲内存块，可能还需要向操作系统申请更大的堆空间，并产生用户态-内核态的切换成本。

所以为了减少二次分配的次数，初次配置空间的时候，可以分配比vector目前所需空间更多一些，后续的若干次插入就不再需要触发昂贵的扩容操作了。这样的可用空间，我们称为vector的容量，是vector在创建时需要的第二个可选参数。

所以我们可以用三个指针来标记vector空间的使用情况，分别是：

1. _start 指针，指向vector第一个元素
2. _finish 指针，指向vector最后一个元素
3. _end 指针，指向vector预留容量的边界

当然，动态数组的两个核心指标就很容易计算出来了：

1. 容量，capacity = _end - _first，表示目前的数组最多能存储多少个元素
2. 大小，size = _finish - _first，表示数组当前已经存储的元素个数
   

对应的代码如下：

1
2
3
4
5
6
7
8
9
10

template <class _Tp, class _Alloc = __STL_DEFAULT_ALLOCATOR(_Tp) >
class vector : protected _Vector_base<_Tp, _Alloc> 
{
       ...
protected:
  _Tp* _M_start; //表示目前使用空间的头
  _Tp* _M_finish; //表示目前使用空间的尾
  _Tp* _M_end_of_storage; //表示目前可用空间的尾  
    ...
};

你不用太关注模版相关的语法，只需要知道这里protected的三个变量就是前面提到的3个指针。

有了这些，我们就可以判断什么时候需要触发扩容操作，以及扩容的方式。因为vector创建的时候会给一个容量，但随着我们不断往数组中插入元素，数组的大小终究会超过当前分配的容量，于是需要重新分配更大的内存，那具体分配多少是一个比较合理的值呢？

STL的扩容方法

来看一下STL怎么做的。除了查询已有资料之外，我个人比较推荐动手实验，不仅能随时检验自己脑海里的想法，通过动手实践对原理的理解和印象也更深刻一些。所以我们来编写一些测试方法，观察Vector在测试过程中的行为，再和官方文档及资料进行对比验证。

要做的实验也很简单，就是往一个数组里不断的插入元素，并观察size和capacity的变化。完整的代码可以在这里找到。

1
2
3
4
5
6

vector<int> v;

for (int i = 0; i < 20; i++) {
    cout << "size: " << v.size() << " capacity " << v.capacity() << endl;
    v.push_back(i);
}

通过实验我们能发现一个很明显的规律：如果每次只插入一个元素，当vector的大小小于容量时，容量不会发生变化，数组大小不断递增。而当vector的大小即将超过容量的时候，插入之后，容量大小会翻番。

所以，倍增就是vector的扩容方式，这种类似倍增的策略也会出现在许多其他使用场景中。

这里很自然会有一个问题，为什么每次扩容时候都是以倍增的方式扩容，而不是增加固定大小的容量呢？

在回答这个问题之前，我们先看一看STL扩容逻辑的实现。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

void push_back(const _Tp& __x) {//在最尾端插入元素
    if (_M_finish != _M_end_of_storage) {//若有可用的内存空间
      construct(_M_finish, __x);//构造对象
      ++_M_finish;
    }
    else//若没有可用的内存空间,调用以下函数，把x插入到指定位置
      _M_insert_aux(end(), __x);
  }
  
template <class _Tp, class _Alloc>
	void 
	vector<_Tp, _Alloc>::_M_insert_aux(iterator __position, const _Tp& __x)
	{
	  if (_M_finish != _M_end_of_storage) {
	    construct(_M_finish, *(_M_finish - 1));
	    ++_M_finish;
	    _Tp __x_copy = __x;
	    copy_backward(__position, _M_finish - 2, _M_finish - 1);
	    *__position = __x_copy;
	  }
	  else {
	    const size_type __old_size = size();
	    const size_type __len = __old_size != 0 ? 2 * __old_size : 1;
	    iterator __new_start = _M_allocate(__len);
	    iterator __new_finish = __new_start;
	    __STL_TRY {
	      __new_finish = uninitialized_copy(_M_start, __position, __new_start);
	      construct(__new_finish, __x);
	      ++__new_finish;
	      __new_finish = uninitialized_copy(__position, _M_finish, __new_finish);
	    }
	    __STL_UNWIND((destroy(__new_start,__new_finish), 
	                  _M_deallocate(__new_start,__len)));
	    destroy(begin(), end());
	    _M_deallocate(_M_start, _M_end_of_storage - _M_start);
	    _M_start = __new_start;
	    _M_finish = __new_finish;
	    _M_end_of_storage = __new_start + __len;
	  }
	}

这段扩容操作在push_back和insert操作中都会触发，我们以简单一点的push_back，也就是往数组尾部插入元素的操作为例，来解释扩容的逻辑。

可以看到，push_back的时候会先做一个判断，看当前的容量是不是不够用了。如果够用，我们只要直接往后插入一个元素；不够用，才进行_M_insert_aux扩容并插入的操作，插入后需要把finish指针往后移动。这里在容量够用的时候，插入逻辑用的是construct函数，是STL容器中通用的构造方法。

我们来重点分析扩容逻辑所在的_M_insert_aux方法：

• 13-20行，实际上是因为还有其他函数会调用这个方法，我们已经确定容量不足，所以不会进入这段逻辑。
• 22-25行，主要做的事情就是读取原有的vector大小old_size，再从内存里申请一段新的空间，大小为2*old_size，创建新的首尾指针并指向新的空间。
• 26-31行，将老空间里的数据逐一搬到新的空间里，并在最后添加新的元素。这样就完成了扩容的主要目的，这是一个O(n)复杂度的操作，因为你需要对原数组进行逐一的深拷贝。
• 最后，在32-38行，我们需要做一些清理和收尾工作，释放掉老的数组空间和指针，将容器的首尾及容量指针都更新到对应的位置。

这样Vector就完成了对扩容操作的封装，是不是其实并不复杂呢？

现在清楚扩容的具体实现之后，来解答前面的问题：为什么扩容是采用倍增的方式，而不是每次扩展固定大小？这背后其实是有严密数学依据的，非常有趣，我们一起来探索一下。

用极端法来考虑这个问题。

先假设是不是可以不倍增，而是每次只扩展一个元素呢？直觉上这当然是不合理的，这会导致每一次插入都会触发扩容，而每次扩容都会进行所有元素的复制操作。所以如果我们要插入n个元素，需要进行的拷贝次数：

$1+2+3+… +n=n^{2}$

复杂度为O(n^2)，均摊下来每次操作时间复杂度就是O(n)。

那如果我们不是每次只拓展一个元素，而是每次扩展C的容量，对复杂度的计算会产生多大的影响呢？同样来计算一下，每插入C次就需要进行一次扩展操作，每次扩展仍然需要复制全部元素，所以总的拷贝次数是：

$C + 2C + 3C + … + ceiling(n/C)*C$

复杂度同样为O(n^2) 。均摊下来每次操作时间复杂度还是O(n)。 虽然次数少了C倍，但仍然不令人满意。

更好的做法就是和STL一样采用倍增的思想，每次都将容量扩展为当前的一倍，它往往能让我们的时间复杂度下降很多。

算一下倍增这种策略下需要拷贝的次数：假设一共还是插入N次，那总拷贝次数，就是从1加到2的X次，其中x是logn向上取整；这是因为容量每次都在翻番，所以每次触发拷贝的时候，容量分别是1、2、4、8 … 一直到logn向上取整。

$1+2+4+8+… +2^{x}=2^{(x+1)}-1$

这样插入N个元素的复杂度就一下减少为O(N)了。均摊到每次插入的扩容复杂度就为O(1)，这当然是一个令人满意的结果啦。

总结

相信经过今天的学习，你一定已经对开头的几个问题有答案了吧，简单总结一下。数组，是支持O(1)基于下标随机访问的数据结构，在内存中是连续存储的。基于下标高效访问元素的核心就在于“相同类型”和“连续存储”的特性，当然，也带来了高昂的插入和删除的时间复杂度。

动态数组之所以能看起来像是无限容量，也仅仅是因为它内置了倍增的扩容策略，每次数组大小超过容量的时候，就会触发数组的扩容机制，封装了繁琐的拷贝细节。

也正是因为上述特性，动态数组广泛应用在需要经常查询、变更，但是很少插入/删除的场景，比如我们在实现一个简单的Web服务器的时候，可以用vector来存储handler的线程，达到线程复用的效果。

你应该感受到了今天的内容比上一讲的文本差分要简单一些，是的，接下来我们会先把数据结构实现的基础打好，了解清楚背后的实现原理，这样在日常开发中，不同的数据结构可能造成什么样的性能瓶颈，你都能烂熟于心。

课后作业

我们已经细致地分析了在vector中插入元素的方法，如果要删除一个元素应该怎么实现呢？在删除元素的时候需不需要缩容呢？如果需要的话，你会怎么做。

这是一个开放问题，欢迎你在留言区与我讨论。我们下节课见。

02｜双向链表：list如何实现高效地插入与删除？

作者: 黄清昊

你好，我是微扰君。

在上一讲实现的一个简易银行账户管理系统中，每个账号都对应了一个余额，系统支持用户的开通、存/取款和查询余额。我们使用动态数组容器满足了频繁随机访问查询的需求。

但是如果要在系统里支持删除的功能，就会有一个问题：我们为了不进行整体的数组移动操作，通常就只能保留这个用户在数组里占用的内存，用将元素标记为特殊值的方式来模拟“删除”；而因为数组是连续存储的，不能单独释放掉数组中间某些区域的内存，所以这段内存空间我们实际上就是浪费的。

如果还有个需求，比如现在某个不讲道理的领导来到这个银行，要求自己在数组中排在最前面；那么我们不得不将所有人的账户信息往后挪动一位来满足他奇怪的自尊心，这也会带来高昂的时间复杂度。

那么有没有办法让我们不再需要连续的存储空间去存储一个序列，同时又可以在序列中快速进行插入/删除操作而不用波及之后的所有元素呢？

这就需要另一种常见的序列式数据结构——链表登场了，这同样是几乎所有高级语言都会原生支持的数据结构。

链表

链表这个数据结构的发明也是很久之前的事了，最早在1955年，它就是IPL这一古老语言的内置数据结构，用于开发当时的人工智能程序。

类似数组，链表同样是一种序列式的数据结构，但存储元素时并不需要使用连续的内存空间，而是采用一系列通过指针相连的节点来存储，因为有了指针来关联节点的地址，就不需要连续存储了。

在每个节点内，我们都会同时存储元素的数据信息和一个指针，存储元素信息的部分是数据域，也就是 data field，存储指针的地方称为引用域，也就是 reference field；其中指针指向该节点的后继节点，也就是记录着链表中存储下一个元素节点的地址。

下图就是一个典型长度为3的链表的示例，我们可以通过指针很容易地从第一个节点遍历完整个链表。

从内存布局能看出来，链表比连续存储的数组，有更灵活的内存使用方式和更高的内存使用率。

因为数组要求事先分配内存，而链表是每次插入新节点的时候，才申请该节点所需的内存空间，灵活得多，也就不会有分配空间没有被使用的浪费问题，自然内存使用率高。

链表存储元素采用的是通过指针的串联方式，而非数组的连续排列方式。我们在任何位置插入或者删除节点，不再需要为了保持元素的连续存储而进行O(n)的整体移动操作，只需要进行O(1)的指针改写，加申请或者释放内存就行。这显然比数组的插入删除要高效得多。

但是毕竟鱼和熊掌不能兼得。也正是因为这样非连续的存储方式，我们需要访问链表中第n个元素的时候就不得不从头节点遍历，导致访问第i个元素的均摊时间复杂度为O(N)，而不能像数组那样直接基于下标和元素大小，计算出指定元素的偏移量。

所以，我们并不能简单地说链表和数组哪个更好，而是要根据使用的场景做出合适的选择。毕竟如果两个相似的数据结构其中一个各个方面都好于另一个的话，另一个数据结构可能也不会存在至今了。

基于上面的特性对比，我们可以得出一个大致的结论：链表更适用于删除、插入、遍历操作频繁的场景，而不适用于随机访问索引频繁的场景。比如在内存池、操作系统进程管理、最常用的缓存替换算法LRU中都有应用，之后讲解到相关专题的时候也会提到。

单链表vs双链表vs循环链表

在实际使用过程中，根据不同的需求，大致有3种常见的链表形式，单链表、双链表和循环链表，它们都需要支持几种最基本的链表操作，包括插入节点、删除节点、修改节点信息，以及访问遍历节点信息。

可以看图直观感受三者的区别。三种链表节点都包含引用域和数据域。

其中，单链表和双链表最大的区别就在于，单链表的引用域，只存了一个后继指针。

而双链表有两个引用域，不只存有后继节点的地址，也存储了前驱节点的信息，这使得我们可以双向遍历链表，拥有了在遍历中回退的能力。在链表首尾的前驱和后继指针，可以设为NULL或者指向一个特殊的虚拟节点，来标记链表的终结。

而循环链表则是一个没有边界的环，既可以用双链表来实现也可以用单链表实现。相比于前两者的主要区别在于，在链表的边界，比如尾部，不再设为NULL或者一个虚拟节点，而是直接将尾部指向链表的头部。这在许多需要循环遍历的场景下非常有用，比如可以用于模拟约瑟夫环。

STL中List的实现

好了，讲解完链表的基础概念和分类，进入今天的重头戏，我们看看如何实现一个链表。

为了契合专栏真实世界的语义，我们继续以STL这一久经考验的C++标准库中的实现来讲解。List就是STL中的链表容器，它实现的是前面提到的双向、循环链表。如果你想要使用更节约空间的单链表，STL中也提供了相应的实现forward_list，有兴趣的话，你可以自己去了解背后的实现。

和上节课一样，因为STL的实现背后涉及了许多繁琐的C++高级语法，我们会对代码做一定的简化方便理解，你对C++不感兴趣的话也不用深究。

链表节点的实现

首先来看一下 list 的主要组成部分node ，也就是链表节点，它是整个链表的关键，存储着元素信息本身和连接链表的前后指针。

Node节点的实现如下：

1
2
3
4
5
6

template <class T>
struct __list_node {
  __list_node<T>* next;  // 前驱节点指针
  __list_node<T>* prev; // 后继节点指针
  T data; //存储数据
};

可以看到，链表节点的定义，除了为了支持各种元素类型而用到的泛型语法。其他的内容和前面说的双链表的节点完全一致，指针域同时包含了前驱和后继节点的地址，成员变量data用于存储元素信息本身。

链表迭代器的实现

所有的STL容器都需要实现迭代器，这也是后面所有操作的基础。

迭代器提供用于遍历的最重要的接口，它本身也是一种重要的设计模式，支持的操作就是自增++和自减--。上一讲的vector，因为内存是连续存储的，可以直接通过地址的++和–操作；但在内存不连续存储的List中，我们需要基于节点引用域中的前驱后继节点信息，来实现自己的迭代方法。

代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

template<typename T>
struct __list_iterator{
    typedef __list_iterator<T>   self;
    typedef __list_node<T>*      link_type;
    link_type ptr; //成员
    __list_iterator(link_type p = nullptr):ptr(p){}
}

T& operator *(){return ptr->data;}
T* operator ->(){return &(operator*());}
// 类似 ++x 返回next节点
self& operator++(){
    ptr = ptr->next;
    return *this;
}
// 类似 x++ 返回当前节点
self operator++(int){
    self tmp = *this;
    ++*this;
    return tmp;
}
// 类似 --x 返回prev节点
self& operator--(){
    ptr = ptr->prev;
    return *this;
}
// 类似 x-- 返回当前节点
self operator--(int){
    self tmp = *this;
    --*this;
    return tmp;
}
bool operator==(const __list_iterator& rhs){
    return ptr == rhs.ptr;
}
bool operator!=(const __list_iterator& rhs){
    return !(*this==rhs);
}

其实就是将迭代器的方法都实现一遍。我们重点需要关注的操作是++和–，对应实现也非常直观。

在迭代器中，ptr是我们的主要成员变量，它指向的就是迭代器当前遍历的节点地址。所以++就是返回一个指向 ptr->next 的指针；而–对应的，就是返回一个指向prt->prev 的指针；同时我们需要把内置的ptr 也指向prt->next 或者 ptr->prev。这样我们就可以自如地用迭代器在链表上进行遍历了。

链表数据结构的实现

有了迭代器和节点，我们要做的就是将STL中双向循环链表的结构，用C++语言描述出来，并将一些链表相关的基本操作实现出来。

所谓链表，就是要将节点链接起来。由于链表节点本身已经存了前置后继节点的地址，所以链表数据结构主要的内涵，其实只需要使用一个节点即可表示出来，用这一个节点，我们就可以找到所有其他的节点。

所以数据结构定义如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

template<typename T>
    class list{
        protected:
            typedef __list_node<T> list_node; // 显示定义list_node类型
            typedef allocator<list_node> nodeAllocator; // 定义allocator类型
        public:
            typedef T                  value_type;
            typedef T&                 reference;
            typedef value_type*        pointer;
            typedef list_node*         link_type;
            typedef const value_type*  const_pointer;
            typedef size_t             size_type;
        public:
            typedef __list_iterator<value_type> iterator; // 迭代器类型重写
        private:
            link_type node; // 只要一个指针，便可表示整个环状双向链表
            // ......
    }

这段代码看起来比较长，其实大部分是一些类型定义，比如简写了带泛型的节点类型，更多是为了保证语义清晰和可读。我们真正要关注的只有private的成员变量 node，事实上，这一个节点就可以表示整个环状双向链表。

在内存中的排布方式如下图所示：

每个链表数据结构，都会有一个虚拟节点的成员变量node，用于标记这整个循环链表的首尾连接处，它既是整个链表的开始，也是整个链表的结尾；也就是说，这个虚拟节点的pre 指向链表的最后一个节点，它的next指向链表的第一个节点。

所以链表初始化容量为零的时候，显然只有一个前后指针都指向自己的虚拟节点。

这里还有一个非常巧妙的设计，我们会让end()迭代器指向这个虚拟节点，begin()则会指向虚拟节点的下一个节点，这完美符合迭代器前闭后开的语义。

因为end()节点指向的是一个并不真实存储数据的元素，是永远取不到值的；而对应的begin()，在链表不空的时候，指向正是链表中的第一个节点。因此要遍历链表所有的元素的时候，就会写出这样的代码：

1
2

for (std::list<int>::iterator it=mylist.begin(); it != mylist.end(); ++it)
    std::cout << ' ' << *it;

我们在判断是否遍历到终点的代码，就和vector一样，写的是 it != mylist.end() 。

链表基本操作的实现

好了，终于来到最激动人心的部分：链表的基本操作。

有了刚才的数据结构和迭代器，很容易访问到链表的节点了，我们开始实现链表的另外几个主要操作：初始化链表、插入节点、删除节点。

先来看一切的开始，链表是如何初始化的。

相比数组其实要简单一些，因为前面说了，一个空的链表，就只包含了一个虚拟节点，STL内置的空间配置器很容易处理这个节点的内存申请。初始化代码如下：

1
2
3
4
5
6
7

void empty_initialize() {
  node = get_node(0);
  node->next = node; // next 指针指向自身
  node->prev = node; // prev 指针指向自身
}

link_type get_node() { return list_node_allocator:allocate(); }

当链表为空时，虚拟节点前、后指针都指向自身，代码就是如此简洁直观。

那怎么往链表里插入数据呢，我们主要看 insert 方法是如何实现的，它用来在链表中的任何一个节点后面插入数据。有了insert，我们当然也能很容易实现 push_back 等常用方法。

那 insert 需要传入哪些参数呢？

前面也说了，相比数组，链表的最大优势之一就是它的插入和删除效率会高效得多。这正是因为内存空间不是线性排列的，所以想要插入数据，我们只需要修改指定位置的前、后指针的指向，把新的节点在逻辑上插入某个位置就可以了。

所以 insert需要两个入参，一个是插入位置，类型是一个迭代器；另一个是插入的值。其实现方法如下：

1
2
3
4
5
6
7
8

iterator insert(iterator position, const T& x) {
  lik_type tmp = create_node(x); // 创建一个临时节点
  tmp->next = position.node; // 将该节点的后继指针指向当前位置的节点
  tmp->prev = position.node->prev; // 将该节点的前驱指针指向当前位置的前驱节点
  (link_type(position.node->prev))->next = tmp; // 将前驱节点本来指向当前节点的后继指针改为指向该临时节点
  position.node->prev = tmp; // 同样，当前位置的前驱指针也要修改为指向该临时节点
  return tmp;
}

代码其实并不复杂，但许多新手还是需要花一些时间理解一下的，整个过程有点像“穿针引线”。

先创建一个节点tmp，将该节点的前驱后继分别指向当前position的前驱和position本身；再将当前position的后继和position->prev的前驱指向这个新创建的节点。这样我们就完成了链表节点的插入。

这些操作的顺序是非常重要的。你可以理解成要先把新的节点全接上去，才能把旧指针一一改过来。

如果调换了操作的顺序，比如先将当前position的后继指向临时节点，那么我们就访问不到插入节点的后继节点了。建议你用纸笔多模拟几遍整个插入的过程，多练习几遍自然就能掌握了。

而链表的删除操作是类似的，基本上就是进行一组和插入相反的操作。找到某个需要删除的节点位置，将该节点的后继和前驱直接关联在一起，最后释放掉待删除节点的空间即可。代码如下：

1
2
3
4
5
6
7
8

iterator erase(iterator position) {
  link_type next_node = link_type(position.node->next);
  link_type prev_node = link_type(position.node->prev);
  prev_node->next = next_node;
  next_node->prev = prev_node;
  destroy_node(position.node);
  return iterator(next_node);
}

有了 insert 和 erase操作，其他一些基础操作当然也很容易衍生出来。比如pop_front/pop_back/push_back，我们只需要在指定的位置调用 erase 和 insert 即可，首尾的位置都可以通过begin、end等迭代器接口快速取到：

1
2
3
4
5
6
7

void pop_front() { erase(begin()) };
void pop_back() {
  iterator tmp = end(); 
  erase(--tmp);
}

push_back(const T& x) { insert(end(), x); }

这三个操作都是在O(1)时间复杂度内可以完成的，比vector对应的操作O(n)的时间复杂度要高效很多。

list其实还支持sort等操作，借助内部实现的transfer方法和归并排序的思想，同样可以做到O(n*logn)的复杂度。但实现比较复杂，如果你有兴趣，可以去看自己搜索一下相关的资料，力扣上有一道关于链表排序的题目也可以练习。

总结

链表，相比于数组，有更好的灵活性和更低的插入、删除的复杂度，更加适用于查询索引较少、遍历、插入、删除操作较多的场景，所以要频繁在容器中间某个位置插入元素的时候，就经常用到，比如在LRU和操作系统进程调度的场景下就都会用到。

链表操作在实现的过程中主要需要注意指针之间的变换顺序，你可以在脑海里多模拟几遍这样的过程，并尝试在不借助参考资料的前提下自己实现几次，这也是面试官在算法面试中经常会考察的点。

课后作业

今天没有讲链表中find方法的实现，这也是STL在各种容器中都会提供的一个通用方法。该方法用于寻找容器中某个值的迭代器，比如链表 1->5->3->4 中，调用find(3)，你应该返回的就是链表中的第三个节点的迭代器。你可以来实现一下find方法吗？时间复杂度又是多少呢？

欢迎留言与我讨论。如果你觉得文章有帮助的话呢，也欢迎你点赞转发，我们下节课见～

03｜双端队列：并行计算中的工作窃取算法如何实现？

作者: 黄清昊

你好，我是微扰君。

目前我们已经学习了 vector 动态数组和 list 双向链表两种STL中的序列式容器了，今天我们继续学习另一种常见的序列式数据结构，双端队列。

在并行计算中，我们常常会用多进程处理一些复杂的计算任务。为了能够通过多进程加速计算，我们除了需要对任务进行合理的切分，也需要将任务合理公平地分配到每一个进程。简单来说就是，我们希望每个进程都不至于闲着。那怎么样能做到这件事呢？

其实有一种非常常用的算法，工作窃取算法，就可以用来达成这个目标，它就需要用到我们今天的主角——双端队列。

队列

要介绍双端队列，我们先来聊一聊队列，queue。什么是队列呢？

从概念上来说其实非常好理解，因为它的特性和“队列”这个词在现实生活中的意思是一致的，那就是FIFO先进先出。简单来说就是排队。

比如说现在到很多餐厅就餐，服务员都会给你发一个号码让你排队，等有空位的时候，服务员叫号是按照取号的顺序来的，肯定是先来取号的人结束排队去入座；这样的约束就是先进先出。

显然这种先进先出的队列也是一种典型的序列式数据结构；和数组最大的区别就在于，它是一个有约束的序列式数据结构，因为先进先出的特性要求我们，所有的插入操作必须在队列的尾部进行，而所有的删除操作则必须在队列的头部进行。

上图就是一个对队列入队、出队操作的示例。我们注意到先入队的元素一定会比后入队的元素更早出队。这一特性和思想在许多业务系统或者基础软件、操作系统、计算机网络中都有应用，比如在操作系统中的CPU调度中，进程资源使用CPU的顺序就用队列来排序。

双端队列

队列和链表一样也会延展出更多种类的队列，比如带权重的优先队列、或者只能一边进一边出的单端队列。

我们今天要实现的double ended queue，双端队列是其中一种，相比于普通队列而言，双端队列是两端开口的，在队列的头尾两端都可以进行进队和出队操作，让我们在使用队列时有了更大的灵活性。

你肯定想问，数组也可以在两边插入数据呀，那双端队列和数组有什么区别呢？

首先，数组头部的插入操作复杂度很高，如果我们并不需要快速随机访问，这种操作的复杂度是完全可以避免的，这是双端队列和数组的一个很大区别。更本质的地方在于，双端队列仅仅是一个两端都支持FIFO插入删除操作的队列，语义上来说并不支持数组基于下标在指定位置的修改、插入和删除的操作。

当然，我们是可以用数组或者链表来模拟实现双端循环队列的，只要暴露出经过剪裁的且满足FIFO的语义方法就可以了。

比如可以开一个大小为N的数组array，用两个数字 rear 和 front 代表队列的前端和尾端。在前端插入 target，只需要 array[(--front+N)%N] = target，这样既扩展了前端的边界，也达到了插入target的效果。%N也就是要对N取模，主要也就是为了处理越界的问题，这样当数组的前端read到达小于0的位置时，就会马上变成N-1，也就实现了一个循环队列。

Deque实现

虽然说，可以用数组或者链表来实现队列，但C++并没有选择依赖已有的序列式容器vector或者list来实现，原因是什么呢？你可以先想一想。

带着这个问题，我们一起来学习后面的内容，看看STL中的deque是如何实现一个高效好用的双端队列的。

我个人认为，在 STL 序列化容器的空间分配中，deque 可能是最复杂的，这也可能会对你阅读源码造成一定的障碍，但是不要害怕，如果只是为了搞清deque设计的大致思想，我们完全可以将内存分配的部分当成黑盒来看，这对搞清楚deque的原理并没有什么影响。

Deque的内存布局

deque的内存布局，可以说同时具备了list和vector的特点。

deque的内存布局是由一段段连续的空间、用另一个类似数组的东西将这些空间的地址信息拼接在一起组成的，真实存放数据的就是那一段段连续的空间。在首尾两端插入和删除的时间复杂度是O(1)。以插入为例，每次一段连续的空间元素被用完的时候，会直接申请一段新的空间并链接到deque的分段空间末尾。

所以deque既不像 vector 那样每次扩容都需要付出复制和拷贝的高昂代价，也不会像链表那样每次插入一个新的节点都需要申请一次内存。

当然这也导致了非常复杂的控制流程，deque的代码量也远远多于vector和list。

为了维护一段段连续的内存空间，deque需要维护一个被称为map的成员变量；这个map数据结构起到了管理真正用于存储队列元素的一段段连续线性空间的作用。那一段段连续的线性空间，我们称为缓冲区。

map的示意图如下：

可以认为map是一个数组，每个元素指向了一段缓冲区的地址。而缓冲区对应了一段指定大小的连续内存空间，默认大小为 512 bytes。

1
2
3
4
5
6
7
8
9
10

template <class _Tp, class _Alloc>
class _Deque_base {
  ...
protected:
  _Tp** _M_map;
  size_t _M_map_size;  
  iterator _M_start;
  iterator _M_finish;
  ...
}

因此 _M_map 在数据结构中的表现就是一个二级指针。_M_map_size指的就是 deque 中 map 的空间大小，即在map中最多能存储多少个指针。如果map的空间已经被用满了，我们也会对map进行一次重新分配迁移的操作，核心思想和vector的重分配其实是一样的，我们马上具体讲。

Deque的迭代器

介绍完内存布局和基本数据结构，下一个重点就是STL的通用访问模式，迭代器的实现了。

正是因为 deque 底层实质是分段连续空间，operator++ 和 operator-- 的实现也变得更困难一些，迭代器既要能找到与当前缓冲区相邻的缓冲区在哪；也需要知道目前访问的地方是否已经到当前缓冲区的边缘，只有这样到边缘时，才能正确跳转。

为了方便达到这一目标，我们需要在迭代器的数据结构中记录一下迭代器在当前缓冲区的位置，同时记录当前缓冲区的开始位置和结束位置，以及缓冲区的map指针：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

template <class _Tp, class _Ref, class _Ptr>
struct _Deque_iterator {
  typedef _Deque_iterator<_Tp, _Tp&, _Tp*>             iterator;
  typedef _Deque_iterator<_Tp, const _Tp&, const _Tp*> const_iterator;
  static size_t _S_buffer_size() { return __deque_buf_size(sizeof(_Tp)); }
  ...
  typedef _Tp** _Map_pointer; // 缓冲区指针
  ... 
  _Tp* _M_cur; // 当前缓冲区的位置
  _Tp* _M_first; // 缓冲区的左边界线
  _Tp* _M_last; // 缓冲区的右边界
  _Map_pointer _M_node;
  _Deque_iterator(_Tp* __x, _Map_pointer __y) 
    : _M_cur(__x), _M_first(*__y),
      _M_last(*__y + _S_buffer_size()), _M_node(__y) {}
}

有了位置的记录，operator++ 可以这样实现：

1
2
3
4
5
6
7
8
9
10
11
12
13

_Self& operator++() {
    ++_M_cur;
    if (_M_cur == _M_last) {
      _M_set_node(_M_node + 1);
      _M_cur = _M_first;
    }
    return *this; 
  }  
  void _M_set_node(_Map_pointer __new_node) {
    _M_node = __new_node;
    _M_first = *__new_node;
    _M_last = _M_first + difference_type(_S_buffer_size());
  }

核心的就是_M_set_node方法，如果我们发现M_cur已经达到了当前缓冲区的尾部，就将它移动到下一段缓冲区的头部，更新迭代器中当前map的位置。另外，也需要将_M_first和_M_last更新为新的缓冲区的左确界和右虚界。

-- 的操作类似：

1
2
3
4
5
6
7
8

_Self& operator--() {
    if (_M_cur == _M_first) {
      _M_set_node(_M_node - 1);
      _M_cur = _M_last;
    }
    --_M_cur;
    return *this;
  }

我们发现M_cur达到缓冲区头部的时候，就要将它移动到当前缓冲区的前一段缓冲区了，调用set_node方法即可。

到这里就完成了迭代器的主要接口，这让我们将内存实质不连续的真相隐藏了起来，取而代之地提供了一个非常简洁好用的遍历deque的接口。

好啦，学完deque 的内存布局和迭代器如何实现，你知道它的基础操作该怎么写了吗？

Deque的基础操作

相比于vector和list来说，deque支持的操作要少得多，只有基本的push和pop实现，因为队列语义保证了我们不会在队列中间进行插入删除操作，也就不用支持insert和erase这样的操作了。

不过正因为内存布局复杂，deque的内存管理扩缩容的逻辑也比较复杂，我们了解大概思想就可以了。如果你感兴趣可以自行查阅deque源码。

push操作

Deque的第一个操作当然是push_front和push_back，因为我们实现的是双端队列，所以头部尾部都有可能插入数据。

遇到内存不足的时候，deque会按照下图的逻辑进行扩容，有几个检查点，首先判断是不是能在当前缓冲区插入元素，如果可以，直接插入就行；如果不能，就要检查缓冲区map两端是否有足够的空间；如果有的话，也很简单，直接创建一个新的缓冲区并存入map。

关键是在map空间不足的时候，也就是插入的数据已经达到map头部或者尾部缓冲区的边界时，我们可以分两种情况讨论：

1. 如果 map使用率已经超过一半，我们就可以重新申请更大的空间，把老的map上的数据拷贝到新的区域。这里注意，map中指向的那些缓冲区里的数据并不用变化，只是需要一个更大的map去放那些缓冲区的指针，和动态数组扩容的方式如出一辙。
2. map使用率没有超过一半，这时候我们认为申请新的空间可能是浪费的，所以只是将数据重新调整到map中间的位置，当然也要进行一次拷贝。这可能会帮我们节约大量的空间。

翻译成代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

void push_back(const value_type& __t) {
    if (_M_finish._M_cur != _M_finish._M_last - 1) {
      construct(_M_finish._M_cur, __t);
      ++_M_finish._M_cur;
    }
    else
      _M_push_back_aux(__t);
  }
template <class _Tp, class _Alloc>
void deque<_Tp,_Alloc>::_M_push_back_aux()
{
  _M_reserve_map_at_back();
  *(_M_finish._M_node + 1) = _M_allocate_node();
  __STL_TRY {
    construct(_M_finish._M_cur);
    _M_finish._M_set_node(_M_finish._M_node + 1);
    _M_finish._M_cur = _M_finish._M_first;
  }
  __STL_UNWIND(_M_deallocate_node(*(_M_finish._M_node + 1)));
}

pop操作

pop操作不再需要处理插入导致的扩容拷贝问题, 相对来说就显得简单很多。以pop_back为例，我们只需要关注是否已经pop到某一段缓冲区的边界。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

void pop_back() {
    if (_M_finish._M_cur != _M_finish._M_first) {
      --_M_finish._M_cur;
      destroy(_M_finish._M_cur);
    }
    else
      _M_pop_back_aux();
  }
// Called only if _M_finish._M_cur == _M_finish._M_first.
template <class _Tp, class _Alloc>
void deque<_Tp,_Alloc>::_M_pop_back_aux()
{
  _M_deallocate_node(_M_finish._M_first);
  _M_finish._M_set_node(_M_finish._M_node - 1);
  _M_finish._M_cur = _M_finish._M_last - 1;
  destroy(_M_finish._M_cur);
}

如果发现当前迭代器已经和缓冲区的首位置相同，除了释放掉当前的内存，还需要释放掉整段缓冲区的内存，并且将迭代器的缓冲区指针，指向当前缓冲区前一段的位置，这可以通过_M_set_node方法达成。当然，由于我们还需要pop一个节点，所以会将_M_cur指向_M_finish._M_last-1的位置。

C++的选择

现在掌握了deque的实现和基本操作，我们来回答一下为什么C++不选择依赖已有的序列式容器来实现deque？

其实我们已有的容器就两个，一个是vector，另外一种就是list。

显然，基于vector实现，不能真的在头部插入元素，会产生O(N)的时间开销，我们只能用一个固定大小的vector来模拟循环队列，具体实现方式前面说过。但这样就导致我们必须事先确定数组的最大容量，让它的大小是实现分配好的，这就和数组一样，也会产生内存浪费和无法动态扩容的问题。

不过在最大容量能确定的场景下，用vector也是一种非常常见的循环队列实现方式。

而基于list，看起来首尾都可以O(1)的时间插入，但对数据的随机读取性能会很差；且每次插入元素都需要申请内存，相比于deque一次申请一段内存的方式也会带来额外的性能开销。而list的最大优势，任意位置的快速插入/删除能力，我们却用不上。

所以基于deque的使用场景，C++设计了基于map分段存储的双端队列的数据结构，能同时具备list和vector的特点。

总结

队列的基本特性是FIFO，也就是先进先出，它能衍生出几种不同的形式，包括循环队列、双端队列，既可以通过数组实现，也可以通过链表实现。

STL的deque是一种双端队列的实现，内存布局是由一段段连续内存串联起来的，在队列两端都可以pop和push数据。因为复杂的内存分配，代码实现的难度要高很多。但更多的复杂性还是体现在内存管理中，只要我们通过迭代器等模式，将底层的逻辑封装起来，相信你也看到了，pop和push操作的思路其实是非常清晰好懂的。

现在你知道为什么说工作窃取算法需要用到双端队列了吗？

我们一起看看。为了更公平也更高效地分配每个进程负责的任务，我们可能会多开很多个队列去存储任务，每个进程就去消费一个队列中的任务，这样就可以有效避免进程间的竞争。因为任务先进先出，用一个普通的单向队列就可以完成了。

但是你可能很难保证任务划分得非常均匀，使得每个进程完成所有任务的时间都差不多。这不是一个很好解决的问题。但是如果我们换一个思路，不再费心让任务分配得均匀，只是简单地允许先完成任务的进程，去其他进程的队列盗取任务，是不是就不会有进程闲置了呢？

不过怎么盗取，可以让我们仍然尽量规避进程间的竞争问题呢？ 相信你已经想到答案了，没错，就是双端队列。我们让盗取任务的进程，从队列的另一端盗取就行了，这样只有队列长度为1的时候才会出现竞争。当然还有很多实现细节，你感兴趣的话可以去看一下Java中ForkJoinPool的实现。

课后作业

最后，同样给你留一个课后作业。我们讲解了如何用数组实现队列，也提到队列同样可以通过链表来实现？你可以试着实现一下吗？

欢迎你留言与我讨论交流～

04｜栈：函数调用的秘密究竟是什么？

作者: 黄清昊

你好，我是微扰君。

目前为止，我们已经介绍了STL里的大部分序列式容器，包括vector、deque和list，也对应着数组、队列和链表这几种基础数据结构；今天我们来学习最后一种常用的线性数据结构，栈。

栈这个词，相信每一个研发同学在学习编程的过程中都会经常听到。不仅仅是因为栈本身就是一种基础的、常见的数据结构，更因为栈在计算机世界里起着举足轻重的作用。

在编程语言中，栈除了作为一种常用的数据结构，也常常用来表示一种叫做“调用栈”的概念，它是编程语言之所以能实现函数调用的关键所在。而在内存分配中，栈也表示一种内存分配的区域，和内存中的堆区是一种相对的概念。

栈区是有结构和固定大小的，区块按照次序存放，每个线程独占一个栈区，总的大小也是事先确定的；而堆区则没有固定的大小，数据可以随意存放。我们常常听到的 stack overflow 错误，也就是栈溢出错误，就是指程序在运行时，存放在栈上的数据已经多于栈区的容量，产生了容量不足的错误。相信说到这，你就更加明白为什么说栈相比于其他数据结构更经常被听到了吧。

其实无论是调用栈还是内存中的栈区，这两种含义都和栈数据结构的LIFO特性有关。如果你已经有了充分的背景知识，可以先想想这是为什么？

栈的特性：LIFO

和队列一样，我们也可以将栈理解成一种有约束的序列式数据结构，但是不同于FIFO的队列，栈的约束在于，插入和移除元素的操作，都只能在该数据结构的一端进行。栈对外暴露的插入和删除接口，我们一般称为push和pop，操作的那一侧，也称为栈顶，不能操作的那一侧则叫做栈底。

我们只能从栈顶删除或者插入元素，这直接保证了LIFO，也就是先进后出的特性。

栈有一个很好理解的生活中的例子就是坐电梯。显然电梯就是这样一种有着单侧开口性质的容器，把电梯里的乘客看成是容器中的元素，我们发现先进入的人肯定不太好直接越过后进来的人们先下电梯，因为出口和入口是一样的，并且只有一个。

这也就是为什么我们坐电梯到目标楼层准备出来的时候，经常会需要前面的人让一让。相信你只要想象一下这样的场景，就可以理解栈这个数据结构的精髓所在啦。

STL 中 stack 的实现

stack 就是 STL 中对栈这一数据结构的实现。其实，相比上一讲内存管理逻辑复杂的队列，stack的代码实现非常简单。

前面说了 stack 有单侧开口、后进先出的特性，有没有想到之前讲过的哪个容器也能实现类似的效果呢？

其实不只一个可以实现。比如 vector 就可以通过在一侧 push_back 和 pop_back 的操作模拟栈的 push 和 pop；同理，deque也可以通过在同一侧（比如头部）的 push_front 和 pop_front 操作进行栈的模拟，这也正是 STL 的做法。也正是因为这个原因，stack 的实现就非常简单了。

不过 stack 并没有像前几讲介绍的数据结构那样真正实现底层接口的逻辑，而仅仅基于 deque 现有的能力去改造出符合 stack 语义的接口，是不是就像一个接口转换器呢？就好像是把一个三口的插头转成了两口的插头。

事实上，这样的封装方式，也正是一种设计模式：“适配器”模式。所以也有人认为 stack 不是一种 container，容器，而是一种 container adapter，适配器容器。

数据结构定义

好了，来看一下 stack 在 STL 中的具体实现，我们同样先来看看 stack 的数据结构定义：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

template<typename _Tp, typename _Sequence = deque<_Tp> >
class stack
{
public:
    typedef typename _Sequence::value_type                value_type;
    typedef typename _Sequence::reference                 reference;
    typedef typename _Sequence::const_reference           const_reference;
    typedef typename _Sequence::size_type                 size_type;
    typedef          _Sequence                            container_type;
protected:
    _Sequence c; // stack 底层容器； 默认为 deque
public:
     reference
      top()
      {
        __glibcxx_requires_nonempty();
        return c.back();
      }
    void push(const value_type& __x) { c.push_back(__x); }
    void pop() {c.pop_back();}
    ...
}

这里有一行 __glibcxx_requires_nonempty ，这实际上是一个宏而不是函数，你不用太过关心，主要是用于帮助你debug的，对代码的运行没有任何影响。

可以看到 stack 下有一个受保护的成员变量 _Sequence，它就是我们说stack作为适配器所适配的容器，在默认的情况下，_Sequence 的选择是用 deque 作为它底层的存储容器，所以其扩容机制，当然也就是依赖了deque的实现。

这也是适配器的魅力所在，给我们所需要的类型提供了一套统一的接口，但底层所适配的容器依旧是可配置的；我们可以根据策略随时改变底层容器的选择，不会对外界造成任何影响。

现在既然已经有了一个功能强大的底层容器“deque”，我们当然可以基于它快速实现 stack 所需要的所有接口。在 STL 的实现中，我们封闭了 deque 的前端，不再有任何地方可以进行 push_front 或者 pop_front 操作了。 而stack最关键的两个方法 push 和 pop ，只需要简单地调用 c.push_back() 和 c.pop_back() ，就可以达到模拟单侧开口的栈的效果。

除了可以利用其他数据结构，stack 实现起来非常简单还有另一个原因，它不需要暴露迭代器。在标准的栈“后进先出”的语义下，我们并不需要对stack做随机访问和遍历的操作，加上只有栈顶的元素才会被外界访问，又省去了实现迭代器的很多逻辑。而栈顶元素top也只需要调一下内置容器的back方法即可取得。

当然，这也让我们失去了一些属于deque的能力，但我认为这正是面向对象6大设计原则之一的接口隔离原则的体现，当我们使用栈的时候，就不应该去关心如何迭代它。

到这里stack 在 STL 中的实现我们就已经全部学完了，是不是非常简单呢？

stack的应用 - 调用栈

掌握了栈作为数据结构的基本特性和实现，我们来回答开头的问题，在编程语言中，函数调用栈里的栈到底是什么，和栈的LIFO特性有什么关系，为什么它也被命名为栈呢？

函数调用

先来简单复习一下什么是函数调用，以及函数调用背后有哪些过程。我们就以JavaScript语言为例来学习，因为现代的Web应用都是跑在Google的V8引擎之上的，这能避免我们涉及太多寄存器和汇编语言相关的细节，毕竟那些细节对理解函数调用栈并没有太直接的帮助。

来看一下这段代码，这里面就包含一个典型的函数调用：

1
2
3
4
5
6
7
8
9
10
11

function add(a, b) {
    console.trace();
    return a + b;
}
function avg(a, b) {
    console.trace();
    let res = add(a, b) / 2;
    console.trace();
    return res;
}
let x = avg(0, 100);

代码中一共包含了两个函数，add用于求两数之和，avg用于求两数均值。其中avg的逻辑里就调用了add函数，而对新变量x的赋值过程，也调用了avg函数。

整个调用链路是：对x赋值 -> call avg -> call add -> return add -> return avg。

这个调用的结构是不是天然看起来就像是前面举的“进电梯-出电梯”的例子呢？ 函数的call和return仿佛就像是进电梯和出电梯的过程，后call的先return，完美符合了 LIFO 的原则。事实上也正是如此，但call和return的过程究竟是什么呢？下面我们就来一探究竟。

因为Chrome浏览器的运行时提供了很好用的打印调用栈信息的功能，我们就调用一下，直观地感受调用栈的状态变化：

可以看到，第一次打印的时候，调用栈中包括avg函数和匿名函数。第二次打印的时候在调用栈的顶部又增加了一个add函数。第三次打印的时候add又一次消失在了栈顶。

这是因为在代码的第2、6和8行，打印了三次调用栈的信息。由于avg函数第一个执行trace命令，是在调用add之前，所以首先打印在console中的调用信息属于avg函数，随后是add函数，最后是avg调用完add之后的第二次打印。

如果将调用栈的信息做一个完整的展示，大概是这个样子：

看起来显然更像一个栈了。现在，调用栈之所以叫调用“栈”，也就不言而喻了，它记录程序正式运行过程中函数调用的后call先return情况。

那么调用栈里的一个个函数到底是什么呢？

我们知道，每个函数都有一个自己的作用域，但不同作用域下的变量可以有相同的变量名。比如在刚才的例子中，avg和add函数中的入参变量名都是a和b，它们互相不影响。这就是因为每个函数都会有一个自己的上下文，而上下文中存放着变量名和值的绑定，不同的上下文是彼此隔离的。

当程序每次执行到一个函数调用的时候，操作系统或者虚拟机就会在栈上分配一块区域，我们称之为栈帧。

简单来说，栈帧中就存放着函数执行的上下文。当前计算完成之后，我们就会将执行结果返回，并绑定到上一个栈帧内的变量里，当前栈帧的所有资源也就可以释放了。

这个释放过程，在操作系统或者虚拟机底层，最后都会转化成几个寄存器值的变化，成本非常低廉。事实上，各个语言在实现函数调用的时候都是不约而同地依赖调用栈去实现的，只不过js建立在V8引擎之上，栈帧里存的就可以是执行上下文，包括了变量和值的绑定等信息；而在C语言里可能就直接操作的几个寄存器的值，如EAX、ESP等，和具体的CPU指令集架构有关。

比如前文中js的例子，当average函数调用add函数时，我们就创建了一个属于add的执行上下文，其中a和b绑定着从average中传递过来的变量。因为上下文是隔离的，在add的执行过程中，无论我们怎么修改a和b的值，对average上下文中的变量其实都是没有影响的。

假设用一个数组来表示执行上下文栈，其过程大概如下，average和add有各自的上下文也就对应着各自的活动对象。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

EXE_STACK = [];

EXE_STACK.push(<main> functionContext);
EXE_STACK.push(<average> functionContext);
Average_AO = {
  arguments: {
    0: 5,
    1: 100,
    length: 2
  },
  a: 5,
  b: 100,
}
EXE_STACK.push(<add> functionContext);
Add_AO = {
  arguments: {
    0: 5,
    1: 100,
    length: 2
  },
  a: 5,
  b: 100,
}

EXE_STACK.pop(); // add 执行完毕
EXE_STACK.pop(); // 执行 average
EXE_STACK.pop(); // 执行 main

等add执行完毕之后，我们会将add的结果返回，并继续执行average后续的操作。返回后，add的上下文也就没有必要再保留了，直接释放掉即可。所以整个代码执行过程看到的调用栈才会是上图中展现的样子。

至此，整个上下文从创建到销毁的过程完美契合了栈LIFO的原则。正是因为每次调用前，我们都有将上下文的信息保留在栈中，调用的计算过程并不会影响到调用前的内存空间，完美地做到了函数调用保留现场的作用；调用完成之后，我们可以延续调用前上下文中的状态，继续进行后续的计算。

在栈上连续的内存分配，以及函数调用完，不会再有其他地方需要函数上下文内变量的特点，让在栈上的内存管理变得简洁而高效。所以可以说，通过使用栈，我们优雅地实现了函数调用这一编程语言中最基础的核心能力。

在栈上连续的内存分配，以及函数调用完函数生命周期就结束，也就是不会再有其他地方需要函数上下文内变量的特点，让在栈上的内存管理变得简洁而高效；释放内存的操作和堆上完全不同，我们只需要直接改变函数栈帧指针的指向即可。所以可以说，通过使用栈，我们优雅地实现了函数调用这一编程语言中最基础的核心能力。

总结

栈的主要特点就是先进后出，其实list、vector、deque都可以用来做stack的底层容器，只需要利用适配器模式封装，屏蔽一部分接口，只保留在容器一端的插入/删除操作即可，对应到stack上也就是push和pop两个操作。

通过今天的学习，相信你就能理解在函数里，调用栈为什么也叫栈了吧？就是因为函数调用的过程中，函数上下文的产生与销毁天然符合栈后进先出的特点。这在各个语言的编译器中都有体现，有兴趣的话你可以看看你熟悉的语言中调用栈实现的细节。

课后作业

最后留一个小问题给你，既然stack可以有很多种实现方式，为什么STL选择了用 deque 来实现栈呢？通过 vector 实现会有什么好处或弊端吗？ 时间复杂度又有什么差异？

欢迎你在留言区留言，一起交流今天的学习感悟。我们下节课见～

05｜HashMap：一个优秀的散列表是怎么来的？

作者: 黄清昊

你好，我是微扰君。

过去四讲我们学习了STL中全部的序列式容器，数组、链表、队列、栈；今天来谈一谈另一类容器，关联式容器。所谓“关联式”，就是存储数据的时候，不只是存储元素的值本身，同时对要存储的元素关联一个键，形成一组键值对。这样在访问的时候，我们就可以基于键，访问到容器内的元素。

关联式容器本身其实是STL中的概念，其他高级语言中也有类似的概念。我们这次就以JDK为例，讲解几种关联式容器的原理和实现。

统计单词次数

我们就从一个实际需求讲起。现在有一篇很长的英文文档，如果希望统计每个单词在文档中出现了多少次，应该怎么做呢？

如果熟悉HashMap的小伙伴一定会很快说出来，我们开一个HashMap，以string类型为key，int类型为value；遍历文档中的每个单词 word ，找到键值对中key为 word 的项，并对相关的value进行自增操作。如果该key= word 的项在 HashMap中不存在，我们就插入一个(word,1)的项表示新增。

这样每组键值对表示的就是某个单词对应的数量，等整个文档遍历完成，我们就可以得到每个单词的数量了。用Java语言实现这个逻辑也不难。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

import java.util.HashMap;
import java.util.Map;
public class Test {
    public static void main(String[] args) {
        Map<String, Integer> map = new HashMap<>();
        String doc = "aaa bbb ccc aaa bbb ccc ccc bbb ccc ddd";
        String[] words = doc.split(" ");
        for (String s : words) {
            if (!map.containsKey(s)) {
                map.put(s, 1);
            } else {
                map.put(s, map.get(s) + 1);
            }
        }
        System.out.println(map);
    }
}

但是HashMap是怎么做到高效统计单词对应数量的？它设计思路的核心究竟是什么呢？这个问题非常有意思，我们一起来思考一下。

一个单词

要统计每个单词的数量有点复杂，如果只统计某一个单词的数量呢，是不是就很好做了？

只需要开一个变量，同样遍历所有单词，遇到和目标单词一样的，才对这个变量进行自增操作；等遍历完成，我们就可以得到该单词的数量了。

按这个思路，一种很简单的想法当然是直接对每一个单词都统计一遍数量，我们把能想到的所有可能出现的单词都列出来，每个单词，单独用一个变量去统计它出现的数量，遍历所有单词，写一堆if-else来判断当前单词应该被累计到哪个变量中。

下面的代码是一个例子：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

import java.util.HashMap;
import java.util.Map;
public class Main {
    public static void main(String[] args) {
        int[] cnt = new int[20000];
        String doc = "aaa bbb ccc aaa bbb ccc ccc bbb ccc ddd";
        String[] words = doc.split(" ");
        int aaa = 0;
        int bbb = 0;
        int ccc = 0;
        int ddd = 0;
        
        for (String s : words) {
           if (s == "aaa") aaa++;
           if (s == "bbb") bbb++;
           if (s == "ccc") ccc++;
           if (s == "ddd") ddd++;   
        }
    }
}

在代码中就对目标文本统计了aaa、bbb、ccc、ddd这四个单词出现的次数。

但这样的代码显然有两个很大的问题：

1. 对单词和计数器的映射关系是通过一堆if-else写死的，维护性很差；
2. 必须已知所有可能出现的单词，如果遇到一个新的单词，就没有办法处理它了。

解决办法有没有呢？这个时候我们不禁想到了老朋友——数组。

我们可以开一个数组去维护计数器。具体做法就是，给每个单词编个号，直接用编号对应下标的数组元素作为它的计数器就好啦。唯一麻烦的地方在于，如何能把单词对应到一个数字，并且可以不同的单词有不同的编号，且每个单词都可以通过计算或者查找等手段对应到一个唯一的编号上？

解决思路

一种思路就是把文档中出现的字符串，也放在数组里，按照单词出现的顺序对应从0开始连续编号。

所以，一共要建立两个数组，第一个数组用于存放所有单词，数组下标就是单词编号了，我们称之为字典数组；第二个数组用于存放每个单词对应的计数器，我们称之为计数数组。这样，单词的下标和计数器的下标是一一对应的。

每遇到一个新的单词，都遍历一遍字典数组，如果没有出现过，我们就将当前单词插入到字典数组结尾。显然，通过遍历字典数组，可以获得当前单词的序号，而有了序号之后，计数数组对应位置的统计计数也非常简单。这样，我们就可以非常方便地对任意文本的任意单词进行计数了，再也不用提前知道文档中有哪些单词了。

至于数组开多大，可以根据英文词典的常用单词数来考虑，相信大部分文档中出现的单词很难超过1w个，那么绝大部分时候，开一个长度为1w的数组肯定就可以满足我们的需求。这样的字典数组，也有人叫做符号表，比如Haskell中的内置map就是基于这个思路实现的。

但很显然，这样的编号方式代价还是非常高，因为基于这么大的数组判断每个单词是否出现过，显然是一个O(D)的操作，其中D代表整个字典空间的大小，也就是一个文档中有多少个不同的单词。整体的时间复杂度是O(D*N)，这并不令人满意。

那这里的本质问题是什么呢？这个问题，其实可以抽象成一个给字符串动态编码的问题，为了让我们不需要遍历整个符号表来完成指定键的查找操作，我们需要找到一个更高效的给字符串编码的方式。

优化思路

整体的优化方式大概分成两类。

• 一种是我们维护一个有序的数据结构，让比较和插入的过程更加高效，而不是需要遍历每一个元素判断逐一判断，下一讲会介绍的关联式容器TreeMap就是这样实现的。
• 另一种思路就是我们是否能寻找到一种直接基于字符串快速计算出编号的方式，并将这个编号“映射”到一个可以在O(1)时间内基于下标访问的数组中呢？

当然是有的，并且方式很多。

以单词为例，英文单词的每个字母只可能是 a-z，那如果想用数字表示单词，最简单的方式莫过于用26进制来表示这个单词了。具体来说就是，我们用0表示a、1表示b，以此类推，用25表示z，然后将一个单词看成一个26进制的数字即可。

基于前面的思路，我们可以开一个比较大的数组来统计每个单词的数量，单词对应的计数器就是计数数组中下标为字符串所对应的二十六进制数的元素。翻译成代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

import java.util.HashMap;
import java.util.Map;
public class Main {
    public static void main(String[] args) {
        
        int[] cnt = new int[20000];
        String doc = "aaa bbb ccc aaa bbb ccc ccc bbb ccc ddd";
        String[] words = doc.split(" ");
        for (String s : words) {
            int tmp = 0;
            for (char c: s.toCharArray()) {
                tmp *= 26;
                tmp += (c - 'a');
            }
            cnt[tmp]++;
        }
        String target = "aaa";
        int hash = 0;
        for (char c: target.toCharArray()) {
            hash *= 26;
            hash += c - 'a';
        }
        System.out.println(cnt[hash]);
    }
}

这样，我们就得到了统计单词数的非常优秀的时间复杂度了，在近似认为单词26进制计算复杂度为O(1)的前提下，我们统计N个单词出现数量的时候，整体甚至只需要O(N)的复杂度，相比于原来的需要遍历字典O(D*N)的做法就明显高效的多。

这其实就是散列的思想。

散列

在散列表中，我们所做的也就是为每一个key找到一种类似于上述26进制的函数，使得key可以映射到一个数字中，这样就可以利用数组基于下标随机访问的高效性，迅速在散列表中找到所关联的键值对。

所以散列函数的本质，就是将一个更大且可能不连续空间（比如所有的单词），映射到一个空间有限的数组里，从而借用数组基于下标O(1)快速随机访问数组元素的能力。

但设计一个合理的散列函数是一个非常有挑战的事情。比如，26进制的散列函数就有一个巨大的缺陷，就是它所需要的数组空间太大了，在刚刚的示例代码中，仅表示长度为3位的、只有a-z构成的字符串，就需要开一个接近20000（26^3）大小的计数数组。假设我们有一个单词是有10个字母，那所需要的26^10的计数数组，其下标甚至不能用一个长整型表示出来。

这种时候我们不得不做的事情可能是，对26进制的哈希值再进行一次对大质数取mod的运算，只有这样才能用比较有限的计数数组空间去表示整个哈希表。

然而，取了mod之后，我们很快就会发现，现在可能出现一种情况，把两个不同的单词用26进制表示并取模之后，得到的值很可能是一样的。这个问题被称之为哈希碰撞，当然也是一个需要处理的问题。

比如如果我们设置的数组大小只有16，那么AA和Q这两个字符串在26进制的哈希函数作用下就是，所对应的哈希表的数组下标就都是0。

好吧，计算机的世界问题总是这样接踵而至。就让我们来逐一解决吧。

JDK的实现

现在，我们来好好考虑一下散列函数到底需要怎么设计。

• 整个散列表是建立在数组上的，显然首先要保证散列函数输出的数值是一个非负整数，因为这个整数将是散列表底层数组的下标。
• 其次，底层数组的空间不可能是无限的。我们应该要让散列函数在使用有限数组空间的前提下，导致的哈希冲突尽量少。
• 最后，我们当然也希望散列函数本身的计算不过于复杂。计算哈希虽然是一个常数的开销，但是反复执行一个复杂的散列函数显然也会拖慢整个程序。

带着这些思考，一起来看看JDK中对散列函数的选择吧。

在JDK（以JDK14为例）中Map的实现非常多，我们讲解的HashMap主要实现在 java.util 下的 HashMap 中，这是一个最简单的不考虑并发的、基于散列的Map实现。

找到用于计算哈希值的hash方法：

1
2
3
4

static final int hash(Object key) {
    int h;
    return (key == null) ? 0 : (h = key.hashCode()) ^ (h >>> 16);
}

可以发现非常简短，就是对key.hashCode()进行了一次特别的位运算。你可能会对这里的，key.hashcode 和 h>>>16，有一些疑惑，我们来看一看。

hashcode

而这里的hashCode，其实是Java一个非常不错的设计。在Java中每个对象生成时都会产生一个对应的hashcode。当然数据类型不同，hashcode的计算方式是不一样的，但一定会保证的是两个一样的对象，对应的hashcode也是一样的；

所以在比较两个对象是否相等时，我们可以先比较hashcode是否一致，如果不一致，就不需要继续调用equals，从统计意义上来说就大大降低了比较对象相等的代价。当然equals和hashcode的方法也是支持用户重写的。

既然今天要解决的问题是如何统计文本单词数量，我们就一起来看看JDK中对String类型的hashcode是怎么计算的吧，我们进入 java.lang 包查看String类型的实现：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

public int hashCode() {
    // The hash or hashIsZero fields are subject to a benign data race,
    // making it crucial to ensure that any observable result of the
    // calculation in this method stays correct under any possible read of
    // these fields. Necessary restrictions to allow this to be correct
    // without explicit memory fences or similar concurrency primitives is
    // that we can ever only write to one of these two fields for a given
    // String instance, and that the computation is idempotent and derived
    // from immutable state
    int h = hash;
    if (h == 0 && !hashIsZero) {
        h = isLatin1() ? StringLatin1.hashCode(value)
                       : StringUTF16.hashCode(value);
        if (h == 0) {
            hashIsZero = true;
        } else {
            hash = h;
        }
    }
    return h;
}

Latin和UTF16是两种字符串的编码格式，实现思路其实差不多，我们就来看StringUTF16中hashcode的实现：

1
2
3
4
5
6
7
8

public static int hashCode(byte[] value) {
    int h = 0;
    int length = value.length >> 1;
    for (int i = 0; i < length; i++) {
        h = 31 * h + getChar(value, i);
    }
    return h;
}

啊哈！我们发现也没有多么高深嘛，就是对字符串逐位按照下面的方式进行计算，和展开成26进制的想法本质上是相似的。

1

s[0]*31^(n-1) + s[1]*31^(n-2) + ... + s[n-1]

不过一个非常有趣的问题是为什么选择了31？

答案并不是那么直观。首先在各种哈希计算中，我们比较倾向使用奇素数进行乘法运算，而不是用偶数。因为用偶数，尤其是2的幂次，进行乘法，相当于直接对原来的数据进行移位运算；这样溢出的时候，部分位的信息就完全丢失了，可能增加哈希冲突的概率。

而为什么选择了31这个奇怪的数，这是因为计算机在进行移位运算要比普通乘法运算快得多，而31*i可以直接转化为(i << 5)- i ，这是一个性能比较好的乘法计算方式，现代的编译器都可以推理并自动完成相关的优化。StackOverflow上有一个相关的讨论非常不错，也可以参考《effective Java》中的相关章节。

h>>>16

好，我们现在来看 ^ h >>> 16 又是一个什么样的作用呢？它的意思是就是将h右移16位并进行异或操作，不熟悉相关概念的同学可以参考百度百科。为什么要这么做呢？

哦，这里要先跟你解释一下，刚刚那个hash值计算出来这么大，怎么把它连续地映射到一个小一点的连续数组空间呢？想必你已经猜到了，就是前面说的取模，我们需要将hash值对数组的大小进行一次取模。

那数组大小是多少呢？在 HashMap 中，用于存储所有的{Key,Value}对的真实数组 table ，有一个初始化的容量，但随着插入的元素越来越多，数组的resize机制会被触发，而扩容时，容量永远是2的幂次，这也是为了保证取模运算的高效。马上讲具体实现的时候会展开讲解。

总而言之，我们需要对2的幂次大小的数组进行一次取模计算。但前面也说了对二的幂次取模相当于直接截取数字比较低的若干位，这在数组元素较少的时候，相当于只使用了数字比较低位的信息，而放弃了高位的信息，可能会增加冲突的概率。

所以，JDK的代码引入了^ h >>> 16 这样的位运算，其实就是把高16位的信息叠加到了低16位，这样我们在取模的时候就可以用到高位的信息了。

当然，无论我们选择多优秀的hash函数，只要是把一个更大的空间映射到一个更小的连续数组空间上，那哈希冲突一定是无可避免的。那如何处理冲突呢？

JDK中采用的是开链法。

哈希表内置数组中的每个槽位，存储的是一个链表，链表节点的值存放的就是需要存储的键值对。如果碰到哈希冲突，也就是两个不同的key映射到了数组中的同一个槽位，我们就将该元素直接放到槽位对应链表的尾部。

JDK代码实现

现在，在JDK中具体实现的代码就很好理解啦，table 就是经过散列之后映射到的内部连续数组：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

transient Node<K,V>[] table;
final V putVal(int hash, K key, V value, boolean onlyIfAbsent,
               boolean evict) {
    Node<K,V>[] tab; Node<K,V> p; int n, i;
    //在tab尚未初始化、或者对应槽位链表未初始化时，进行相应的初始化操作
    if ((tab = table) == null || (n = tab.length) == 0)
        n = (tab = resize()).length;
    if ((p = tab[i = (n - 1) & hash]) == null)
        tab[i] = newNode(hash, key, value, null);
    else {
        Node<K,V> e; K k;
        // 查找 key 对应的节点
        if (p.hash == hash &&
            ((k = p.key) == key || (key != null && key.equals(k))))
            e = p;
        else if (p instanceof TreeNode)
            e = ((TreeNode<K,V>)p).putTreeVal(this, tab, hash, key, value);
        else {
            // 遍历所有节点 依次查找
            for (int binCount = 0; ; ++binCount) {
                if ((e = p.next) == null) {
                    p.next = newNode(hash, key, value, null);
                    if (binCount >= TREEIFY_THRESHOLD - 1) // -1 for 1st
                        treeifyBin(tab, hash);
                    break;
                }
                if (e.hash == hash &&
                    ((k = e.key) == key || (key != null && key.equals(k))))
                    break;
                p = e;
            }
        }
        if (e != null) { // existing mapping for key
            V oldValue = e.value;
            if (!onlyIfAbsent || oldValue == null)
                e.value = value;
            afterNodeAccess(e);
            return oldValue;
        }
    }
    ++modCount;
    if (++size > threshold)
        resize();
    afterNodeInsertion(evict);
    return null;
}

通过hash函数的计算，我们可以基于这个数组的下标快速访问到key对应的元素，元素存储的是Node类型。

估计你会注意到第21行进行了一个treeifyBin的操作，就是说当哈希冲突产生的链表足够长时，我们就会把它转化成有序的红黑树，以提高对同样hash值的不同key的查找速度。

这是因为在HashMap中Node具体的实现可以是链表或者红黑树。用红黑树的整体思想和开链是一样的，这只是在冲突比较多、链表比较长的情况下的一个优化，具体结构和JDK中另一种典型的Map实现TreeMap一致，我们会在下一讲详细介绍。

好，我们回头看整体的逻辑。

开始的5-8行主要是为了在tab尚未初始化、或者对应槽位链表未初始化时进行相应的初始化操作。从16行开始，就进入了和待插入key的hash值所对应的链表逐一对比的阶段，目标是找到一个合适的槽位，找到当前链表中的key和待插入的key相同的节点，或者直到遍历到链表的尾部；而如果节点类型是红黑树的话，底层就直接调用了红黑树的查找方法。

这里还有一个比较重要的操作就是第40行的resize函数，帮助动态调整数组所占用的空间，也就是底层的连续数组table的大小。

1
2

if (++size > threshold)
        resize();

随着插入的数据越来越多，如果保持table大小不变，一定会遇到更多的哈希冲突，这会让哈希表性能大大下降。所以我们有必要在插入数据越来越多的时候进行哈希表的扩容，也就是resize操作。

这里的threshold就是我们触发扩容机制的阈值，每次插入数据时，如果发现表内的元素多于threshold之后，就会进行resize操作：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77

final Node<K,V>[] resize() {
    Node<K,V>[] oldTab = table;
    int oldCap = (oldTab == null) ? 0 : oldTab.length;
    int oldThr = threshold;
    int newCap, newThr = 0;
    if (oldCap > 0) {
        if (oldCap >= MAXIMUM_CAPACITY) {
            threshold = Integer.MAX_VALUE;
            return oldTab;
        }
        else if ((newCap = oldCap << 1) < MAXIMUM_CAPACITY &&
                 oldCap >= DEFAULT_INITIAL_CAPACITY)
            newThr = oldThr << 1; // 翻倍扩容
    }
    else if (oldThr > 0) // 初始化的时候 capacity 设置为初始化阈值
        newCap = oldThr;
    else {               // 没有初始化 采用默认值
        newCap = DEFAULT_INITIAL_CAPACITY;
        newThr = (int)(DEFAULT_LOAD_FACTOR * DEFAULT_INITIAL_CAPACITY);
    }
    if (newThr == 0) {
        float ft = (float)newCap * loadFactor; // 用容量乘负载因子表示扩容阈值
        newThr = (newCap < MAXIMUM_CAPACITY && ft < (float)MAXIMUM_CAPACITY ?
                  (int)ft : Integer.MAX_VALUE);
    }
    threshold = newThr;
    @SuppressWarnings({"rawtypes","unchecked"})
    Node<K,V>[] newTab = (Node<K,V>[])new Node[newCap];
    table = newTab;
    if (oldTab != null) {
        for (int j = 0; j < oldCap; ++j) {
            Node<K,V> e;
            if ((e = oldTab[j]) != null) {
                oldTab[j] = null;
                if (e.next == null)
                    newTab[e.hash & (newCap - 1)] = e;
                else if (e instanceof TreeNode)
                    ((TreeNode<K,V>)e).split(this, newTab, j, oldCap);
                else {
                    // 新扩容部分，标识为hi，原来的部分标识为lo
                    // JDK 1.8 之后引入用于解决多线程死循环问题 可参考：https://stackoverflow.com/questions/35534906/java-hashmap-getobject-infinite-loop
                    Node<K,V> loHead = null, loTail = null;
                    Node<K,V> hiHead = null, hiTail = null;
                    Node<K,V> next;
                    // 整体操作就是将j所对应的链表拆成两个部分
                    // 分别放到 j 和 j + oldCap 的槽位
                    do {
                        next = e.next;
                        if ((e.hash & oldCap) == 0) {
                            if (loTail == null)
                                loHead = e;
                            else
                                loTail.next = e;
                            loTail = e;
                        }
                        else {
                            if (hiTail == null)
                                hiHead = e;
                            else
                                hiTail.next = e;
                            hiTail = e;
                        }
                    } while ((e = next) != null);
                    if (loTail != null) {
                        loTail.next = null;
                        newTab[j] = loHead;
                    }
                    if (hiTail != null) {
                        hiTail.next = null;
                        newTab[j + oldCap] = hiHead;
                    }
                }
            }
        }
    }
    return newTab;
}

看起来 resize 的代码比较复杂，但核心在做的事情很简单，就是要将哈希桶也就是内置的table数组，搬到一个更大的数组中去。主要有两块逻辑我们需要关注一下。

第一部分在第6-26行，主要的工作就是计算当前扩容的数组大小和触发下一次扩容的阈值threshold。

可以看到命中扩容条件的分支都会进入13行的逻辑，也就是每次扩容我们都会扩容一倍的容量。这和c++中STL动态数组的扩容逻辑是相似的，都是为了平衡扩容带来的时间复杂度和占用空间大小的权衡；当然这也是因为我们仍然需要保持数组大小为2的幂次，以提高取模运算的速度。其他行主要是为了处理一些默认参数和初始化的逻辑。

在第22行中，我们还会看到一个很重要的变量loadfactor，也就是负载因子。这是创建HashMap时的一个可选参数，用来帮助我们计算下一次触发扩容的阈值。假设 length 是table的长度，threshold = length * Load factor。在内置数组大小一定的时候，负载因子越高，触发resize的阈值也就越高；

负载因子默认值0.75，是基于经验对空间和时间效率的平衡，如果没有特殊的需求可以不用修改。loadfactor越高，随着插入的元素越来越多，可能导致冲突的概率也会变高；当然也会让我们有机会使用更小的内存，避免更多次的扩容操作。

总结

好，今天关于HashMap源码的分析就到这里啦。红黑树的部分我们下一节讲解TreeMap的时候再好好讨论，到时候你就会知道红黑树和链表之间的性能差距，也能体会到构造可快速访问键值对集合的另一种思路。

现在，如果不借助系统自带的HashMap，相信你应该也可以手写数据结构统计单词的数量了吧？正确的思路就是，根据全文长度大概预估一下会有多少个单词，开一个数倍于它的数组，再设计一个合理的hash函数，把每个单词映射到数组的某个下标，用这个数组计数统计就好啦。

当然在实际工程中，我们不会为每个场景都单独写一个这样的散列表实现，也不用自己去处理复杂的扩容场景。JDK的HashMap或者STL的unordered_map都是非常优秀的散列表实现，你可以好好学习一下相关源码。当然，你也可能会注意到我们在代码中没有任何处理并发的逻辑，这当然导致了线程不安全，在需要保证线程安全的场合可以用ConcurrentHashMap替换。

课后作业

前面我们有提到loadfactor，建议不要修改，那你知道什么时候需要修改吗？欢迎你在评论区和我一起讨论。

拓展资料

关于HashMap的put操作，美团总结了一个非常不错的流程图，可以参考。

06｜TreeMap：红黑树真的有那么难吗？

作者: 黄清昊

你好，我是微扰君。

上一讲，我们讲到如何利用散列表解决类似“文档中不同单词计数”的问题，并以JDK中HashMap的实现为例讲解了散列表背后的思想。

单词计数这个问题最基本的解决思路就是建立一个线性的符号表，每次计数的时候，遍历符号表就可以找到对应单词的计数器，做相应的累计计数操作就可以了。

为了更快地查找到单词的计数器，有两种优化思路，一种是我们上一讲学习的基于哈希表的思想，直接将符号表映射到一个连续线性的数组空间，从而获得O(1)的访问时间复杂度；另外一种思路就需要维护一个有序排列的符号表，JDK中的TreeMap就是基于这种思路。

试想，如果能够让符号表是有序排列的，我们查找的时候是不是就不用遍历每一个元素，而可以采用二分查找之类的手段了呢？当然也要尽量降低维护这个有序排列的数据结构所花费的代价。

那一种常见的用于实现有序集的数据结构就是红黑树，这也是JDK中TreeMap中Tree的意思。如果你有一定的Java开发经验，相信你一定会知道相比于HashMap，基于红黑树的TreeMap的一个显著特点就是其维护的键值对是有序排列的。

如果你一听到红黑树这个词，就有点慌张，觉得这不是自己能驾驭的，今天这节课就来帮你打消这个顾虑。

很多人觉得红黑树很难理解，其实很大程度是因为无论是本科的教学还是网上流传广泛的资料，大部分只是描述了红黑树的一些规则和结论，这就让学习者往往只能死记硬背红黑树是“由红色黑色节点构成的一种近似平衡树”、“不能有连续的两个红色节点”之类的规则，而不知道为什么有这样的规则，当然会非常复杂，很难理解。

但是如果我们追本溯源，从红黑树为什么被发明出来讲起，让你了解红黑树的本质，你就会发现它相当简单好理解了。

而且相信大部分程序员应该不会有什么机会要手写红黑树，在正经面试中也绝对不会碰到手写红黑树这样的题目，所以我们掌握红黑树的本质和特性就足够了。当然，红黑树的部分实现细节，我们在最后也会讲到，如果你学有余力可以尝试自己实现一下。

好啦，我们开始今天的学习。

二分查找树

要想轻松理解红黑树，我们需要先来简单复习一下二分查找树和平衡二分查找树的概念，因为红黑树就是一种近似平衡的二分查找树实现。

我们知道，在一个线性的链表里，如果要查找某个特定节点，唯一能做的事情就是从头到尾遍历节点，这带来了平均为O(N)的查找时间复杂度。但是如果数据存储在有序的二分查找树上，情况就大有不同。

二分查找树的二分是指什么呢？

举个非常简单的例子，你在便利店购物，有一个商品忘记扫码触发门禁警报，怎么从一堆商品里迅速定位呢？一个一个扫显然很慢，聪明的方法是将商品先分为两堆过门禁，找到触发的那堆再二分，继续下去，就能迅速找到啦。本质就在于二分每次可以排除掉一半的查询范围。

二分查找也是非常常见的算法。

比如在有序排列的数组中，因为数组访问任意下标元素的时间复杂度都是O(1)，我们就可以通过二分查找法，在O(logN)的时间复杂度里，快速定位任意元素的位置。我们之后有一讲会详细学习数组上的二分搜索算法。

而二分查找树，则是另外一种可以用来实现二分搜索的数据结构。具体来说，我们在一棵普通的二叉树上放置需要存储的符号表，并保证所有节点和其左右子节点满足这样的关系：每个节点的左节点要不为空，要不为比当前节点小的元素（符号）；每个节点的右节点要不为空，要不为比当前节点大的元素（符号）。

在这样的约束下，我们的查找只需要从根节点出发，比较当前节点和目标元素之间的大小，要么往左走，要么往右走；这是因为比当前节点大的元素一定在当前节点右边，反之则在当前节点左边，所以我们每次比较总可以排除左右子树中的一颗。

反复进行这样的搜索过程，直到当前节点已经是叶子结点，或者当前结点和目标元素相等为止。需要比较的次数最多为树的最大高度h，因此整个搜索过程的时间复杂度就是O(h)，显然，O(h)在大部分时候是一个比O(n)小得多的数。

比如要查找值等于6的结点，在原始的链表中我们可能需要遍历完整个链表，需要6次，而在二分搜索树中，我们只需要沿着根节点一路往右子节点遍历，一共比较3次即可。

但是这样的二叉树在极端的情况下也会退化成链表。

平衡二分查找树

比如下图也是一个满足约束的二分查找树，但所有的结点都在树的一侧排列。在这样带有极度偏向性的树中，我们查找节点的效率其实和链表没有什么区别，反而还用了更多的空间。

所以理想中，如果要实现具备良好查找特性的OrderedMap，我们需要同时保证树的有序性和平衡性，这里平衡性指的是，树上每个节点的左右子树的高度差要尽量小，最好不要超过一。比如前面的图就是一个很好的平衡二分查找树，这里的图就是一个退化成单链表的情况。

对于一个平衡的有N个元素的二分查找树，其高度可以近似认为是logN，所以查找的时间复杂度就是O(logN)。 这意味着一张有10000个元素的符号表，我们想要查询出任意一个元素，最多也只需要进行大约13次左右的比较即可。这显然是一个非常令人满意的结果。

哦，为什么高度是O(logN)呢？我们近似计算一下。

以满二叉树为例，也就是一颗除了叶子结点之外的结点都有两个子节点的二叉树。相信你很容易发现每一层能承载的容量范围都是上一层的一倍，那么，最多在第logN层，只把这一层的节点加起来，就足够承载需要存储的所有N个元素了。

而平衡树相比于满二叉树的最大高度显然是偏差有限的，高度应该也就是O(logN)这个数量级了。

那如何才能兼顾有序性和平衡性呢？这就是我们今天的主角红黑树出场的时候了。

红黑树

没错，红黑树就是这样一种兼顾了有序性、平衡性的自平衡二叉树的实现。因为它查找高效，成为许多语言实现内置ordered_map的首选，另外在Nginx的Timer管理、Linux虚拟内存管理等场景下，红黑树也都承担着重要的角色。

其实之前还有人提出了一些不同的平衡二分搜索树的实现，比如 AVL-Tree、2-3-4树等，但因为这些实现保持绝对平衡的代价比较高且往往实现复杂，并没有像红黑树这样遍布于各大需要有序键值对存储的场景。

那么红黑树到底是如何实现的呢？

要更好地理解红黑树的设计，首先你要理解红黑树的由来——红黑树本质上是对“2-3树”的一种实现。

那就让我们先一起来学习一下“2-3树”的实现，相信你学完一定会恍然大悟，红黑树这么奇怪的红色黑色的节点设计原来是这样来的。

2-3树

2-3树，也是一种平衡查找树的实现，思想很简单，为了让树能更好地平衡自己，我们除了普通的2节点之外，还引入了一种3节点，这让我们在平衡树高度的时候增大了很大的灵活性。看一个典型的2-3树例子。

在2-3树中，2节点，和普通二叉树的节点其实没有什么太大的区别，有一个键和两条链，分别链向左子树和右边子树。

而3节点，则在2节点的基础上增加了一个键，构成了一个有两个键和三条链的结构。下图是3节点的示意图，左链链向了小于a的节点，右链链向了大于b的节点，中间的区域则是a和b之间的节点。

在2-3树中搜索的过程和二叉树并没有太多的区别，只是遇到3节点的时候，多判断一次是否介于a、b之间即可。

设计有了， 我们看插入新元素的时候会发生什么。

在插入过程中，当我们查找到了某个叶子结点发现并不存在该键时，如果遇到了2节点，非常好办，直接加个键将该节点升级为3节点即可。比较麻烦的是遇到了3节点，因为我们已经不能再在该节点中直接多加一个键创造一个4节点了，怎么办呢？

其实办法也不难想，我们把当前的4节点多出的键向上转移。看图理解，比如要对下图中的2-3 树插入26节点，那首先会沿着根节点一路查询到“19 24”子结点，发现该节点为一个3节点。

那么我们首先将26放入该子节点，使之成为一个4节点，然后将4节点的中间键也就是24，提升到上一层，将其父节点替换成一个包含24的3节点。

如果原父节点也是一个3节点的话，我们就递归进行同样的操作直至根节点。最后，如果根节点也是一个3节点，我们就将根节点的中键提升到第一层，然后左右链分别链向原来根节点的左键和右键。以下图为例，b键就被独立地提升为新的根节点，左右节点指向a和c，而原4节点的中间两个链也分别成为a的右链和c的左链。

这样的操作可以保证整个2-3 Tree是一个真正意义上的平衡树。但是，因为它的实现引入了两种异构的节点，导致代码写起来相当复杂，并没有被广泛使用。

而红黑树，正是采用标准的二叉查找树节点附着上额外的颜色信息来表示2-3树的实现，每一个红色节点都和它的父亲节点一起，构成了一个3节点的模拟，这就是红黑树设计的本质。

红黑树

所以，我们再把红黑树的定义拿出来，红黑树是一个满足下述几个约束且所有节点要么为红色要么为黑色的二分有序查找树：

1. 根节点为黑色
2. 相邻的节点不能同时为红色
3. 每个节点从自身到各个能到达的叶子结点的所有路径中的黑色节点数量相等
4. 红节点只能作为左子节点存在（这是左偏红黑树特有的要求，我们以左偏红黑树为例讲解）

所有这些约束，都是为了保证每一颗红黑树和2-3 Tree是一一对应的，相信你看下面这颗“展平”的红黑树就能理解我在说什么了。

我们一起顺一下。

一个3节点有两个键、三条链，那我们完全可以把一个以红节点为左子节点的黑节点和子节点一起看成一个3节点。在下图中，上下两个图其实就可以认为是等价的。

我们再来看红黑树的3个普遍约束，你会发现很好理解：

1. 因为红色节点只是3节点的一部分，那对应到红黑树上，显然不会出现两个连续的红色结点；
2. 2-3树上，每个节点到叶子节点的数量一定是一样的，且每个节点对应到红黑树上一定包含且只包含一个黑色节点，所以红黑树每个节点到叶子结点路径中的黑色节点数量也必然是一样的。

唯一不一样的就是“根节点为黑色”。事实上，如果只是为了让红黑树保持平衡，我们完全可以抛弃这条规则。因为在2-3树中，我们也完全是可以用3节点作为根节点的。

对应到红黑树中，当根节点为红色，插入新节点后很可能为了使根节点到每条路径上的黑色节点数量相等，进行变色和旋转操作，最终根节点还是会变成黑色；既然如此，我们何不直接约束根节点必须调整成黑色，方便进行插入操作呢。

这样，我们就可以将每一个红黑树都映射成一个2-3 树，也因此就获得了2-3 树高效的平衡插入的能力，并保留了二叉树查找的简洁性。之后在理解红黑树的时候，如果你能时刻展平成2-3 树看待，一定会觉得，哦，红黑树的实现其实也没有想象中的那么困难。

最后，我们来看一些红黑树的基本操作，帮助你更好地理解红黑树和2-3树之间的关系。

旋转操作的实现

红黑树的所有实现细节，其实也都是围绕着2-3树的2、3节点的诞生和转移展开的，我们就以“插入方法”的实现来具体讨论（仍以左偏红黑树为例）。

红黑树中最基本的自平衡操作就是“旋转”，分为“左旋”和“右旋”两种。这两种操作主要用于处理在插入和删除时产生的右偏红节点或者连续的两个红色节点，通过调整红节点的位置，我们可以修复这些不满足约束的情况。

看“左旋”和“右旋”的具体操作。以左旋为例，本质上就是将某个3节点从以较小的键为根转移成较大的键为根，也就是从a为根转到b为根，当然同时需要把介于a和b之间的节点挂到a的右节点下。这样得到的新树就是以b为根结点的结构，并且在整个过程中，树的平衡性和有序性都没有被破坏，而原来不符合约束的右偏红节点已经被转移成“正确”的左偏红节点。

现在，我们根据插入时是在“2节点”还是“3节点”分开讨论，看看旋转操作具体是如何用于保证约束正确的。

2节点插入

首先我们来看针对“2节点”的插入，对应到红黑树的语境中，也就是针对普通黑色节点的插入。那显然只有两种情况，要么插入在左边，要么插入在右边。

如果插入在左边，非常简单，我们可以直接将2节点提升为3节点，由于新增的是左侧的红节点，完全不会破坏树的平衡性。对应到红黑树上，就是简单地将新节点放到查找到的最后一个节点的左边。

如果插入在右边，其实是一样的，我们也将2节点提升成一个不符合“左偏”规则的3节点，然后进行一次左旋转即可。由于插入的也是红节点，并不影响树的平衡性。

3节点插入

再看插入“3节点”，情况和操作都会复杂一些，我们根据插入的结点在3节点中左键的左侧、右键的右侧和两者之间分开讨论（还是左偏红黑树）。

最简单的情况就是插入在右键的右侧。和3节点分解的方式一样，我们只需要把中间的结点提升到上一层，并左右节点变成黑色即可。

而另外两种情况单独看这一个节点的变化，也并不复杂，只需要插入后进行一到两次旋转操作即可。

但这样是不是就完成了所有操作呢？ 并不是。

还有一个很大的问题我们没有处理，就是对3节点的操作中，我们虽然保证了“插入操作”对当前子树的“平衡性”没有被破坏，但由于将红色节点变成了黑色，就有可能导致当前子树的黑色节点高度比其他子树高了。

所以我们还需要进行一种叫做“颜色反转”的操作。

每次插入时，最后一步，除了将3节点的左右节点都变成黑色，同时要将3节点的中间键变成红色，这样当前子树到各个子节点路径中的黑色节点数量就不会有变化啦。

当然这个操作是需要递归进行的。因为父节点如果变成红色，也同样可能造成右偏红节点或者连续红节点这样不符合约束的情况，这其实等价于在父节点的父节点下插入了一个新的红节点。我们用类似的逻辑自下而上递归即可，递归的终点就是，遇到根节点我们将其拆分成3个“2节点”，或者遇到某个2节点我们将其升级为“3节点”时，我们就可以结束递归。

讨论好这些case，我们就可以在整个插入节点的过程中保证不破坏“有序性”和“平衡性”了。

删除的操作和插入类似，感兴趣的话，你可以课后自己通过纸笔模拟一下。整个过程确实比较复杂，许多细节过一段时间可能也会有所遗忘，不过只要你理解了红黑树的本质，是在二叉树基础上对“2-3 Tree”的实现，一定能迅速回忆起红黑树的约束条件。到这里，无论是准备面试，还是帮助你了解许多用到红黑树的中间件源码来说，都已经绰绰有余了。

操作红黑树的复杂度

最后我们来稍微计算一下红黑树上检索数据的复杂度。

前面我们提到，二叉搜索树检索键的最差时间复杂度取决于树的最大高度，是O(logN)，那求红黑树上检索数据的时间复杂度就等同于求红黑树的最大高度了。

我们将一个红黑树展平，并对应到2-3树。

2-3树是一颗平衡的树，由于每个节点只可能比二叉树多出一个键，在相同的高度下只会承载更多的元素，所以其高度不可能高于二叉平衡树的高度logN。

而红黑树只是把2-3树的每个“3-节点”拆成了一黑一红两个节点，其高度不可能比2-3树的最大高度翻倍还多；当然你也可以从“红色节点不能相邻”这一约束得出类似的结论，毕竟本质这两者是等价的。

所以，满足这样约束的红黑树的最大高度最多也就是2*logN，因此可以得到良好的O(logN)的查询复杂度。插入、删除复杂度显然也只和高度有关，我们最多需要进行常数倍于树的最大深度次旋转和颜色反转操作，所以复杂度也是O(logN)。

统计单词数量

好了，有了红黑树这样一个可用于动态、高效查找/删除/插入数据的数据结构，维护一个符号表也当然就很简单啦。如果依旧让你统计某文档中不同的单词出现的数量，我们就可以维护一颗以不同单词为键，出现次数为值的红黑树，在比较树的有序关系时只比较键的关系即可。

这样，每次遇到一个新的单词就在红黑树中查找相应的键是否存在，如果有存在就将对应节点的出现次数自增，如果没有存在就插入一个新的节点，让其值为1。

写成Java代码，你会发现和上一讲HashMap也没有有什么区别，只是将HashMap换成了TreeMap而已，底层机制就我们刚刚梳理的一样。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

import java.util.TreeMap;
import java.util.Map;
public class Test {
    public static void main(String[] args) {
        Map<String, Integer> map = new TreeMap<>();
        String doc = "aaa bbb ccc aaa bbb ccc ccc bbb ccc ddd";
        String[] words = doc.split(" ");
        for (String s : words) {
            if (!map.containsKey(s)) {
                map.put(s, 1);
            } else {
                map.put(s, map.get(s) + 1);
            }
        }
        System.out.println(map);
    }
}

假设有N个单词，则红黑树最大高度为logN，所以单词查询的最大时间复杂度为O(logN)，这样的查询一共会进行N次，所以整体的时间复杂度为O(NlogN)，从时间效率上来说确实是低于HashMap的。

总结

红黑树的查询、插入、删除的时间复杂度都非常良好且稳定，广泛运用于各种中间件里，还是非常值得掌握的。

当然红黑树是一个比较复杂的数据结构，如果直接从定义和几条绕口令一样的约束入手，很不好学。不过只要掌握它的本质，其实很简单。只要搞清楚，红黑树本质是2-3树在二叉树上的一种模拟，通过旋转操作完成2-3节点的合并和分裂，从而在不改变二叉树节点结构的前提下，保证二叉树的有序性和平衡性。

相信你理解了这一层关系，就可以对红黑树有一个比较不错的掌握了；相比于只是死记硬背红黑树的各种特性和复杂约束，追本溯源显然是一个更好的选择，这在你学习技术知识乃至其他领域，都是值得推荐的学习方式。

课后作业

那什么时候我们该用TreeMap什么时候该用HashMap呢？欢迎你留言与我一起讨论哦！

拓展阅读

给感兴趣的同学推荐一下我觉得最好的学习材料《Algorithms》，作者是左偏红黑树的发明者。

07｜堆：如何实现一个高效的优先队列？

作者: 黄清昊

你好，我是微扰君。

上一讲学习了基于红黑树的ordered_map的实现，今天我们来介绍另外一种有趣的树，heap，也就是堆。堆的应用非常广泛，我们常说的堆排序的堆就是指这种树状数据结构，除此之外还可以用来解决诸如TopK，或者合并多个有序小文件之类的问题。

堆也是我们最后一个基础数据结构容器-优先队列的常见底层实现方式。

优先队列

你一定还记得之前讲过的线性数据结构queue吧，也就是队列，一种先进先出的数据结构，生活中这种先进先出的场景也很常见，我们当时举了一个在餐厅排队取号的例子，先来的人一定会先取到号，也先被叫到号，这完美地符合队列的语义。

那如果我们现在希望允许一部分人插队呢？比如在医院中，大部分病人都有序的挂号排队，但这时候如果来了一个重症病患，我们就很可能需要破坏先进先出的规则，让医生优先诊断治疗这位病患。这种场景中的队列，我们就可以定义为“优先队列”。

优先队列中的每个元素，我们会赋予它一个优先级，优先级相同的元素我们还是遵循先进先出的原则，但一定会保证队列中优先级更高的元素先出队，即使它进队时间更晚。比如例子中的重症病患来的并不早，但依旧得到了医生的优先治疗。

对于这种优先队列，我们应该如何高效地实现呢？

如何实现

其实也很容易想到不止一种方案。

比如，一种比较暴力的思路可以是：我们依旧用线性容器存储元素。在入队的时候，我们不关心优先级的影响直接按顺序存入容器中，出队的时候，则遍历容器找到最高优先级的元素出队。

由于入队的时候没有对优先级做任何处理，所以出队的元素显然可能在线性容器中任意一个位置，基于之前所学的知识，遇到节点删除的场景，用链表显然比用动态数组有更好的时间复杂度。但即使如此，每次出队时我们也需要遍历链表，所以时间复杂度为O(N)。

那与之相对的，另一种同样基于链表的思路也可以是，我们每次在入队的时候进行一些额外的调整，使得整个队列一直满足优先级更高的元素在更前面的约束，这样出队的时候就比较简单。当然这样会导致入队的时候都需要进行一次类似于插入排序的操作，最差情况下也会要遍历完整个链表，时间复杂度同样为O(N)。

假设出队和入队操作数量相等，均摊下来，每一次操作的时间复杂度就是O(N)。

看示意图辅助你理解，队列里的数字代表权重，a、b、c、d、e 代表着入队的值，我们假设入队顺序和值的字典序是一致的。上面的队列画的就是入队的时候就按照优先级排好序的情况，所以直接从队尾出队即可；下面的队列是入队的时候直接放到队尾，出队的时候要按照优先级取出元素。

那有没有一种入队和出队都相对来说比较高效的方式呢？答案是肯定的，只是我们需要抛弃线性的存储结构。

不知道你是不是也想到了上一章讲的红黑树。其实你的想法是对的，红黑树当然是可以用来实现优先级队列的一种方式，我们建红黑树的时候以优先级为key作为排序依据即可。入队的时候可以直接push入队，出队pop的时候先从树中找到优先级高的，也就是树的最右节点，然后移除即可。

这些操作的复杂度都是O(logN)，所以出队和入队的复杂度自然也就是O(logN)。所以，基于红黑树的优先队列复杂度均摊下来，相比于之前基于线性表的O(N)复杂度，显然更胜一筹。

但是由于我们不会进行类似“找出优先级第3高的元素出队”这样的操作，其实并不需要一直维护完全的顺序信息，只是需要能在每次出队时，找到优先级最高的元素即可。那有没有更合适的选择呢？

相比复杂的红黑树，简明的“二叉堆”就是这样一种特别适合用来动态维护一组元素中最大或者最小值的数据结构，它也是各大语言实现优先级队列的首选。

二叉堆

二叉堆，这个数据结构是1964年斯坦福大学教授Robert W．Floyd和J．Williams在发明堆排序的时候提出，之所以想介绍一下它的历史，主要是因为Robert W．Floyd是一个文科出生，后来自学计算机科学，逆袭成为斯坦福终身教授的大佬。我想这多多少少能证明热爱的力量，也能给可能是非科班出身的你我一些信心吧。

好啦回归正题，“二叉堆”，也就是binary heap，顾名思义，这个数据结构也是建立在一种特别的二叉树上的。

它主要有两个约束：

1. 二叉堆是一颗满二叉树。也就是说，除了最后一层外的每一层都没有空节点，且最后一层所有节点靠左排列，不存在从左到右中间有某些节点为空。
2. 二叉堆中的每个节点和其子节点都有一样的偏序关系，要么大于要么小于，这两种情况分别对应大顶堆和小顶堆。所以大顶堆就要求堆中所有节点的值，一定大于其左右子树中的任何一个节点的值，也就是说顶部的节点一定是最大的，故称为大顶堆。小顶堆就正好相反。

有了这样的约束，可以保证根节点要么是最大的要么是最小的，也让我们在出队入队的操作里调整的成本很小，整个过程有点像冒泡排序的感觉，我们马上讲解具体的细节。

比如下图中左边就是一个二叉堆，但中间因为不满足满二叉树的约束，就不是一个二叉堆；右边因为不满足完全的有序关系也不是一个二叉堆。

为什么这么设计

那现在我们来看看，这样的数据结构设计对于维护优先队列有什么帮助呢？和红黑树一样，二叉堆这样的树状结构同样暗含了关键的顺序信息，而我们核心就是利用这样的信息，在插入时进行更少的操作，而避免像线性表那样做从头到尾的顺序遍历。

为了实现优先级队列，我们会用优先级，priority，来作为二叉堆节点间大小比较的依据。假设我们始终希望出队的是优先级更高的元素，那可以采用大顶堆作为优先队列的底层实现，这样每次只需要从顶部取出元素即可获得优先级最高的元素。

当然很重要的一点是，取出元素后显然会在树的顶部产生一个空位，我们需要进行一定的操作使得大顶堆的性质得以保全。

同样，为了享受直接从顶部取出优先级最高元素的便利，我们在插入元素时也要让二叉堆的性质得以保持。

幸运的是，正是因为二叉树是一个满二叉树，其高度约等于LogN，其中N为优先队列中元素的个数。而第二条约束父子节点之间的有序关系，让我们每次做pop和push操作，只需要经过最多二叉树高度次的交换调整即可保持堆的所有特性。

这样，我们就得到了一个入队和出队操作复杂度都为O(LogN)的数据结构，虽然均摊的时间复杂度和红黑树是一样的，但实现的难度却要小很多，所以今天我们就结合源码来讲解。

PriorityQueue的实现

以JDK14中的PriorityQueue为例（后面简称PQ），我们来分析一个生产环境中的优先队列要怎么基于堆实现。

先来看看JDK中优先队列数据结构的基本定义，我们会讲一些重要的成员变量和方法。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

public class PriorityQueue<E> extends AbstractQueue<E>
    implements java.io.Serializable {
      /**
       * Priority queue represented as a balanced binary heap: the two
       * children of queue[n] are queue[2*n+1] and queue[2*(n+1)].  The
       * priority queue is ordered by comparator, or by the elements'
       * natural ordering, if comparator is null: For each node n in the
       * heap and each descendant d of n, n <= d.  The element with the
       * lowest value is in queue[0], assuming the queue is nonempty.
       */
      transient Object[] queue; // non-private to simplify nested class access
      
      /**
       * The number of elements in the priority queue.
       */
      int size;
      
      /**
       * Inserts the specified element into this priority queue.
       *
       * @return {@code true} (as specified by {@link Queue#offer})
       * @throws ClassCastException if the specified element cannot be
       *         compared with elements currently in this priority queue
       *         according to the priority queue's ordering
       * @throws NullPointerException if the specified element is null
       */
      public boolean offer(E e) { ... }
      
      /**
       * Increases the capacity of the array.
       *
       * @param minCapacity the desired minimum capacity
       */
      private void grow(int minCapacity) { ... }
      
      public E peek() {
          return (E) queue[0];
      }

      /**
       * Inserts item x at position k, maintaining heap invariant by
       * promoting x up the tree until it is greater than or equal to
       * its parent, or is the root.
       *
       * To simplify and speed up coercions and comparisons, the
       * Comparable and Comparator versions are separated into different
       * methods that are otherwise identical. (Similarly for siftDown.)
       *
       * @param k the position to fill
       * @param x the item to insert
       */
      private void siftUp(int k, E x) { ... }
      
      public E poll() { ... }
    }
}

首先要看堆元素在内存中到底是怎么存储的。可以看到：

• 成员变量size，它显然用于表示优先队列中元素的大小。
• 成员变量中有一个叫queue的数组，这就是堆中元素存放的地方。

你这个时候可能就会有些疑惑了？诶？那我们堆的树状结构到哪里去了，为什么还是一个线性的存储结构呢？

其实，在JDK中的PQ实现里，并没有和我们想象中一般的树那样采用节点+指针来实现树状数据结构，而是用了内存上连续的数组来模拟。代码中的queue数组就是用来表示堆这一树状结构的成员变量。

这么讲可能你不是很理解，我们看下面的图。图中画的就是一个大顶堆的示例，上下分别代表堆的树状结构本身和存储到queue数组中的样子。你可以通过元素的值来判断两种表示间的对应关系。

相信你也发现了，图中的数组就是按照对树做层序遍历的方式依次排列的，所以数组中下标为0的元素就是堆顶的元素，我们也用箭头将数组元素中的父子关系都标记出来了。由于二叉树每一层元素个数都是上一层两倍的特性，你会发现，queue[k]节点，它的左子节点为queue[2k+1]，右子节点为queue[2k+2]。

事实上，不止二叉堆，其实一般的二叉树也是可以用数组表示的，如果你做过LeetCode上相关试题的话，也会经常碰到类似的表示方式。大部分时候，用数组表示树写起来比较简单，不需要引入类似指针的概念，也不用定义树节点的类或者结构体。

但如果二叉树不是满的，数组中会有大量的空值，非常浪费空间。而堆本身满二叉树的特性，则让我们可以选择用数组作为底层二叉树实现而不至于浪费大量的内存，这就是JDK中为什么可以使用数组作为底层存储的原因。

堆的操作

有了底层的数据表示，下面我们来了解一下堆的两个重要操作，插入和删除。

首先看堆的插入操作，也就是 offer 方法，对应的代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

/**
 * Inserts the specified element into this priority queue.
 *
 * @return {@code true} (as specified by {@link Queue#offer})
 * @throws ClassCastException if the specified element cannot be
 *         compared with elements currently in this priority queue
 *         according to the priority queue's ordering
 * @throws NullPointerException if the specified element is null
 */
public boolean offer(E e) {
    if (e == null)
        throw new NullPointerException();
    modCount++;
    int i = size;
    if (i >= queue.length)
        grow(i + 1);
    siftUp(i, e);
    size = i + 1;
    return true;
}

该方法接收一个待插入的元素e，在18行中将堆的大小增加1，这个非常直观。有两个函数详细解释一下，grow和siftUp。

grow函数

grow函数用于扩展数组的大小。

我们用数组模拟堆，就必然要给出一个数组的大小，这也就限制了二叉堆的最大大小。但这在使用过程中并不方便，我们没有办法在各种场景下都准确预估堆所需使用的最大元素个数。所以和STL的vector一样，JDK中的PriorityQueue，通过内置的grow函数和扩容机制解决了堆动态大小的问题。grow函数具体做的事情我们最后再讲。

所以在offer方法里，如果发现当前插入的元素已经超过了内置queue数组的容量，我们需要进行扩容操作，这就是第15-16行代码所做的事情。

siftUp函数

现在我们来看siftUp的过程，具体的插入操作其实就隐藏在了siftUp这个函数中。

之所以叫siftUp，就是因为这个插入的过程是自下而上的。我们结合具体例子，来讲解siftUp的过程，先搞清楚思路，再来看对应的代码就很好理解了。

假设有这样的一个二叉堆，我们想要插入一个新的元素。

可以先将该元素插入到数组的尾部，也就是堆中的最后一个位置，这是一个O(1)的操作，因为不涉及任何元素的移位。然后要做的，就是将这个元素一路往上交换使得二叉堆可以保持自身的2条特性。

由于我们是将新的元素放到堆的尾部，没有空开任何一个子节点，所以只是进行元素的交换，并不会破坏堆是满二叉树的特性，因而我们只需要关注父节点比所有子节点大这一特性。

所以将该节点不断和其父节点比较，如果比父节点大，将该节点和父节点交换，并继续和该节点新的父节点进行比较。由于原来的父亲节点一定比其左右子节点（其中一个是这次调整的节点，另一个是他的兄弟节点）都大，所以将其和父亲节点交换位置，一定也能保证该节点比其兄弟节点大。

而只要发现该节点已经比父节点小了，我们就可以结束这次的比较、交换之旅，堆已经重新满足了特性。

结合示意图理解，比如在堆的最后一个位置插入92。

我们要先把92和47比较，发现92比47大的时候，互换位置；然后继续比较，会交换92和76的位置，关注父子节点的大小要求特性，因为76本身比52大，所以交换后一定也能保证92比52。这就是堆siftUp的过程。

这里再稍微补充两句，堆和红黑树不一样，我们在整个过程中只关心父子节点之间的大小关系，而不用在意兄弟节点之间的大小关系，比如原本4（47）节点是比5（52）节点小的，现在4节点（76）比5节点（52）大了，这并不会破坏堆的性质。这也是堆之所以调整比红黑树简单很多的地方。

了解这个过程，对应的代码就非常好懂了。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

private void siftUp(int k, E x) {
    if (comparator != null)
        siftUpUsingComparator(k, x, queue, comparator);
    else
        siftUpComparable(k, x, queue);
}



private static <T> void siftUpComparable(int k, T x, Object[] es) {
    Comparable<? super T> key = (Comparable<? super T>) x;
    while (k > 0) {
        // 计算父节点的下标
        int parent = (k - 1) >>> 1;
        Object e = es[parent];
        // 比较当前节点和父节点的关系 如果当前节点优先级更高，我们可以直接结束比较
        if (key.compareTo((T) e) >= 0)
            break;
        // 交换节点
        es[k] = e;
        k = parent;
    }
    es[k] = key;
}

因为PriorityQueue里存放的可以是任何元素，用户也可以自定义比较关系，所以Java的PQ实现里引入了比较符和泛型的概念。就是Java中泛型相关的语法，siftUpUsingComparator 则给了用户自定义比较符的自由；不熟悉的同学可以自行搜索相关资料了解。

我们直接看siftUpComparable方法，也就是元素类型自带比较方法的情况。

我们要比较这个新节点和父节点的值，那怎么定位父节点呢？回忆一下前面讲queue底层存储的时候讲过的父子节点关系，对queue[k]来说，它的父节点一定是queue[(k-1)/2]。当然k需要大于0，否则k就已经是根节点了。

所以11-18行的循环就是在做siftUp中和父节点比较并交换的操作，第14行就是对应key比e大，也就是子节点比父节点大的情况，此时我们就可以退出循环了。这也说明PQ默认情况下实现的是小顶堆。

删除的poll操作

好，下面看PQ中第二个主要操作，poll操作，这是我们返回并删除堆顶元素的操作，也就是从优先队列中取出最高优先级元素的操作。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

public E poll() {
    final Object[] es;
    final E result;
    // 取出堆顶元素
    if ((result = (E) ((es = queue)[0])) != null) {
        modCount++;
        final int n;
        // 其实就是要将最后一个元素放到顶部
        final E x = (E) es[(n = --size)];
        // 将最后一个元素置空
        es[n] = null;
        // 进行siftdown操作
        if (n > 0) {
            final Comparator<? super E> cmp;
            if ((cmp = comparator) == null)
                siftDownComparable(0, x, es, n);
            else
                siftDownUsingComparator(0, x, es, n, cmp);
        }
    }
    return result;
}

重点理解第7行和关键的siftDown函数。

要取出堆顶元素，数组第一个元素就会有所空缺。基于siftUp的想法（还是用刚才的大顶堆），我们第一反应很容易想到，直接从空缺的根节点开始，找其左右子节点中大的一个提拔到当前空缺的位置，然后依次找新的空位左右子节点哪个位置可以提拔上来，直到没有子节点为止。

这个思路确实很自然，但有一个比较大的问题就是，如果我们在第一次比较的时候选择了左节点提升上来，当右节点并不为空时，最后得到的树一定不是一个满二叉树了，这就破坏了堆的基本性质。

那怎么做呢？一个比较巧妙的想法就是和代码里第7行一样，将根节点删除后，我们把二叉堆中最后的元素提到根节点的位置，这样又可以保证新的二叉树是一颗满二叉树了，然后要做的就和前面所说的一样，比较+交换。

看大顶堆的例子，现在要删掉最大的元素92，怎么做呢？首先把堆中最后一个节点也就是47提到堆顶的空位，然后依次比较左右节点；47比90、79都小，但90更大，所以我们用47和90进行交换。同理47和76也需要交换。这样我们就完成了优先队列优先级最高元素出队的操作。

siftDown函数

回到代码，这里说的比较+交换的过程就是由siftDown这个函数完成的：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

private static <T> void siftDownComparable(int k, T x, Object[] es, int n) {
    // assert n > 0;
    Comparable<? super T> key = (Comparable<? super T>)x;
    int half = n >>> 1;           // loop while a non-leaf
    while (k < half) {
        int child = (k << 1) + 1; // assume left child is least
        Object c = es[child];
        int right = child + 1;
        if (right < n &&
            ((Comparable<? super T>) c).compareTo((T) es[right]) > 0)
            c = es[child = right];
        if (key.compareTo((T) c) <= 0)
            break;
        es[k] = c;
        k = child;
    }
    es[k] = key;
}

5-14行就是之前所说的父亲节点和左右子节点循环比较并交换的过程。k的左子节点下标为(k<<1)+1，右子节点下标为左子节点+1，这就是6行和8行代码的意义。在发现左右子节点都比根节点小之后，同样可以退出循环，否则和左右节点中小的一个交换即可。

嗯，如果只是要查看优先级最高的节点而不用出队，当然是非常简单了，我们直接取堆顶元素即可。也就是之前PQ定义的这几行代码：

1
2
3

public E peek() {
    return (E) queue[0];
}

扩容机制

那，最后我们来讲讲PQ的扩容机制，这是PQ虽然用数组作为底层存储，却不用限制优先队列大小的核心。

1
2
3
4
5
6
7
8
9
10
11
12
13
14

/**
 * Increases the capacity of the array.
 *
 * @param minCapacity the desired minimum capacity
 */
private void grow(int minCapacity) {
    int oldCapacity = queue.length;
    // Double size if small; else grow by 50%
    int newCapacity = ArraysSupport.newLength(oldCapacity,
            minCapacity - oldCapacity, /* minimum growth */
            oldCapacity < 64 ? oldCapacity + 2 : oldCapacity >> 1
                                       /* preferred growth */);
    queue = Arrays.copyOf(queue, newCapacity);
}

本质和之前讲的STL中的vector扩容思想如出一辙，就是将当前的数组搬到一段更大的连续数组中，新的数组容量为newCapacity，旧的数组容量为oldCapacity。

它俩的关系在第11行中可以看出来：如果原来的数组已经比较大了，那新数组的大小是旧数组大小的1.5倍，否则是2倍再+2。至于为什么要成倍增长，你可以回看vector的章节，都是为了保证良好的均摊时间复杂度。

用于计算容量的newLength的方法还有个参数用于保证最小的扩容大小，如果你感兴趣可以自己研究一下。

总结

学完了JDK中PriorityQueue的主要实现机制和源码，现在你有没有掌握用数组模拟树的技巧呢？如果让你手写一个堆应该也不难了吧，这也是面试中算法的常考题目。

核心就是数组中父子节点的下标关系。下标为k的节点queue[k]，它的左子节点为queue[2k+1]，右子节点为queue[2k+2]。理解了这一点，去实现siftDown和siftUp操作，相信对你来说不在话下。

当然JDK为了提供一个更通用的优先队列，也引入了泛型，也提供了动态扩容的能力，这些内容和动态数组的实现非常相似的，你可以回去复习并尝试手写实现一下。

课后作业

有了堆，实现优先队列的语义其实并不难。但有个问题想问问你，你觉得Java的优先队列是否真正保证了优先队列语义呢？比如优先队列中相同优先级的元素，是否应该要保证先进先出的特性？如果，JDK中的没有这个保证，但我们却有这样的需求，你觉得应该如何改造呢？

这是一个开放问题，欢迎你在留言区参与讨论。如果你觉得这篇文章对你有帮助，也欢迎你转发给身边的朋友一起学习。我们下节课见～

08｜外部排序：如何为TB级数据排序？

作者: 黄清昊

你好，我是微扰君。

之前已经学习了常用数据结构的工业级实现（包括动态数组、双向链表、双端队列、栈、哈希表、红黑树、堆），从今天开始，我们来讲讲一些经典的算法思想在工程实践中的应用。

那讲哪些算法呢？我们都知道算法是一个很大的命题，也有很多分类的方式，比如就有人总结过非常经典的五类常用算法：贪婪算法、动态规划算法、分治算法、回溯算法以及分支限界算法。力扣上的每一道算法题也有相应标签，你感兴趣的话可以到题库看一下。

不过有些算法可能只会在特定的场景下被特定的中间件所使用，比如布隆过滤器、前缀树等等，我们在后面的章节结合实际的系统或中间件来讲解；有一些算法思想应用更为广泛，我们会在这个部分学习，所以基础算法篇主要包括了贪心、分治、二分的算法思想，也会涵盖排序、搜索、字符串匹配这些更为常见的应用场景。

今天就让我们从经典的排序问题开始讲起吧。

排序

排序，应该是我们学习算法和数据结构时最早就会学习到的几个算法问题，按时间复杂度这个标准大体可以分为O(n^2)、 O(nlogn) 、O(n) 三大类。

O(n^2)的选择排序、冒泡排序、插入排序，这些常用的算法相信你应该非常熟练了；几种O(n)的算法在工程中其实也都有实际应用，你也可以自己在网上搜索资料了解学习，最好再找几道相关算法题做一做加深印象。

O(nlogn) 的三个常见算法从概念上看也不难理解，但细节实现起来还是有一些复杂度，需要花点时间刻意练习，是面试中相对重要的算法考点。

其中归并排序和快速排序的常用写法都可以采用递归的方式实现，背后是分治的算法思想，也就是分而治之，把大问题递归拆小然后递归得出结果。

而堆排序的思路和实现，在上一讲优先队列中我们详细讲过，相信你现在应该很容易用Java的PriorityQueue实现堆排序，主要思路其实就是建立一个堆，借助堆动态调整的能力，只需要将所有待排序元素依次入队，再依次出队直到堆元素全部出队为止；比如在小顶堆中，每次出队的都是当前最小的元素。

但只是能写出这样的排序代码，往往并不足以让你解决真实世界的工程问题。

举个例子

比如有这样一个基于真实场景的经典面试题：假设现在有1TB的任意文本，请问如何能将其中出现的单词按照字母序排列，得到一个新的文本？

这个问题，你可以回答好吗？

你也许会觉得很简单呀。我们就直接用堆排序，建立一个小顶堆，然后遍历整个文本进行分词，将每个单词都依次push进堆，最后再逐一出队输出到一个文本，最后就可以得到一个按字典序升序排列的文本了。

这当然是一个正确的思路，但是，你忽略了一个至关重要的问题，就是我们的内存可能没有这么大！

这也是面试官考察这个问题的核心知识点，1TB的数据量级，意味着绝对不可能一次性将所有的数据都放入到内存中。而大部分单纯考察算法的面试题，其实都是在一个比较理想的环境下的，比如和排序相关的问题，绝大部分题目肯定会给你一个数组去存放需要排序的元素，隐含了内存可以一次性将所有数据读入的条件。

但在实际工作中，我们经常会遇到内存中放不下所有数据的排序场景。

早期可能因为内存的容量确实很小，而现在更多是因为我们需要存储的数据越来越大了，甚至不只是内存放不下，单机的硬盘可能也不够了，需要考虑分布式环境下的排序问题，比如在一个分布式数据库中进行超大表的order by操作，这往往需要花费几分钟甚至几小时的运算才能完成。

好言归正传，我们就借助这道经典面试题一起来看看如何解决大量数据的排序问题。

TB级数据排序

这个经典的算法问题我们一般称之为外部排序，这里的“外”指的其实就是外部存储的意思。

读写较慢的外存，相比快速但昂贵的内存而言，有着更低廉的成本，通常是硬盘，它可以存放更大的数据。当我们不能直接在内存中进行排序，而需要借助外存去处理极大量数据的排序时，就需要使用外部排序算法了。

如果遇到这样的面试题，首先可以来向面试官确认一下已有的硬件环境，比如面试官可能会告诉你，你现在有1GB的内存可用。那么我们知道整个1TB的文件，至少要读1024次才能遍历一遍，所以直接在内存里排序显然是不现实的。

但是文件其实是可以一部分一部分读的，如果内存中一次放不下全部的数据，也许我们可以将文件分成若干段，分别读入内存中，并采用常见的内排序算法（比如堆排序），对这段可以在内存中存储的段落进行排序；得到若干个有序的文件段后，最后通过一些合并的方式，得到整体有序的文件。

当然在这个过程里会有大量的中间结果，比如那些有序的文件片段，这些我们都需要借助外存存储，这个思路就是最常见的一种外部排序的方式。

其实你会发现我们刚刚描述的想法和归并排序如出一辙，归并排序也是常用的外排实现方式，只不过我们在学习它的时候，一般都是针对数组，也就是在内存中排序的场景。

外部排序

好我们用严谨的语言来重新描述一下基于归并思想的外排过程，整体分为两个阶段：

• 部分排序阶段

我们根据内存大小，将待排序的文件拆成多个部分，使得每个部分都是足以存入内存中的。然后选择合适的内排序算法，将多个文件部分排序，并输出到容量可以更大的外存临时文件中，每个临时文件都是有序排列的，我们将其称之为一个“顺段”。

• 归并阶段

我们对前面的多个“顺段”进行合并，思想和归并排序其实是一样的。以2路归并为例，每次都将两个连续的顺段合并成一个更大的顺段。

因为内存限制，每次可能只能读入两个顺段的部分内容，所以我们需要一部分一部分读入，在内存里将可以确定顺序的部分排列，并输出到外存里的文件中，不断重复这个过程，直至两个顺段被完整遍历。这样经过多层的归并之后，最终会得到一个完整的顺序文件。

举一个简化的例子，配合示意图理解。

假设现在有含有1000个记录的文件，而内存最多只能读取100个记录。那么我们要做的第一步就是先把1000个记录拆成十个文件，每个文件有100个记录，读入后在内存中排序得到10个有序的临时文件，并输出到外存也就是硬盘中。

然后我们进行多次归并操作，每次都把相邻的文件合并。在这个例子中可以看到只需要进行4轮归并，就得到了一个最终有序的文件。

运行时间

那整个过程里，运行时间主要和哪些因素有关呢？

在第一个阶段部分排序中，由于内存可以装下每个顺段的所有元素，所以几种主流的O(nlogn)的算法都是可以的，其中快速排序在大部分场景下是最快的，因此我们可以首选快速排序。

比较复杂的是归并阶段。因为内存不足以装下所有需要排序的元素，所以O(nlogn)的堆排和快排都已经没办法被应用在外排的场景中了，但基于分治思想的归并排序却依然可以很好地发挥作用。

而且相比很多其他排序方式比如选择排序、冒泡排序，归并排序O(nlogn)的复杂度已经是理论上相当好的复杂度了。当然在一些特定场景下我们也可以用一些线性排序算法比如桶排序来解决外部排序问题，感兴趣的同学可以自己搜索了解一下。

但是和内排中的归并排序不同，外部排序场景下，我们还有个非常大的时间消耗就是IO，也就是输入输出。

相比内存中的读写操作，在磁盘中的读写是一个慢得多的过程，两者之间可能有千倍以上的时间开销差距。所以考虑外排效率时，非常重要的一点就是我们要尽量减少从磁盘中读取数据的耗时，而这主要关系要访问多少次外存。

那我们在外存中需要读取多少次数据呢？从图中其实可以看出来，每一层我们读取外存的数据总量其实是一样的，本质上就是将所有的数据都遍历一遍。

而内存大小是一样的，所以每一层中读取外存的次数也就是一样的，那么显然关系我们读取次数的多少主要就取决于所需归并的层数了。因此，我们要做的事情就是让归并的层数越低越好。

怎么样做到这件事呢？答案很简单也很直观，就是增加更多的归并路数或者降低初始的顺段数量。

如何降低归并层数

我们先算出归并层数，以2路归并为例，每次合并两个连续的顺段，如果上一层有n个顺段，到下一层就会有n/2个顺段，每一层的顺段都会减少一半，直至只剩一个顺段，也就是需要的排序结果。因而，假设初始一共有n个顺段，那么我们大致需要log2n层。

同样的道理，如果进行k路归并，每一层的顺段数量都会变成上一层的1/k，所以就大概只需要logk(n)层即可完成整个归并。比如一个5路归并的例子。

所以，为了增加归并路数，也就是尽量增加k。

另外为了降低初始n个顺段的数量，我们会做的事情也很简单，就是在第一次进行逐段内排序的时候尽可能多地将数据读入内存中并进行内排。

但是增加k的大小，其实也会导致每次归并的时候合并的成本变大，一个显著的问题就是在k路归并中，我们需要从k个元素中选择出最小的元素，代价比2路归并的更高。如果用最暴力的方式，遍历k个元素，每次选择最小的元素的过程将产生O(k)的时间复杂度，这一定程度上会抵消前面通过增加k减少磁盘IO所带来的时间提升。

但是我们仔细想想这个问题，选择k个元素中的最小元素，显然有优于暴力遍历O(k)复杂度的算法。比如，上一讲介绍的堆就可以解决这个问题。

而败者树，则是解决从k个元素中选取最小元素并可以动态更新的另一种方法，也是更广泛运用于多路归并中的算法，我们来学习一下它的思路。

败者树

败者树也被称为，淘汰赛树，也就是Tournament Tree，思想来自体育比赛。

我们知道在淘汰赛中，每一场比赛都有两个参与者，其中胜者可以晋级下一轮。整体可以画成一颗树的形状，如果看过足球比赛，相信你对这个图一点也不陌生。

败者树算法就是基于这一思想实现的，我们用叶子节点存储所有待比较的元素，对叶子结点两两比赛，在它们共同的父节点中存储失败者；然后对获胜者节点再两两比较，得到更上一层的败者，取出胜者继续往上比较，这个过程和归并的思路其实也是比较相似的；这样一层一层往上比较，最后就可以得到一颗锦标赛状的树。

因为除了叶子结点外的每一层的父节点存储的都是其子节点中的失败者，所以我们称其为败者树。根节点我们会稍微做一点特别的处理，除了在根结点存储失败者，同时，我们在根节点之上会悬挂上整棵树的最终获胜者。

我们可以给刚刚的例子画出对应的败者树，大概就是下面这个图的样子。其中根节点上的方框存储的就是整个树的胜者，也就是1，它一定是所有元素中的最小值，和锦标赛的冠军是一个意思。

至于为什么在节点中存储的是失败者，而不是获胜者呢？就留给你做课后思考题啦。事实上，在父节点中存储获胜者的树也是存在的，和你想的一样，我们也称之为胜者树。

和堆一样，我们也会需要对败者树进行类似于出队的操作；在上面的例子中，就是我们需要将1从败者树中取出，寻找下一个最小的元素。

这里有两种情况，一种是我们取出1之后，用一个新的元素替代1，另一种就是取出1之后不再添加新的元素，分别对应某一路元素被取出之后仍有元素未取完，和该路元素已经全部取出的情况。在这两种情况中，我们其实都只需要对整个树重新比赛一次即可，只是在第二种情况里，我们会用一个无限大的数字替换1，因为无限大的数字一定不会在这次重赛中胜出。

我们用一个具体的例子来讲解一下这个过程，假设取出1之后，我们用8来替换1原来的位置。

由于每个父节点存储的都是两个子节点中的失败者，所以我们只要用更新的值和父节点的值比较，也就是和之前两个子节点中的失败者比较。

因为原来的获胜者已经被取走了，这里的父节点现在存储的其实就是这颗子树中原来的亚军，如果我们要看这次新来的选手能否能成为新的冠军，只需要和原来的亚军进行一次比较即可；当然这次比较结束后，两者中的胜者还需要到更上一层继续比较。

这个过程和运动员从市队、省队一路选拔到国家队参加奥运会的过程也是颇为神似的，希望这个例子能帮助你更好地理解败者树的工作机制。

整个过程写成伪代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

// 定义节点的value为具体的值；index为每一路数据的索引
// 用index.next可以取到每一路某个元素的后继元素。

function merge(L1, …, Ln)
    buildTree(heads of L1, …, Ln)
    while tree has elements
        winner := tree.winner
        output winner.value
        new := winner.index.next
        replayGames(winner, new) // Replacement selection

function replayGames(node, new)
    loser, winner := playGame(node, new) // 比较新节点和老节点的大小
    node.value := loser.value // 当前节点记录为比较中失败的节点
    node.index := loser.index
    if node != root
        replayGames(node.parent, winner) // 胜者跟父节点继续比较
 
function buildTree(elements)
    nextLayer := new Array()
    while elements not empty // 自下而上而上建树
        el1 := elements.take()
        el2 := elements.take()
        loser, winner := playGame(el1, el2) // 两两比较
        parent := new Node(el1, el2, loser) 
        nextLayer.add(parent) // 将获胜者放入上一层 继续
    if nextLayer.size == 1 // 只有根节点 直接返回即可
        return nextLayer 
    else
        return buildTree(nextLayer)

时间复杂度

最后我们来分析一下败者树的整体时间复杂度，我们假设一共有k个节点。首先，初始化的过程需要花费 O(k) 的时间，因为对于k个子节点，一共只需要进行k-1场比赛即可完成淘汰赛。

然后在归并排序的每一次合并中，只需要进行replay操作，从新元素到根路径上逐一重赛。在每一层中，只需要进行一次比较。由于树是平衡的，从叶子结点到根路径仅包含 O(logk) 元素（这里高度的计算和之前讲解堆的推导是差不多的，你可以复习之前的章节）。所以，总的时间复杂度为 O(klog k) 。

现在有了败者树的加持，多路归并排序就可以比较高效地解决外部排序的问题了。对于1TB任意文本的排序问题，大致思路就是：

1. 先用内排序算法，尽可能多的加载源文件，将其变成n个有序顺段。
2. 在内存有限的前提下每k个文件为一组，每次流式地从各个文件中读取一个单词，借助败者树选出字典序最低的一个，输出到文件中，这样就可以将k个顺段合并到一个顺段中了；反复执行这样的操作，直至所有顺段被归并到同一个顺段。

这里稍微补充一下，看起来我们每次从文件中只读取了一个单词，但操作系统在读文件的时候是会按页为单位读取并缓存下来的，所以某一次磁盘访问之后的若干次访问，其实都会直接命中cache，也就是说，并不是每次从败者树中取出元素时都会真的产生磁盘IO，请不用担心。

当然在工业级实现中肯定还是有很多优化空间的。比如待合并的文件比较大的时候，我们可以利用二分搜索对文件进行分段，并行地合并，相关研究也比较多，感兴趣你可以自行搜索了解。

总结

随着互联网的发展，数据量一直在稳步的提升，许多算法问题都不能只简单地考虑内存中可以存储，甚至单机磁盘可以存储的情况了。相信我们今天学习的外排算法思想，一定会给你一些解决此类问题的启发，希望你可以举一反三在实际生产中也能将算法更好地运用在有各种限制的真实环境中。

借鉴内排中归并排序的想法，我们可以实现一个多路归并的外排算法，解决内存空间不足的问题。但也因为涉及外部存储，需要重点考虑IO的成本。通过尽可能多地利用内存中的排序，得到尽量少的初始顺段，以及选择合适的多路归并参数，我们就可以做到外存访问次数尽量少了。

多路归并，可以通过堆或者败者树实现，这里我也给你贴一道力扣上的算法题供你练习。

课后作业

最后留两个思考题。

1. 既然，我们已经有了堆，为什么还要用败者树这样的数据结构呢？还有前面提到的胜者树，它不是一个更直观的表示方式吗？所以相比堆和胜者树，败者树在解决求多个元素中最小元素的问题时，又有什么样的优势呢？
2. 除了基于O(n*logn)的归并排序的思想，有没有可能基于线性排列的几种算法来实现外部排序呢？

欢迎你留言与我讨论。如果有收获也欢迎你转发给身边的朋友，邀他一起学习。我们下节课见～

09｜二分：如何高效查询Kafka中的消息？

作者: 黄清昊

你好，我是微扰君。

今天我们来学习另一个常用的算法思想，二分法。这个算法思想相信即使你没有什么开发经验也不会感到陌生，而且之前讲红黑树的时候我们也简单聊过。

不知道你有没有玩过“猜数字”的游戏。大家规定一个范围，一个人在心里想一个这个范围内的具体数字，比如一个1-100的自然数，然后另几个人来猜数字；每次猜错，这个人都会提示他们的猜测是大了还是小了，看谁最快猜到数字。

如果你做这个游戏会怎么猜呢？从1开始顺次猜吗？我反正不会这么猜，出于一个很简单的直觉，如果1猜错了，那么出题的同学给你的提示对可选范围的缩小非常有限，也就是从1-100变成了2-100。

我想很多人第一反应也都会是从比较中间的位置，比如50，开始猜起。毕竟如果50猜错了，因为要提示是大了还是小了，范围就要么缩小到1-49，要么缩小到51-100，这样猜测范围就可以成倍的缩小。

所以，如果每一次我们都猜测可能范围内的中间值，那么即使猜错了也能成倍的缩小范围，这样的策略其实就是二分查找算法。

有了二分查找算法，即使更大的范围内进行游戏，比如在1-1,000,000的范围内，我们按照二分的策略，最多也只需要20次即可完成任意数字的猜测，这是遍历数字猜测所远远做不到的。可以看下图有一个直观的认知。

不过这样凭感觉的分析肯定是不行的，我们来严谨地讨论一下二分查找相比于线性查找，到底有多大的优势吧。

二分查找

在具体比较它俩的复杂度和实现之前，首先，我们要知道二分查找相比于线性查找更快是有先决条件的，就是查找的范围内的元素一定是有序排列的。

比如在刚刚说的猜数字游戏里，我们之所以每次能排除一半的搜索空间，就是因为数字整体是有序排列的，如果某次猜测的数x，比目标值target大，那么当然比x更大的数就没有必要猜测了。

那什么是无序的排列呢？我们换个例子，假设要从一个由字母构成的数组中寻找某个目标字母“G”是否存在。

如果字母数组本身是按照字母序排序的，那么显然可以用二分查找法，和翻字典的过程其实差不多，如果我们打开的当前页比目标字母要字母序更靠前，那么我们肯定会往后翻，反之则会往前翻。

但是如果字母数组并不是按照字母序排列的，而是随机排列没有规律可言，这样唯一的做法就只有遍历数组的每个元素逐一对比了，因为在乱序的数组中的任意位置和目标字母进行比较，不会有任何有用的信息可以告诉我们应该要往前找还是往后找。

这也是为什么有序的结构在很多情况下是更受偏爱的，我们在许多场景下，会先对数组元素进行排序预处理，再进行后续的其他操作；哪怕我们知道，排序，即使是内排序，也会花费不菲的代价，但它带来的收益可能是更高的。

在算法面试中，你可能会经常碰到这样要先进行排序预处理的题目，比如力扣上经典的两数之和题，有一种基于双指针的做法就是要先进行排序预处理的；而采用快排这样的O(nlgn)时间复杂度的算法，即使多出了预处理的步骤，也可以让我们获得比暴力法更好的时间复杂度的算法。如果你感兴趣可以课后尝试解决一下相关的题目。

二分查找

所以我们也来看一下数组上的二分查找算法在维基百科上的严格定义：

二分查找是一种在有序数组中查找某一特定元素的搜索算法。

搜索过程从数组的中间元素开始，如果中间元素正好是要查找的元素，则搜索过程结束；如果某一特定元素大于或者小于中间元素，则在数组大于或小于中间元素的那一半中查找，而且跟开始一样从中间元素开始比较。如果在某一步骤数组为空，则代表找不到。

这种搜索算法每一次比较都使搜索范围缩小一半。

核心就是查找元素需要可比较且有序的排列。

好，我们来简单分析一下二分查找的时间复杂度。也很简单，假设我们搜索的空间内一共有N个元素，它们可以根据某种比较函数升序排列，那基于二分查找的策略，均摊下来需要比较多少次呢？

比较理想的情况，比如查找的元素正好在序列正中间，搜索一次就可以返回了。但我们需要考虑最差的情况。由于每次检索至少能排除一半的空间，假设一开始的搜索空间大小是N，那么我们的搜索空间在最差情况下会构成一个等比数列，下图是一个N=16时具体的例子：

当搜索空间里只剩一个可能元素时，也就是最后一次猜测，我们要么猜到了答案，要么就是答案不存在。这样最坏的搜索次数就是最大搜索空间折半多少次可以变成1。所以二分搜索的时间复杂度就是O(logn)了。

二分查找的实现

用代码实现的时候，往往会用数组来存储有序排列的待搜索元素，这里假设整个数组元素是升序排列的。我们用left和right两个整型作为数组的下标，代表搜索空间的左右边界，进行循环猜测。

每次猜测我们都会选择可选范围内最中间的元素去和目标值比较，最中间元素的数组下标为(left+right)/2。如果和目标值一样，我们就猜中了答案；如果比目标值大，说明比当前元素大的元素都不可能了，我们应该把可能范围的右边界移到当前位置之前；反之，就应该把左边界移到当前位置之后。

以猜数字为例，力扣上的374题描述的就是这个游戏，对应的二分查找代码可以这样写：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

// Forward declaration of guess API.
// @param num, your guess
// @return -1 if my number is lower, 1 if my number is higher, otherwise return 0
int guess(int num); // num比答案高返回-1； 否则返回1

class Solution {
public:
    int guessNumber(int n) {
        int l = 1;
        int r = n;

        while(l < r) {
            int mid = l + (r-l)/2; // 每次用左右边界的中点作为猜测值

            if (guess(mid) == 0) return mid; //猜中直接返回

            if (guess(mid) < 0) { // 猜的数大了
                r = mid - 1;
            } else { // 猜的数小了
                l = mid+1;
            }
        }

        return l;
    }
};

你只要记住我们始终让l和r表示可能的范围，再根据中间值比较的结果，进行边界的缩放；代码是很容易实现的。

这里我们也对比线性搜索的代码看一看：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

// Forward declaration of guess API.
// @param num, your guess
// @return -1 if my number is lower, 1 if my number is higher, otherwise return 0
int guess(int num);

class Solution {
public:
    int guessNumber(int n) {
      for (int i = 1; i <= n; i++) {
        if (guess[i] == 0) return i;
      }
      return -1;
for        
    }
};

可以看到，在线性搜索的代码里，其实没有用到guess[i]为-1或者1的信息，也就没有利用到数字本身是有序的特点。当目标值为n的时候，我们需要比较n次才能得到答案，所以均摊的整体时间复杂度为O(N)。在N很大的时候，线性查找会比二分查找慢很多。

二分查找的应用

那二分查找在工程中常用吗？ 可太常用了，下面我们就以Kafka的索引查询为例，学习一下二分查找在工程实战中可以发挥的巨大威力。

当然你可能会说，平时写业务代码的时候好像也没怎么写过二分查找。这其实也很正常，一是因为大部分时候，业务代码很少会在内存中存储大量的线性有序数据，在需要比较大量数据的检索时，我们往往会依赖底层的中间件；而数据量比较小时，线性查找和二分查找可能也差别不大了；另外我们也常常会用一些如红黑树这样的结构去存储有序集合，检索的时候也不会用到二分搜索这样在线性容器内的操作。

不过作为有追求的程序员，这种非常基础的算法思想我们还是很有必要掌握的，不止是能帮助你通过面试，更能帮助你更好地理解像Kafka这样的中间件的部分底层实现原理。

Kafka

我们知道Kafka是一款性能强大且相当常用的分布式消息队列，常常用于对流量进行消峰、解耦系统和异步处理部分逻辑以提高性能的场景，不太了解的同学可以去看一下官网的介绍。

在Kafka中，所有的消息都是以“日志”的形式存储的。这里的“日志”不是说一般业务代码中用于debug的日志，而是一种存储的范式，这种范式只允许我们在文件尾部追加新数据，而不允许修改文件之前的任何内容。

简单理解，你可以认为Kafka的海量消息就是按照写入的时间顺序，依次追加在许多日志文件中。那在某个日志文件中，每条消息自然会距离第一条消息有一个对应的offset，不过这里的offset更像是一个消息的自增ID，而不是一个消息在文件中的偏移量。

为什么说是许多日志文件，而不是一个巨型的日志文件呢？这也是一个常用的计算机思想：分片。在这里，分片可以让我们更快速、更方便地删除部分无用文件，提高磁盘的利用率。

Kafka日志文件具体的存储方式可以参考这张图。Kafka的每个topic会有多个partition，每个partition下的日志，都按照顺序分成一个个有序的日志段，顺次排列。

怎么找到消息

我们知道，Kafka虽然不允许从尾部以外的地方插入或者修改数据，但我们在Kafka中还是很可能需要从某个时间点开始读数据的，这就意味着我们要通过一个offset，快速查找到某条消息在日志文件的什么位置。这里再强调一下，kafka中的offset实际上是类似于消息自增ID的存在，并不是真的在磁盘上的偏移量。

但由于每条消息的消息体不同，每条消息所占用的磁盘大小都是不同的，只有offset，没有办法直接定位到文件的位置。所以我们要么遍历日志文件进行查找，要么我们为日志文件建立一套索引系统，将消息offset和在文件中的position关联起来，这样我们就可以利用消息offset的有序性，通过二分法加速查找了。

下面是一个典型的某个topic的某个partition下file（日志文件）的存储情况。

1
2
3
4
5
6

00000000000000000000.log
00000000000000000000.index
00000000000000000000.timeindex
00000000000000000035.log
00000000000000000035.index
00000000000000000035.timeindex

• .log文件就是存储了消息体本身的日志文件；
• .index文件就是用于帮我们快速检索消息在文件中位置的索引文件；
• 这里还有个.timeindex后缀的文件，它和index其实差不多都是索引文件，只不过在这个文件中关联position的变成了时间戳。

所有的文件名都是在这个file下的第一条消息，距离Kafka整体的第一条消息的offset，也就是绝对偏移量，那么在一个index文件内，我们就只需要用更小的空间存储相对偏移量即可。

而index文件的内容也很简单，就是用固定大小的记录来标记一对“消息offset”和“消息在log文件中的位置position”的关系，当然我们会保证消息offset是递增的。下图是一个简单的示意。有了这个索引文件，就可以快速地进行二分查找了。

当然这里还有个小细节不知道你有没有注意到，在index文件中，文件中的offset并不是连续存储的。这会导致我们拿着offset，在index中查询时，只能大致查找到一段可能的范围；之后在.log文件中，我们还需要在查找的最接近的消息的位置往后顺序遍历，才可以找到真正的offset所在精确位置。

比如要查询offset为29的消息，在索引表中只能定位到offset为26的位置在838，然后我们还要从838的位置开始，在.log文件中往后遍历查询，直到找到offset为29的消息。

这其实是一个时空效率的权衡，为了使用更少的内存空间，Kafka采用的是稀疏不连续的索引，在实战中起到了非常好的效果。

好说到这，相信让你基于Kafka的存储体系，去实现指定offset消息的查询也可以轻松实现了吧。不过这里要稍微再多说一点，很可能你所面对的不是一个可以在内存中放得下的简单索引文件，而是一个比内存大得多的存放在磁盘上的东西。怎么办？

Kafka的做法是基于mmap技术，将硬盘上的文件和内存进行映射；当然由于硬盘的空间可能比内存大很多，所以并不能够直接将内存在物理层面上与磁盘进行一一映射，这里我们需要引入虚拟内存的手段。这点我们会在操作系统篇讲解LRU缓存置换算法的时候进一步讨论。

那最后我们来看一下kafka中的代码到底是怎么写的吧。

kafka源码实现

你可以对比一下自己想的实现看看有没有什么差别。当然Kafka用的是scala语言，你可能需要花一点时间理解一下基础的语法。

在整个Kafka中，二分搜索的核心作用就是用于加速索引指定offset的消息，所以相应的代码都在 core/src/main/scala/kafka/log/AbstractIndex.scala 中。 indexSlogRangeFor 就是用于检索目标值的函数，其返回值就代表可能范围的上下界，我们会不断的递归搜索，如果最终返回的下界和上界相等，就说明我们找到了目标值。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

/**
	   * Lookup lower and upper bounds for the given target.
	   */
	  private def indexSlotRangeFor(idx: ByteBuffer, target: Long, searchEntity: IndexSearchEntity): (Int, Int) = {
	    // 检查index是否为空
	    if(_entries == 0)
	      return (-1, -1)
	
        // 二分搜索
	    def binarySearch(begin: Int, end: Int) : (Int, Int) = {
	      var lo = begin
	      var hi = end
	      while(lo < hi) {
	        val mid = ceil(hi/2.0 + lo/2.0).toInt
	        val found = parseEntry(idx, mid)
	        val compareResult = compareIndexEntry(found, target, searchEntity)
	        if(compareResult > 0)
	          hi = mid - 1
	        else if(compareResult < 0)
	          lo = mid
	        else
	          return (mid, mid)
	      }
	      (lo, if (lo == _entries - 1) -1 else lo + 1)
	    }
	
	    val firstHotEntry = Math.max(0, _entries - 1 - _warmEntries)
	    // 查询的目标offset是否在热区
	    if(compareIndexEntry(parseEntry(idx, firstHotEntry), target, searchEntity) < 0) {
	      return binarySearch(firstHotEntry, _entries - 1)
	    }
	
	    // 查询的目标offset是否小于最小的offset
	    if(compareIndexEntry(parseEntry(idx, 0), target, searchEntity) > 0)
	      return (-1, 0)
	
	    return binarySearch(0, firstHotEntry)
	  }

看第9-25行代码，和前面写的“猜数字”代码看起来是不是如出一辙呢？就是简单的二分搜索，相信你应该没有什么问题了。

我们稍微解释一下27行到37行的代码。你可能会很疑惑，有了二分搜索函数，我们直接检索 binarySearch(0, _entries - 1) 不就可以了吗？为什么还要分两段检索呢？

这其实也涉及到操作系统、内存和mmap的工作机制。

前面我们提到Kafka利用mmap，将更大的磁盘文件映射到了一个虚拟内存空间，但底层的内存存储其实是相对小的；所以很多时候，我们需要将一部分暂时不用的空间，从内存中置换出去，把需要访问但此时不在内存中的文件，置换进来，这个方法叫做内存置换，每次内存置换都会触发一次“缺页中断”，之后我们会在LRU的章节里展开讲解，现在你只需要知道这个操作显然是需要比较高昂的成本就可以了。

而Kafka消息队列的特性也决定了，我们大部分的索引查询其实都是在日志比较靠近尾部的区域(数据比较新)。

那么，如果我们将索引中最后的 8KB 认定为“热区”，是大部分查询所会命中的区域，剩余的区域认定为是“冷区”，但每次查询的中间位置随着日志的增长就很容易变成冷区，就很容易触发缺页中断。一个典型查询在内存中产生的冷热页面的对比例子可以参见下图：

如果我们在查询的时候分成两段，优先查询热区，没有命中时再查询冷区，是不是就能大大减少“缺页中断”的次数了呢？是的，对于查询常在尾部出现的情况下，采用冷热分区的二分查询算法能很好地优化性能。具体issue可以参考这个。

可以看到，相比简单的内存中的二分查询，Kafka中的二分查询考虑了更多的“现实”问题，这也是我们在工程中遇到算法问题和平时做算法题的算法的一个很大的差异。所以如果我们想要真的把算法很好的应用于工程中，除了对算法本身的掌握需要过硬；也需要真正打好计算机基础知识；对程序和操作系统的运行了如指掌，才能真正写出高性能的代码。

总结

相信通过今天的学习，你已经学会了如何给一个基于“日志”存储的消息队列，建立消息的索引查询了吧。通过线性有序的索引文件，我们其实可以为任何需要查询的系统，进行基于二分法的查询，以优化查询效率。二分查找的核心就在于，相比于线性查找，我们可以在每次查询中成倍地缩小可能的查询范围，达到O(logN)的时间复杂度。

所以在工程中，建立索引文件，或者是对业务数据进行排序，都是常用的预处理手段。通过一次计算，就可以帮助我们在之后的查询操作中获得更好的效率，是典型的空间换时间的手段。希望你在工作生活中可以灵活运用今天学习的知识。

课后作业

那最后提一个小问题。 既然建立线性的索引文件就可以帮助我们加速查询的过程，那为什么在许多情况下，我们还需要使用诸如红黑树、B+树这样的复杂索引结构呢？比如InnoDB的索引文件就采用了B+Tree，它和Kafka所选择的顺序稀疏索引文件各有什么优劣呢？

欢迎你在留言区和我一起讨论。如果有收获也欢迎你转发给身边的朋友，邀他一起学习。我们下节课见～

10｜搜索算法： 一起来写一个简单的爬虫？

作者: 黄清昊

你好，我是微扰君。

你玩过井字棋的游戏吗？在一个九宫格中，双方轮流用X和O占领一个格子，某一方的O或者X三个连成一线时即可获胜。

这样一个简单井字棋的游戏，如果要让你自己写代码实现一个AI，你会怎么做呢？怎么把博弈过程清晰地表示出来呢？

实际上，许多博弈类游戏的过程，我们都可以用树来表示。根节点就是棋盘为空的状态，终点就是各个棋下完的状态，这样的树也被称为博弈树。下图是井字棋某个局面3步内的树状展示：

一般来说，对弈双方在做的事情，其实就是找到这棵树上对于自己最优的一种落子方式，使得之后的每条路径，自己都有必胜或者必不败的策略。如果你想要找出一个AI策略，最暴力的方式就是直接遍历每一种情况，找到最优的下法，这就是一个典型的搜索问题了。

事实上，这类博弈的游戏要么是先手必不败，要么是后手必不败，所以对全空间的搜索一定是可以写出一个无敌AI的，对证明感兴趣的同学可以去搜索“策梅洛定理”了解。

如果暴力遍历，有多少种情况呢？相信你也发现了，就是这么一个简单的井字棋小游戏，终局的数量非常多，达到了255168种。我们可以这样来简单地估计它，第一步有9种下法，第二步有8种下法，显然通过排列组合的知识，占满棋盘一共有9！=362880种下法，当然还需要去掉一些中间获胜不应该继续进行对弈的情况。

这本身是一道挺有意思的数学问题，也可以通过写代码更快地计算出来，留给你作为课后作业。当然这个数字还不算天文数字，尚且在计算机的处理范围之内，如果我们稍微把游戏的复杂度提升一下，比如围棋，还能通过暴力搜索的方式得到一个优秀的AI吗？

我们知道，围棋盘有19*19个落子点，所以刚开始的每一步可能都有接近361个选择，那整体的情况可能接近361！种。这是一个天文数字，在现在的计算机架构下，直接计算和存储这样的问题是不可能的。所以我们想要写出一个靠谱的围棋AI，就需要采取一些新的策略，只选择部分分支进行遍历，从中找出一个比较好的方案。

对于人类而言这个过程就是依靠经验，对于AI来说，就是依托于数据，你从AlphaGO核心算法的名字“蒙特卡洛搜索树”中，就可以看出来，这本质上还是一个搜索的问题，只不过人类棋手和AI都采用了比较高明的搜索策略。

我们今天就不讲那么进阶的内容了，就讲一讲平时常用的广度优先搜索算法BFS和深度优先搜索算法DFS。

它们是两种最常见的暴力搜索算法，在面试中也相当常见，前者的实现需要用到我们之前讲解的队列这一数据结构，后者则是递归思想最常用的场景之一。在工程中它们也发挥着巨大的作用。比如，DFS在前端开发中DOM树相关的操作里就非常常见，我们可以用它来实现对DOM树的遍历，从而对比两颗DOM树的差异，这就是React中虚拟DOM树算法的关键点之一。

BFS和DFS

BFS和DFS，作为两种最暴力、也相当常用的搜索策略，最大的特点就是无差别地去遍历搜索空间的每一种情况，因此但凡是可以抽象成图上的问题，基本上都可以考虑用BFS、DFS去做。只不过效率可能不是最优的，所以我们也常常称之为暴力搜索算法，在各大刷题网站题解区中，你应该常常能见到“暴搜”这样的关键词，说的一般就是DFS和BFS这两种算法。

所以，在爬虫这样本来就需要无差别遍历全部空间的场景下可以说是非常合适的了。至于DFS和BFS具体选择哪一种，我们可以结合一个具体的爬虫场景来分析。

如果让你手写一个爬虫，从豆瓣上爬取一个用户关注的所有用户，是不是很简单？只要直接遍历某个用户的关注者列表就可以了，除了需要处理一些鉴权和页面解析的问题，没有什么复杂的地方。

那我们升级一下挑战，爬取这个用户关注的人的所有关注的人，也就是和这个用户有二度关系的所有用户，你要怎么实现呢？如果不是二度，而是让你查找三度关系，也就是找出需要三跳的所有用户，你的代码能否很简单地通过配置就完成这件事呢？

这其实就是一个非常适合用DFS和BFS解决的问题，因为它天然就是一个需要无差别遍历所有图上节点的问题。

你可以把豆瓣用户看成节点，用户之间的关注关系就是边，它们一起构成了一个复杂的社交网络。相信你也听过社交网络中的“六度分割理论”，说的就是世界上任何一个人和你之间的距离不会超过6度，描述了社交网络的小世界特性。这种网络关系也是许多人在研究的。

BFS实现思路

好我们先来使用BFS解决这一问题。我们优先去遍历所有到源用户距离为一度的用户，然后再遍历这些用户的邻居，用层层深入的方式进行搜索。广度优先搜索，其实是一个很直观的定义，我们把对应到图上的搜索顺序画出来，就很清晰了。

看简化的情况，假设我们搜索的源用户为图的0节点，一度关系包含3个节点，二度关系包含了6个节点，每条连边都是一个关注关系。那我们基于广度优先搜索策略遍历时，就会按照标号顺序进行遍历。

为了在代码中实现这样的遍历效果，求出所有和0节点构成两度以内关系的用户，我们就需要借助之前学习的“队列”了。

因为广度优先搜索是由源点向外逐层推进的，每遍历下一层的时候，我们都需要用到上一层的节点，所以我们需要一个容器记录每一层的元素，并依次遍历，先进先出的队列就可以帮助我们很好地解决这个问题。

我们首先把遍历的初始节点加入queue中，然后循环读取queue中的元素，每次读出一个元素，就把它的所有相邻节点都放入新队列中，直到目前队列为空，就代表我们遍历完了所有的元素。队列FIFO的特性保证了，下一层的元素一定会比上层的元素更晚出现。

当然我们经常需要设置自己的搜索退出条件，比如在最短路径问题中，我们并不需要遍历所有的路径，当搜索到终点的时候其实就可以退出了；在我们的例子中，我们的退出条件也不是遍历完整个社交网络，遍历到第二度关系就可以结束了，因此我们还需要在代码中记录当前遍历的层数。

另外，有时候需要对插入队列中的元素做一些判重，防止重复的搜索。

在搜索最短路径或者求几度关系所有用户的情况下就很有用，因为重复的节点已经没有必要再搜索了；如果不这样做，甚至可能导致你的搜索永远无法结束，比如在图有环的情况下。

实现

下面我们试着写一下，就用爬虫常用的脚本语言Python来实现这次的代码。

解释一下几个关键的变量。

degree就是广度优先搜索中我们用于记录搜索层数的变量。为了每一次出队的时候，把一层的元素全部出队，在遍历中我们采用了两层while循环，这样内层循环结束后，我们就可以保证当前层的元素已经全部被访问，也可以将degree进行自增操作。

之所以开了一个新的变量next_degree_urls也是同样的道理，当前层的邻居不能干扰到这一层的出队操作，所以我们将邻居们放到一个新的队列中；内层循环结束后，再将已经为空的队列更新为next_degree_urls。

而类型为set的变量res，除了记录所有的用户主页，也起到了判重的作用；如果已经出现在res集中，说明我们已经遍历过这个用户主页了，不需要再遍历一次了，所以在循环中直接通过continue跳过。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

# 这里我们需要一个合适的 douban html parser
    def crawl(self, startUrl: str) -> List[str]:        
        urls = deque()
        urls.append(startUrl)
        res = set()
        
        degree = 0
        N = 2
        
        while urls:
            # 遍历层数超过N层，停止遍历
            if degree > N: break 
            # 用于记录下一层的节点
            next_degree_urls = deque()
            # 遍历当前层 
            while urls:
              u = urls.popleft()
              if u in set: continue
              for url in doubanHtmlParser.getFollowings(u):
                  next_degree_urls.append(url)
              res.add(u)
  
            urls = next_degree_urls
            # 当前层元素全部出队；进入下一层遍历，记录遍历层数的变量加1
            degree = degree + 1

当然，既然是爬虫，我们肯定是需要对网页进行解析的，就用一个getFollowing函数表示这个过程，做的事情就是输入一个用户主页的URL，返回该用户关注的其他好友的个人主页。

大致实现思路就是，通过网络请求库获取指定URL的HTML文本信息，从中解析出表示用户关注好友列表的部分，一般列表中的每个元素都会指向该好友的个人主页，我们把相关的href标签里的URL解析出来即可。

好了，这样我们就用基于队列的广度优先搜索策略完成了一个简易的爬虫，感兴趣你可以自己补全HTML解析器的代码，完整实现一下。

DFS实现思路

再来用DFS解决这个问题。深度优先搜索，就不会再像广度优先搜索那样严格由内而外逐层推进了，它和棋手下棋的思路其实会更像一点，我们就用下棋举个例子。

假设下棋的时候，当前局面可以有若干个落子点，棋手一般会先顺着其中一个落子点在脑海中模拟若干步，发现某一步不行，我们回溯到分叉点，再看一下其他选择；最终遍历完当前选择的落子点的各种局面之后，再依次进行其他落子点的判断，直到选出一种比较优的策略。

画成图对比一下，你会更直观地感受到两者的区别，同样用刚刚假想的豆瓣用户关注关系图来举例：

上图的数字，表示深度优先搜索在同样的关系图中的遍历顺序；可以看到相比于BFS的逐层推进，在DFS中，是一条条分支顺次遍历到终点再进行下一种尝试的，这也是深度优先搜索命名的由来。

实现

这种遍历方式也天然符合回溯法的适用场景，所以常规做法就是通过递归来实现。写成代码如下，关键就是11-13行的for循环，递归地对当前节点的每个子节点进行同样的DFS过程，细节你可以参考代码中的注释来理解，网页解析的逻辑和前面所说的是一样的：

1
2
3
4
5
6
7
8
9
10
11
12
13

# 结果集 用于存放所有N度关系以内的用户
    res = set()
    N = 2 # 记录找N度关系以内的所有用户；N=2即找2度关系以内的用户
    def crawl(startUrl, degree):  
        # 如果已经超过N度关系，我们不用继续遍历      
        if degree > N : return
        # 如果已经搜索过，我们也不用继续搜索
        if startUrl in res : return
        # 将当前搜索的用户主页加入结果集
        res.add(startUrl)
        for url in doubanHtmlParser.getFollowings(startUrl):
          # 遍历关注的所有用户，注意需要将度数增加1
          crawl(url, degree+1)

DFS的代码看起来明显要简短很多，这就是递归的威力。通过对自身的调用，很多时候，我们可以让代码变得非常简单。

时空复杂度

两种方法的时空复杂度都非常容易分析。

先说时间复杂度。我们做的事情就是遍历一次网络或者图中的每个节点，因为借助结果集判重后，即使在图中多次出现的节点，也只会入队一次或者被递归一次。那么假设图中有V个顶点，E条边。

在BFS中，图中所有顶点入队一次、出队一次，每条边都会在边起点出队的时候被遍历一次，所以整体的复杂度为O(V+E)。而在实际的复杂网络中，E一般是远大于V的，所以可以近似认为复杂度为O(E)。

在DFS中，是从起点出发递归遍历图，通过结果集判重，保证重复的节点不会被递归两次，从而每条边只会被遍历一次，整体复杂度为O(E)。

所以时间复杂度上两者没有太大的差别。但是在一些求最短路径的场景下，比如求从当前位置走出迷宫的最短路径，就会有一定差异。这是因为BFS是从内到外逐层搜索，所以最早搜索到终点的时候，就对应了最短路径；因此找到终点我们就可以提前结束搜索过程。

而在DFS中，由于我们优先遍历每条路径的最大深度，即使找到了终点也只能说明找到了一条路径，这并不能保证这条路径是最短的，所以哪怕找到终点也不能结束搜索过程，需要遍历完整个搜索空间，找到所有可能的路径之后再从中选择一条最短的。

在搜索空间很大，但已知搜索路径不会特别长的情况下，DFS可能会比BFS要慢很多，所以你要根据实际情况选择一种合适的算法。

当然在今天豆瓣爬虫的场景下，我们需要的就是遍历整个空间找到所有构成N度关系的用户，所以两种算法的时间复杂度其实没什么区别。

空间复杂度

再来看空间复杂度。

在BFS中，主要的空间复杂度就是queue和res所占用的大小。那这里的，res本身并不是所有的BFS场景下都会需要的，因为我们并不一定需要返回所有遍历过的节点，可能只需要记录一个距离之类的值。

但是，在大部分的BFS下，我们是不希望重复遍历节点的，所以仍然需要一个类似于res的集合去标记所有经过的点。它所需要的最大空间和图中总结点数量V一致。而queue存储的就是图上的节点，其所占用的空间最大也不会超过V。所以BFS的空间复杂度是O(V)。

在DFS中，所消耗的内存同样主要与res相关。因为虽然相比于BFS，我们少了queue的内存消耗，但是也多了隐含递归中调用栈所消耗的空间。由于调用栈最多就是描述一条经过了所有节点的路径，其最大空间大小也不会超过顶点数量V。因而DFS的空间复杂度同样是O(V)。

总结

作为两个相当常用的暴力搜索算法，BFS和DFS比较适合用来解决图规模不大，或者本身就需要无差别遍历搜索空间的每一种情况的问题；这两者的时间空间复杂度是相当的。

而至于DFS和BFS具体选择哪一种，我也总结出一些自己的经验，供你参考。

BFS因为是由内向外地毯式地搜索，所以首次搜索到目标位置的时候一定是源点到目标位置的最短路径，所以求最短路径类的问题往往可以用BFS解决。当然，这里的“最短路径”是有条件的，只有在图中所有边权重相等时首次搜索到的才是最短路径；另一类边权不等的图上的最短路径求解问题我们之后会单独讲解。

而DFS实现起来比BFS更简单，且由于递归栈的存在，让我们可以很方便地在递归函数的参数中记录路径，所以需要输出路径的题目用DFS会比较合适。毕竟想用BFS实现相同的路径记录，除了需要在queue中记录节点，还需要关联到此节点的路径才可以，占用的空间比DFS高得多。

一般情况下我们都可以优先使用DFS实现，但这完全建立在我个人觉得DFS写起来更简单的前提下。而在需要求解路径本身的问题中，强烈建议你采用DFS作为搜索算法的实现。

课后作业

最后，我再给你留三个小作业。

1. 既然说是可以用DFS或者BFS写一个爬虫，希望你尝试补全一下爬虫中解析HTML和HTTP请求的逻辑。或者动手写一个你自己想写的爬虫感受一下，体会搜索算法在实战中的应用。
2. 在搜索N度关系的所有用户时，如果我们希望同时把源用户和这些用户的关注关系记录下来，比如A->B->C就表示A关注了B，B关注了C；更广泛地意义上来说就是让你记录搜索过程中的路径。你会怎么实现这个逻辑呢？
3. 前面提到的井字棋中，尝试用代码计算一共有多少种最终合法的局面呢？

欢迎你在留言区与我讨论。如果有收获也欢迎你转发给身边的朋友，邀他一起学习。我们下节课见～

11｜字符串匹配：如何实现最快的grep工具

作者: 黄清昊

你好，我是微扰君。

grep命令，相信使用过Linux的同学都会非常熟悉，我们常常用它在Linux上进行文本搜索操作，具体来说就是从一段文本中查找某个字符串存在的行。下面一个典型的grep的使用例子，比如我可以用它来看看自己在LeetCode上用Java做了多少题：

GNU Grep 则是 grep 命令的一个工业级实现，在项目官方 Readme 中作者是这样介绍它的：

This is GNU grep, the “fastest grep in the west” (we hope).

其实就是在说这是世界上最快的grep程序。当然，这款从上世纪就诞生的软件，敢这么说自己也是因为它有着十足的底气。

GNU Grep 确实是将“文本搜索”这一简单的功能做到了极致。作者 Mike Haertel 自己写了一封邮件解释 GNU Grep 为什么这么快，主要有两点：

1. 它避免了检查每一个byte
2. 对于被检查的byte，只需要执行非常少的指令

第一点的主要优化就在于 GNU Grep 用到了非常知名的字符串匹配算法：Boyer Moore 算法，也就是我们常说的 BM 算法，它是目前已知的在大多数工业级应用场景中最快的字符串匹配算法，因而被广泛应用在各种需要搜索关键词的软件中，许多文档编辑器快捷键 ctrl+f 对应的搜索功能都是基于这个算法实现的。

那第二点呢，就是当你发现查询的速度已经优化到足够好时，也需要让IO的速度更快一些，查询所需的指令也更少一些，这里可以优化的地方就更多了。

比如由于 grep 是按行查找的，许多版本的 grep 实现都会去遍历查找\n 换行符先进行分行，但 GNU Grep 则是将搜索文本直接读入一个缓冲区优先查找目标字符串，只有命中时才会在命中位置的前后进行换行符的查找；又比如，GNU Grep提供了基于mmap映射内存到文件的参数，可以减少一些内存拷贝的时间开销。具体的细节还有很多，比较繁琐，有兴趣的同学可以自行查阅 Mike Haertel 的邮件。

这个例子也再次说明了一件事情，要写出真正高性能的程序，不只要懂算法，也要懂计算机底层原理；只有这样，才能真正了解程序在运行时可能存在的各种性能瓶颈，找到不同场景下的最优解。

好我们回到今天的主题，字符串匹配。这也是一个经典问题了，相关算法非常多种，比如最暴力的 Brute-Force 算法、将前缀信息运用到极致理论性能极佳的KMP算法，还有利用哈希思想和滑动窗口思想的Rabin-Karp算法等等。

那为什么BM算法的性能在工程实战中最好呢？

别急，老规矩，我们还是先来严谨地定义一下字符串匹配问题，方便展开后面的讨论。

字符串匹配问题

假设给定长度为n的主串 s[0…n-1] 和长度为m的模式串 p[0…m-1]，一般n远大于m，请实现一个函数 match(string s, string p) 用于找出所有的p在s中出现的位置。

那如何解决这个问题呢？

容易理解、复杂度也相对差的方法就是，遍历主串的每一个位置，看当前位置是否能和模式串匹配上；能否匹配的判断方式也很简单，从主串的当前位置开始，逐一对比主串对应字符是否和模式串相等。如果可以匹配，说明找到了一个匹配的位点，记录下来；如果不可以匹配，我们就继续尝试下一个位置，直到整个主串遍历完全。这也是最暴力的Brute-Force算法的思路。

写成代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

/*
 * s: 主串
 * p：模式串
 */
std::vector<intstring> match(string s, string p) {
&nbsp; std::vector<int> ans;
  int n = s.size();
  int m = p.size();
&nbsp; int i, j;
&nbsp; for (i = 0; i < n - m + 1; i++) {
&nbsp; &nbsp; for (j = 0; j < m; j++) {
&nbsp; &nbsp; &nbsp; if (s[i + j] != p[j]) break;
&nbsp; &nbsp; }
&nbsp; &nbsp; if (j == m) ans.push_back(i);
&nbsp; }
&nbsp; return ans;
}

代码非常清晰易懂，相信你看懂没什么压力。

通常在字符串不长的时候，不同的匹配算法之间的效率差异不大。Brute-Force算法的实现和理解都非常简单，不容易出错，完美地符合了KISS（Keep it simple, stupid）原则，也就是让代码尽量简单从而避免出错。所以BF算法在真实开发的环境中出镜率很高，在日常工作中如果有手写字符串匹配的需求，你也可以考虑这种方式。

但这个算法在最坏的情况下时间复杂度确实不是很理想。

比如s = AAAAAAAA、p = AAAB时，在每个位置匹配p最终都会失败，但是都需要匹配到p的最后一个字母“B”才能发现匹配失败；这就导致我们总共需要匹配 mn 次，其时间复杂度就是 O(mn)。

那有没有办法优化它呢？我们再来认真观察一下BF算法，首先会从主串和模式串的头开始遍历匹配，看第一次匹配的情况，BF算法之所以慢，就在于匹配p[3]失败后，我们又从模式串的第一个字符p[0]和主串的下一个位置s[1]开始比较，而s[1]这个位置其实在之前的搜索过程中出现过了。

所以，我们有没有办法通过一些预处理的手段，利用p[0…2]和当前正在匹配的主串中s[0…2]相等的已知信息，跳过一些肯定不可能的匹配，从失配处s[3]继续匹配呢？ KMP和BM算法其实都是这样做的，只不过手段有些差别。

KMP算法将前缀的信息利用到了极致，用匹配串自身的信息建立了一张部分匹配表，在每次失配的时候可以用来加速模式串，而不是每次都只向后移动一位。其算法逻辑整体比较复杂，感兴趣的同学可以网上搜索一下相关资料自行学习。

而GNU Grep 中用到的BM（Boyer Moore）算法，不仅理解起来容易很多，实际应用时性能也更好，它同样是基于预处理来避免不必要的重复匹配。但BM算法引入了两条很好懂的规则，“坏字符”和“好后缀”规则，并采用从后往前的匹配顺序进行匹配，构思非常巧妙。

后面的内容我们就用 moore 教授本人提供的例子来讲解。

其中模式串p是EXAMPLE，主串s是 HERE IS A SIMPLE EXAMPLE。

坏字符规则

先来看第一条规则：“坏字符”规则，描述的是主串上的失配字符，目的就是为了跳过一些肯定不可能成立的匹配位置。

在BM算法中，我们同样将s和p对齐，开始遍历匹配，但匹配的顺序和BF算法不同，采用从后往前匹配的方式。这其实是一种非常巧妙的设计，你马上可以看到它配合坏字符规则使用时有着绝佳的效果。

所以在例子中，第一次尝试匹配，首先会把p[6]的“E”和s[6]的“S”匹配，发现它们不匹配，所以这里的“S”就是一个坏字符。

那此时我们有两种选择，一种就是直接将模式串往后移动一位尝试继续匹配，这就和之前BF算法的想法差不多，没有利用到模式串中任何先验的信息。

而另一种呢，就是BM的做法了。

我们先看失配的坏字符“S”在模式串p中是否有出现，如果没有出现，那说明模式串其实不可能和这个位置有重叠，可以直接跳过这段位置，从主串的下一个位置开始匹配。在例子中，“S”显然不属于模式串EXAMPLE，我们就应该跳过“S”继续匹配，这样就大大加速了匹配的过程。

同样在下一步匹配时，因为主串的“P”和模式串的末尾“E”不匹配，但失配的“P”在模式串中就有出现，我们可以将模式串中最后一次出现的“P”和主串中的“P”对齐，同样从模式串尾开始匹配。

至此，坏字符的主要内涵就全部展示出来了，也就是，每次失配的时候，我们需要将匹配串往后移动 （失配位置下标 - 失配字符最右出现的位置下标） 位；如果失配字符不存在，则位置为-1。

这里你可能会有个疑问，为什么是最右的位置呢，不应该是记录上一次出现的位置吗？我的理解是，如果在每个位置都存储相比于当前位置的上一次失配字符出现的位置，存储开销会大得多；而如果只存每个字符最右出现的位置，我们所需要的只是一个字符集大小的哈希表，用一个长度为256的数组即可实现。

当然，这个公式会导致我们有时候求出的移动值可能是负的，让模式串反而向前移动了。比如在 BBBBBB 和 ABB 匹配时，第一次失配的坏字符B，会让匹配串往后移动（0-2=）-2位，导致前移。

那往前移显然是没有意义的，因为当前位置之前的匹配可能我们已经全部排除了；所以当移动位数出现负数时，我们也要让模式串至少往后移动一位，这点通过对基于坏字符的移动值和1取max操作即可实现。

而在这种时候，我们另一条规则“好后缀”也就可以发挥作用了。

好后缀规则

我们继续来看刚刚的例子。

在SIMPLE和EXAMPLE的匹配中，我们发现“MPLE”都匹配得上，但主串中的“I”和模式串中的“A”出现了失配。那这里的“MPLE”，我们就会称之为好后缀；同样“PLE”、“LE”、“E”其实也都是好后缀。

此时如果应用之前的坏字符规则，我们应该将模式串往后移动（2-(-1)=）3位，因为“I”在模式串中不存在。

但是，有没有办法利用已经匹配上的好后缀“MPLE”的信息，往后移动更多位呢？

当然是可以的，我们只要看匹配上的好后缀“MPLE”及它的子串“PLE”、“LE”、“E”是否之前也出现在模式串中即可。这里只有子串“E”之前也出现在了模式串中，所以我们可以直接把模式串移动至和这里主串的“E”对齐即可，这样我们向后移动了6位，显然比坏字符规则跳过了更多不可能的情况。

总结起来，好后缀规则移动的方式就是，找到好后缀在模式串中最右的匹配位置，总计向后移动（模式串字符长度 - 1 - 好后缀在模式串上次出现的最右位置）位。以EXAMPLE为例，好后缀“E”在模式串中上一次出现的下标是0，整个字符串长度为7，所以向后移动（7-1-0）6个位置。

这里还需要注意一点，好后缀匹配的时候，只有最长的好后缀被允许出现在模式串的中间位置；其余子串只能匹配在模式串的前缀中。比如下面的例子，主串中的“A”和模式串中的“C”失配，“MABC”是最长好后缀，但之前并没有出现在模式串中。

• 我们不能直接将模式串直接移到“MABC”之后，因为这样会错过好后缀子串“ABC”的匹配点。
• 但同样我们也不用匹配红色虚线框中的“ABC”，因为“MABC”没有匹配上，后面所有的MABC的子后缀匹配肯定只能发生在模式串的前缀中。

好后缀和坏字符规则其实都是可以单独使用的；BM算法，为了尽可能多地跳过不可能匹配的字符，会选择两条规则中的较大移动值来往后移动。而且这两个规则和主串都没有关系，只和模式串自身有关，我们显然可以通过预处理得到两个规则的偏移表，来加速整个模式匹配的过程。

好了，现在讲完了BM算法“好后缀”、“坏字符”的两个规则和从后往前匹配的策略，我们一起来把例子匹配完成吧。

在查表发现好后缀的规则能跳过更多的位置后，我们选择将模式串往后移动了6位。这时“P”和 “E”没有匹配成功，我们采用坏字符规则，拿着坏字符“P”，找到模式串中出现的“P”位于p[4]，向后移动(6-4=) 2位和主串的“P”对齐。从尾部往前遍历匹配，此时，我们发现所有的字符都匹配上了，因而找到了一个完全匹配的位置。

具体实现

相信你现在已经大体理解整个BM算法的思路了，但正所谓，“细节是魔鬼”，BM算法从概念上理解其实并不是很难，但真要手写实现还是比较困难的，不熟练的时候debug很容易花费很多的时间。为了方便起见，我们就用Python来实现这个算法。

具体实现我们可以分为三个大块：“坏字符”最右位置计算、“好后缀”偏移表计算、在主串上的搜索实现。

坏字符最右位置计算

“坏字符”的部分是最简单的，只需要开一个dict，遍历一次模式串，找到每个字符出现在模式串中的最右侧的那个位置即可。事实上，我们可以用一个[0,256]的数组来替代HashMap以提高性能，大部分工业级实现也都是这样做的。

1
2
3
4
5
6

def get_bc(pattern):
    bc = dict() # 记录每个badchar最右出现的位置
    for i in range(len(pattern) - 1):
        char = pattern[i]
        bc[char] = i + 1
    return bc

由于遍历的时候我们会不断地覆写dict，所以最后遍历完成，就能得到每个badchar在模式串中最右侧的位置。

好后缀偏移表计算

“好后缀”的部分相对来说比较复杂，尤其是工业级的实现对性能要求很高，代码有很多trick，非常不易于理解，这里我们做一些简化的处理；而且在大部分时候，由于模式串比主串要短的多，即使预处理时间复杂度稍微高一些，问题也不是很大。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

def get_gs(pattern):
    gs = dict()
    gs[''] = len(pattern)

    # suf_len 用于标记后缀长度
    for suf_len in range(len(pattern)):
        suffix = pattern[len(pattern) - suf_len - 1:]
        # j 用于标记可用于匹配的位置
        for j in range(len(pattern) - suf_len - 1):
            substr = pattern[j:j + suf_len + 1]
            if suffix == substr:
                gs[suffix] = len(pattern) - j - suf_len - 1

    for suf_len in range(len(pattern)):
        suffix = pattern[len(pattern) - suf_len - 1:]
        if suffix in gs: continue
        gs[suffix] = gs[suffix[1:]]

    gs[''] = 0
    return gs

我们同样开一个dict，用于标记失配时每个字符串应该往后移动多少，也就是对应的好后缀应该和之前哪个子串或者前缀匹配。怎么做呢？

一种比较暴力的做法就是遍历所有可能的后缀，然后从前往后看这个后缀是否在模式串中的其他位置也出现了，后面遍历的会覆盖之前的记录，所以记录下来的就是最右的匹配位置。

记得前面说过如果一个后缀在模式串中不存在，我们不能直接跳过整个字符串，因为该后缀的子串还可能和模式串中的前缀重合。比如例子中的“MPLE”后缀虽然不再存在于“EXAMPLE”中，但是其子串“E”与“EXAMPLE”的前缀“E”是重叠的。

所以在后缀不存在的时候，还需要检查一下其子后缀是否在dict有对应的匹配，如果有的话，也应该采用；这个通过一次循环赋值即可实现，对应到代码里就是第14到17行。

我这里实现的时间复杂度为O(m^3)，你可以自己推导一下，也欢迎去留言区讨论。

匹配过程

有了好后缀的偏移表和坏字符的最右位置，我们就可以来实现整个匹配的过程了。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

def bm(string, pattern, bc, gs):
    # i 用于标记当前模式串和主串哪个位置左对齐。
    i = 0 
    # j 用于标记当前模式串匹配到哪个位置；从右往左遍历匹配。
    j = len(pattern)

    while i < len(string) - len(pattern) and (j > 0):
            # 从右往左匹配每个位置
            a = string[i + j - 1]
            b = pattern[j - 1]
            if a == b: # 匹配的上，继续匹配前一位
                j = j - 1
            else: # 匹配不上，根据两个规则的预处理结果进行快速移动
                i = i + max(gs.setdefault(pattern[j:], len(pattern)), j - bc.setdefault(a, 0))
                j = len(pattern)
            # 匹配成功返回匹配位置
            if j == 0:
                return i
    # 匹配失败返回 None
    return -1
    
if __name__ == '__main__':
    string = 'here is a simple example ' 
    pattern = 'example'

    bc = get_bc(pattern)  # 坏字符表
    gs = get_gs(pattern)  # 好后缀表

    print(gs)

    x = bm(string, pattern, bc, gs)

    print(x)

参照详细的注释，整个过程和前面讲解的原理是一一对应的，你可以配合代码一起理解。 完整的代码我放到了GitHub上。

时间复杂度

Boyer-Moore 算法，在最好情况下复杂度可以达到 O(n/m)，在字符集比较大的时候，坏字符和好后缀规则可以帮助我们快速跳过大部分不必要的查询，达到接近最好的时间复杂度的概率是比较大。

但BM算法的最坏时间复杂度估计就是一个很难的数学问题了，许多学者都尝试做过相关的证明，目前我知道相对精细的比较上限次数的估计是Guibas和Odlyzko给出的3n，你感兴趣的话可以阅读原始论文了解。

因而和KMP一样，BM算法的理论时间复杂度也在O(m+n)之内，但由于字符集比较大的时候，BM常常能达到更好的时间复杂度，所以在实际应用中得到了更广泛的使用。

总结

我们来总结一下BM算法的特性。

BM算法，最大的特点就是利用了对目标串的预处理，用空间换时间，避免了许多不必要的比较，预处理的方式主要来自于对“坏字符”和“好后缀”两条规则的观察，因为这两个规则和主串都没有关系，只和模式串自身有关，显然可以通过预处理得到两个规则的偏移表，来加速整个模式匹配的过程。

总的来说，BM算法不难理解但实现起来有一定复杂度，感兴趣的同学可以自行练习。不过这一个特定的字符串匹配算法的学习其实还是次要的，空间换时间和预处理的思想你可以好好感受。

课后作业

相信通过今天的学习，你已经知道了如何基于Boyer-Moore算法实现一个高效的grep命令了吧。这里我也把grep源码中BM算法出现的地方分享给你，代码中运用了许多不同的技巧，可读性其实并不是很好，作为今天的课后作业，留给你课后研究。

如果你在阅读代码的时候有什么问题欢迎留言和我一起讨论。如果你觉得有收获，也欢迎分享给身边的朋友一起学习，我们下节课见～

12｜拓扑排序：Webpack是如何确定构建顺序的？

作者: 黄清昊

你好，我是微扰君。

Webpack是现在最流行的前端构建工具，见证了前端工程化在过去十年里的繁荣发展。如果你是前端工程师，相信你在日常工作中应该会经常使用到。

Webpack让我们可以模块化地进行现代Web应用的前端开发，并基于我们的简单配置，从入口开始，递归地自动构建出复杂的依赖关系图，最终把所有的模块打包成若干个浏览器可理解的bundle。

整个过程比较复杂，但是其中显然会有一个构建顺序的问题需要处理。以 html-webpack-plugin v3.2.0 为例，我们用Webpack打包HTML文件的时候，文件之间会有一定的引用依赖关系，因而所构建的chunk之间也会有相应的依赖关系。

那问题就来了，打包的时候，按照什么样的顺序去打包才更合理呢？这就是拓扑排序需要解决。

我的第一份工作就是前端工程师，对前端开发感情挺深，也一直觉得其实前端里用到算法的地方绝对不在少数。所以今天我们就以Webpack为例，一起来学习拓扑排序在实战中所发挥的威力。

当然，如果你对Webpack了解不多，也不用担心，完全可以把这个问题看成如何在一堆有依赖关系的源文件中找到合适的加载或者编译的顺序，和你常用的maven、makefile这类编译和依赖管理工具，去编译项目的原理并无二致。

好，我们马上开始今天的学习，首先我们得来讲解一下拓扑排序的依据——拓扑序是什么。

拓扑排序

“拓扑”序，光听名字你是不是感觉很抽象不好懂，我们看一个生活中的例子。

在大学，学生往往有更大的选课自由，可以按照自己的兴趣选择不同的课程、在不同的时间修读。但是，显然许多课程，和代码之间的依赖一样，也有着修读顺序的先后要求；有些课程还可能有不止一门先修课程，比如电磁场就要求先修读大学物理和数理方法两门课。

所以请问如果给了你一张这样的必修课表，每一行包含了课程本身的信息和它所依赖的其他课，你应该怎么安排你的课程学习顺序呢？这也是力扣上出镜率非常高的一道面试题课程表。

看表格不太清晰，这些课程之间的修读关系，显然可以用有向图来抽象，节点就是课程本身，边代表了两门课程的依赖关系。把课程表的信息用图的方式表示出来就会是这个样子：

照着图去选择修读顺序就会容易很多，你会优先选择没有入边也就是没有先修要求的课程，所以，等修读完线性代数和高等数学之后你就会发现，诶这个时候数理方法和大学物理的先修课程已经全部被你解锁了；继而你就可以继续修读这两门课程；最后再修读电磁场这门。

当然你会发现修读顺序依旧不只一种，比如：

1
2

线性代数、高等数学、数理方法、大学物理、电磁场
高等数学、线性代数、大学物理、数理方法、电磁场

这两种修读顺序的都是合理的，但它们一定保证了，要修读的课程的先修课程一定在当前课程的前面。

讲到这里，拓扑序的概念就清晰了，我们给出一个序列使得每个节点只出现一次，且保证如果存在路径P从A指向B，那么A在序列中一定出现在B之前；满足这个条件的序列就被称为满足拓扑序，所以它常常被用来解决有依赖关系任务的排序问题。

对于一个有向图而言，通常存在不止一种拓扑排序的方式。那有没有什么样的图是不能被拓扑排序的呢？当然是有的。假设有这样一张课表：

乍一看，可能和其他课表也没什么区别，但如果你把依赖关系同样表示在图上就会发现一些问题了。

它们之间构成了循环依赖的关系，你无法找到第一门修读的课程，所以这种先修课程的安排关系在现实世界中也是不存在的，这样的情况其实和我们常说的死锁有点像。你永远没有办法在一个有向有环的图中找到可以被拓扑排序的方案。

所以，所有拓扑排序都是建立在有向无环图（DAG）上的，DAG这个词相信很多同学都很眼熟，在Flink和Spark这类可以用来做数据批计算或者流计算的框架中，就常常可以见到DAG这样的概念，用来做计算任务的调度。

好，现在我们已经知道拓扑排序是什么了，也知道它可以用来解决有依赖关系任务的排序问题。那拓扑排序的具体实现步骤到底是什么样的呢？

拓扑排序如何实现

有两种经典的算法可以实现，一种叫Khan算法，基于贪心和广度优先算法BFS的思想来解决这个问题，另一种则是基于深度优先算法DFS的思想（如果你对DFS和BFS的概念还不够熟悉，可以再复习一下之前的章节）。

khan算法

Khan算法是更符合我们直觉的一种方法，也更容易理解，我们就先来讲解基于BFS的Khan算法如何解决课程表问题，也就是LeetCode 210，感兴趣的话你可以课后可以去网站上提交一下这道题。

题目是这样的：假设你总共有 numCourses 门课需要选，记为从 0 到 numCourses - 1，给你一个数组 prerequisites 存储课程之间的依赖关系，其中 prerequisites[i] = [ai, bi] ，表示在选修课程 ai 前必须先选修 bi 。

可以看到，输入的是课程间两两的依赖关系，事实上，这也是一种图的存储方式，prerequisites就是图上所有边的集合，我们一般称为边表。但是边表的存图方式不能让我们快速找到某个节点的后继节点，所以我们需要一种新的表示方式。

因为每个课程都是用 0 到 numCourses - 1 的数字作为课程标识，我们可以直接用一个二维数组next来表示图，next[i]里存储的是所有需要i作为先修课程的课程。这也是一种图的表示方法，我们一般叫做邻接表，通常可以用HashMap、数组、链表等数据结构实现。

相比于邻接矩阵存储所有节点之间是否存在边，在邻接表中，我们只用存储实际存在的边，所以在稀疏图中比较节约空间，实际应用和算法面试中都非常常见。

1
2
3
4
5
6
7

vector<vector<int>> next(numCourses, vector<int>());
        for (auto edge: prerequisites) {
            int n = edge[0];
            int p = edge[1];

            next[p].push_back(n);
        }

为了构建这样一张邻接表，我们要做的事就是遍历所有的先修关系，把edge[0]推入edge[1]所对应的课程列表也就是邻接表数组中。

输入处理完成，所有课程的先修关系图我们就有了。下一步就是选择合适的修读顺序。

相信你很自然就会想到，诶，那我们把所有不用先修课程的课都修完，然后从剩下的课程里找出接下来可修读的课程是不是就行了？之后按这样的顺序反复进行，直到所有课程都修读完毕。其实基于贪心思想的Khan算法就是这样做的。而这样一轮一轮遍历的感觉是不是也让你想到之前讲过的地毯式搜索的BFS呢？

我们一起来看整个题目Khan算法的实现：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

class Solution {
public:
    vector<int> findOrder(int numCourses, vector<vector<int>>& prerequisites) {
        // 邻接表数组
        vector<vector<int>> next(numCourses, vector<int>());
        // 入度表
        vector<int> pre(numCourses, 0);
        // 标记是否遍历过
        vector<int> visited(numCourses, 0);
        // 记录最终修读顺序
        vector<int> ans;

        // 构图
        for (auto edge: prerequisites) {
            int n = edge[0];
            int p = edge[1];

            next[p].push_back(n);
            pre[n]++;
        }

        queue<int> q;

        // 所有没有先修课程的课程入队
        for (int i = 0; i < numCourses; i++) {
            if (pre[i] == 0) {
                q.push(i);
            }
        }

        // BFS搜索
        while(!q.empty()) {
            int p = q.front();
            q.pop();
            ans.push_back(p);

            visited[p] = 1;
            // 遍历所有以队首课程为先修课程的课程
            for (auto n: next[p]) {
                // 由于队首课程已经被修读，所以当前课程入度-1
                pre[n]--;
                // 如果该课程所有先修课程已经修完；将该课程入队
                if (pre[n] == 0) {q.push(n);}
            }
        }

        // 环路检测： 如果仍有课程没有修读；说明环路存在
        for (int i = 0; i < numCourses; i++) {
            if (!visited[i]) return vector<int>();
        }

        return ans;
    }
};

我们用一个队列来表示所有满足修读条件的课程，一开始把所有没有先修课程的节点加入其中；再遍历出队，每次出队说明该课程已经修读过了，然后我们需要再遍历一下这个课程的所有后继课程，看看哪些在修完这门课程之后就可以修读了。

那哪些课程可以呢？这里我们非常聪明地用一个图的入度数组就解决了这个问题，入度，其实就是有向图上每个节点入边的数量，我们在代码里用一个计数数组就可以记录。

具体来说，我们给每个节点增加一个先修课程计数器pre，表示该课程有多少入度，也就是有多少先修课程，这一步操作，在遍历所有先修关系建图的时候就会完成。

所以之后每修读完一门课，也就是在出队的时候，会把这门课程的先修课计数器pre进行自减操作，如果发现哪个课程的pre已经为0了，那我们就可以把这个课程加入queue中了；因为这个时候这门课已经没有没修过的先修课程了。

按照这个方法遍历，所有课程的出队顺序显然就是一个合理的“拓扑序”的排列。

当然，我们还需要检查一下是不是所有课程都修读了，也就是代码的47-50行。如果有课程没修读完，其实就说明课表里有环存在，这不是一个合理的课表，按题目意思我们需要输出一个空集合。

Khan算法的全部思路和具体实现就学习到这里，如果你想更好地掌握该算法，建议你可以在力扣上找一些相关题目多做练习。

dfs算法

通过Khan算法的学习，相信你对拓扑排序已经有了非常具体的认知，我们来看看第二种基于DFS的拓扑排序算法，这也是html-webpack-plugin v3.2.0 所采用的策略，它又是如何解决Webpack打包HTML chunk的排序问题的呢？

先上代码，注释很清晰易懂，不过由于是前端框架，你可能需要稍微花一点时间熟悉一下JavaScript的语法，不做前端的同学不用太深究具体语义，就像我们前面说的，把chunks当作有依赖关系的节点就行，和课程表中的课程其实是很像的。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78

/**
  Sorts dependencies between chunks by their "parents" attribute.

  This function sorts chunks based on their dependencies with each other.
  The parent relation between chunks as generated by Webpack for each chunk
  is used to define a directed (and hopefully acyclic) graph, which is then
  topologically sorted in order to retrieve the correct order in which
  chunks need to be embedded into HTML. A directed edge in this graph is
  describing a "is parent of" relationship from a chunk to another (distinct)
  chunk. Thus topological sorting orders chunks from bottom-layer chunks to
  highest level chunks that use the lower-level chunks.

  @param {Array} chunks an array of chunks as generated by the html-webpack-plugin.
  - For webpack < 4, It is assumed that each entry contains at least the properties
  "id" (containing the chunk id) and "parents" (array containing the ids of the
  parent chunks).
  - For webpack 4+ the see the chunkGroups param for parent-child relationships

  @param {Array} chunks an array of ChunkGroups that has a getParents method.
  Each ChunkGroup contains a list of chunks in order.

  @return {Array} A topologically sorted version of the input chunks
*/
module.exports.dependency = (chunks, options, compilation) => {
  const chunkGroups = compilation.chunkGroups;
  if (!chunks) {
    return chunks;
  }

  // We build a map (chunk-id -> chunk) for faster access during graph building.
  const nodeMap = {};

  chunks.forEach(chunk => {
    nodeMap[chunk.id] = chunk;
  });

  // Next, we add an edge for each parent relationship into the graph
  let edges = [];

  if (chunkGroups) {
    // Add an edge for each parent (parent -> child)
    edges = chunkGroups.reduce((result, chunkGroup) => result.concat(
      Array.from(chunkGroup.parentsIterable, parentGroup => [parentGroup, chunkGroup])
    ), []);
    const sortedGroups = toposort.array(chunkGroups, edges);
    // flatten chunkGroup into chunks
    const sortedChunks = sortedGroups
      .reduce((result, chunkGroup) => result.concat(chunkGroup.chunks), [])
      .map(chunk => // use the chunk from the list passed in, since it may be a filtered list
    nodeMap[chunk.id])
      .filter((chunk, index, self) => {
        // make sure exists (ie excluded chunks not in nodeMap)
        const exists = !!chunk;
        // make sure we have a unique list
        const unique = self.indexOf(chunk) === index;
        return exists && unique;
      });
    return sortedChunks;
  } else {
    // before webpack 4 there was no chunkGroups
    chunks.forEach(chunk => {
      if (chunk.parents) {
        // Add an edge for each parent (parent -> child)
        chunk.parents.forEach(parentId => {
          // webpack2 chunk.parents are chunks instead of string id(s)
          const parentChunk = _.isObject(parentId) ? parentId : nodeMap[parentId];
          // If the parent chunk does not exist (e.g. because of an excluded chunk)
          // we ignore that parent
          if (parentChunk) {
            edges.push([parentChunk, chunk]);
          }
        });
      }
    });
    // We now perform a topological sorting on the input chunks and built edges
    return toposort.array(chunks, edges);
  }
};

在Webpack4之前，所有的chunks是有父子关系的，你可以认为父节点是需要依赖子节点的。Webpack在生成HTML的chunks时，需要按照拓扑序的方式一次遍历打包并嵌入到最终的HTML中，这样我们就可以把后面父chunk所用到的子chunk放在更前面的位置。

所以，59-73行，这一段代码其实就是用来把这个问题的边表提取出来的，我们通过forEach方法遍历了所有的chunks，然后把父子关系全部用edges变量记录下来，这和课程表的先修数组是一样的存图方式，也就是边表。

有了边表，我们就要对它排序了。真正的拓扑排序代码其实在第76行，调用了toposort.array方法，这个方法来自一个js的包toposort，它提供了非常干净的接口，输入一个DAG的所有节点和边作为参数，返回一个合法的拓扑序。

具体是怎么实现的呢？我们看一下toposort包的源码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97

/**
	 * Topological sorting function
	 *
	 * @param {Array} edges
	 * @returns {Array}
	 */
	
	module.exports = function(edges) {
	  return toposort(uniqueNodes(edges), edges)
	}
	
	module.exports.array = toposort
	
	function toposort(nodes, edges) {
	  var cursor = nodes.length
	    , sorted = new Array(cursor)
	    , visited = {}
	    , i = cursor
	    // Better data structures make algorithm much faster.
	    , outgoingEdges = makeOutgoingEdges(edges)
	    , nodesHash = makeNodesHash(nodes)
	
	  // check for unknown nodes
	  edges.forEach(function(edge) {
	    if (!nodesHash.has(edge[0]) || !nodesHash.has(edge[1])) {
	      throw new Error('Unknown node. There is an unknown node in the supplied edges.')
	    }
	  })
	
	  while (i--) {
	    if (!visited[i]) visit(nodes[i], i, new Set())
	  }
	
	  return sorted
	
	  function visit(node, i, predecessors) {
	    if(predecessors.has(node)) {
	      var nodeRep
	      try {
	        nodeRep = ", node was:" + JSON.stringify(node)
	      } catch(e) {
	        nodeRep = ""
	      }
	      throw new Error('Cyclic dependency' + nodeRep)
	    }
	
	    if (!nodesHash.has(node)) {
	      throw new Error('Found unknown node. Make sure to provided all involved nodes. Unknown node: '+JSON.stringify(node))
	    }
	
	    if (visited[i]) return;
	    visited[i] = true
	
	    var outgoing = outgoingEdges.get(node) || new Set()
	    outgoing = Array.from(outgoing)
	
	    if (i = outgoing.length) {
	      predecessors.add(node)
	      do {
	        var child = outgoing[--i]
	        visit(child, nodesHash.get(child), predecessors)
	      } while (i)
	      predecessors.delete(node)
	    }
	
	    sorted[--cursor] = node
	  }
	}
	
	function uniqueNodes(arr){
	  var res = new Set()
	  for (var i = 0, len = arr.length; i < len; i++) {
	    var edge = arr[i]
	    res.add(edge[0])
	    res.add(edge[1])
	  }
	  return Array.from(res)
	}
	
	function makeOutgoingEdges(arr){
	  var edges = new Map()
	  for (var i = 0, len = arr.length; i < len; i++) {
	    var edge = arr[i]
	    if (!edges.has(edge[0])) edges.set(edge[0], new Set())
	    if (!edges.has(edge[1])) edges.set(edge[1], new Set())
	    edges.get(edge[0]).add(edge[1])
	  }
	  return edges
	}
	
	function makeNodesHash(arr){
	  var res = new Map()
	  for (var i = 0, len = arr.length; i < len; i++) {
	    res.set(arr[i], i)
	  }
	  return res
	}