01｜中间件生态（上）：有哪些类型的中间件？

作者: 丁威

你好，我是丁威。

最近十年是互联网磅礴发展的十年，IT系统从单体应用逐渐向分布式架构演变，高并发、高可用、高性能、分布式等话题变得异常火热，中间件也在这一时期如雨后春笋般涌现出来，那到底什么是中间件呢？存在哪些类型的中间件呢？同一类型的中间件，我们该怎么选择？接下来的两节课，我们就来聊聊这些问题。

中间件的种类很多，我们无法把所有类型和产品列出来逐一讲解。但是每个类别的中间件在设计原理、使用上有很多共同的考量标准，只要了解了最重要、最主流的几种中间件，我们就可以方便地进行知识迁移，举一反三了，然后学习其他中间件将变得非常简单。

所以呢，你可以把这两节课看作是提纲挈领的知识清单。下面我们讲到的中间件你不一定都能够用上，但在需要的时候，可以帮你从更加高屋建瓴的角度迅速决策。

什么是中间件？

先来说说什么是中间件，我认为中间件是游离于业务需求之外，专门为了处理项目中涉及高可用、高性能、高并发等技术需求而引入的一个个技术组件。它的一个重要作用就是能够实现业务代码与技术功能之间解耦合。

这么说是不是还有点抽象？在这里定义里，我提到了业务需求和技术需求，关于这两个词我需要再解释一下。

业务需求，笼统地说就是特定用户的特定诉求。以我们快递行业为例：人与人之间需要跨城市传递物品，逢年过节我们需要给远方的亲人寄礼物，这就是所谓的业务需求。

技术需求，就是随着业务的不断扩展，形成规模效应后带来的使用上的需求。例如上面提到的寄件服务，原先只需要服务1万个客户，用户体验非常好，但现在需要服务几个亿的用户，用户在使用的过程中就会出现卡顿、系统异常等问题，因此产生可用性、稳定性方面的技术诉求。

为了解决各式各样的业务和技术诉求，代码量会越来越多。如果我们任凭业务代码与技术类代码没有秩序地纠缠在一起，系统会变得越来越不可维护，运营成本也会成指数级增加，故障频发，最终直接导致项目建设失败。

怎么解决这个问题呢？计算机领域有一个非常经典的分层架构思想，还有这样一句话“计算机领域任何一个问题都可以通过分层来解决，如果不行，那就再增加一层。”要想让系统做得越来越好，我们通常会基于分层的架构思想引入一个中间层，专门来解决可用性、稳定性、高性能方面的技术类诉求，这个中间层就是中间件，这也正是“中间件”这个词的来源。

中间件生态漫谈

明白了中间件的内涵，我们再来看看市面上有哪些中间件。我在开篇词中已经提到过了，中间件的种类繁多，我整理了一版分布式架构体系中常见的中间件，你可以先打开图片仔细看一看。

结合我10多年的从业经验，特别是对互联网主流分布式架构体系的研读，我发现微服务中间件、消息中间件、定时调度的使用频率极高，在解决分布式架构相关问题中是排头兵，具有无可比拟的普适性。这三者的设计理念和案例能对分布式、高可用和高并发等理念实现全覆盖。

所以，在专栏的第三章到第五章，我会深度剖析微服务、消息中间件和定时调度这三个方向，结合生产级经典案例深入剖析它们的架构设计理念，带你扎实地掌握分布式架构设计相关的基本技能。

• 微服务

具体而言，作为软件架构从单体应用向分布式演进出现的第一个新名词，微服务涉及分布式领域中服务注册、服务动态发现、RPC调用、负载均衡、服务聚合等核心技术，而Dubbo在微服务领域是当仁不让的王者。所以在微服务这一部分，我们会以Dubbo为例进行实战演练。

• 消息中间件

随着微服务的蓬勃发展，系统的复杂度越来越高，加上互联网秒杀、双十一、618等各种大促活动层出不穷，我们急切需要对系统解耦和应对突发流量的解决办法，这时候消息中间件应运而生了，它同样成为我们架构设计工作中最常用的工具包。常用的消息中间件包括RocketMQ、Kafka，它们在适用性上有所不同，如何保障消息中间件的稳定性是一大挑战。

• 定时调度

而定时调度呢？我们既可以认为它是个技术需求，也可以认为它是一个业务类需求，通过研读ElasticJob、XXL-Job等定时调度框架，可以很好地提升我们对业务需求的架构设计能力。

这三部分我们会在后面的模块中重点展开，所以这一模块不做深入讲解。接下来，为了让你对主流中间件有一个更全面的认知，我会分两节课对另外的几类中间件（数据库、缓存、搜索、日志等）进行简要阐述，以补全你的中间件知识图谱，帮助你更加有底气、有效率地进行决策。这节课，我们先来看看数据库中间件。

数据库中间件

数据库中间件应该是我们接触得最早也是最为常见的中间件，在引入数据库中间件之前，由于单体应用向分布式架构演进的过程中单表日数据急速增长，单个数据库的节点很容易成为系统瓶颈，无法提供稳定的服务。因此，为了解决可用性问题，在技术架构领域通常有如下两种解决方案：

• 读写分离；
• 分库分表。

我们先分别解析下这两个方案。最后再来看一看，引入数据库中间件给技术带来的简化。

读写分离

这是我在没有接触中间件之前，在一个项目中使用过的方案：

这个方案的实现要点有三个。

第一，在编写业务接口时，要通过在接口上添加注解来指示运行时应该使用的数据源。例如，@SlaveofDB表示使用Slave数据库，@MasterOfDB表示使用主库。

第二，当用户发起请求时，要先经过一个拦截器获取用户请求的具体接口，然后使用反射机制获取该方法上的注解。举个例子，如果存在@SlaveofDB，则往线程上下文环境中存储一个名为dbType的变量，赋值为slave，表示走从库；如果存在@MasterOfDB，则存储为master，表示走主库。

第三，在Dao层采用Spring提供的路由选择机制，继承自AbastractRoutingDataSource。应用程序启动时自动注入两个数据源(master-slave)，采用key-value键值对的方式存储。在真正需要获取链接时，根据上下文环境中存储的数据库类型，从内部持有的dataSourceMap中获取对应的数据源，从而实现数据库层面的读写分离。

总结一下，读写分离的思路就是通过降低写入节点的负载，将耗时的查询类请求转发到从节点，从而有效提升写入的性能。

但是，当业务量不断增加，单个数据库节点已无法再满足业务需求时，我们就要对数据进行切片，分库分表的技术思想就应运而生了。

分库分表

分库分表是负载均衡在数据库领域的应用，主要的原理你可以参考下面这张图。

简单说明一下。分库分表主要是通过引入多个写入节点来缓解数据压力的。因此，在接受写入请求后，负载均衡算法会将数据路由到其中一个节点上，多个节点共同分担数据写入请求，降低单个节点的压力，提升扩展性，解决单节点的性能瓶颈。

不过，要实现数据库层面的分库分表还是存在一定技术难度的。因为分库分表和读写分离一样，最终要解决的都是如何选择数据源的问题。所以在分库分表方案中，首先我们要有两个算法。

• 一个分库字段和分库算法，即在进行数据查询、数据写入时，根据分库字段的值算出要路由到哪个数据库实例上；
• 一个分表字段和分表算法，即在进行数据查询、数据写入时，根据分表字段的值算出要路由到哪个表上。

不管是上面的分库、还是分表都需要解决一个非常关键的问题：SQL解析。你可以看下面这张图。

如果订单库的分库字段设置为order_no，要想正确执行这条SQL语句，我们首先要解析这条SQL语句，提取order_no的字段值，再根据分库算法(负载均衡算法)计算应该发送到哪一个具体的库上执行。

SQL语句语法非常复杂，要实现一套高性能的SQL解析引擎绝非易事，如果按照上面我提供的解决方案，将会带来几个明显的弊端。

• 技术需求会污染业务代码，维护成本高

在业务控制器中需要使用注解来声明读写分离按相关的规则进行，随着业务控制的不断增加、或者读写分离规则的变化，我们需要对系统所有注解进行修改，但业务逻辑其实并没有改变。这就造成两者之间相互影响，后期维护成本较高。

• 技术实现难度较大，极大增加开发成本

由于SQL语句的格式太复杂、太灵活，如果不是数据库专业人才，很难全面掌握SQL语法。在这样的情况下，你写出的SQL解析引擎很难覆盖所有的场景，容易出现遗漏最终导致故障的发生；这也给产品的性能带来极大挑战。

那怎么办呢？其实，我们完全可以使用业界大神的开源作品来解决问题，这就要说到数据库中间件了。

引进数据库中间件

技术类诉求往往是相通的，极具普适性，为了解决上面的通病，根据分层的架构理念，我们通常会引入一个中间层，专门解决数据库方面的技术类需求。

MyCat 和ShardingJDBC/ShardingSphere是目前市面最主流的两个数据库中间件，二者各有优势。

MyCat服务端代理模式

先来看下MyCat代理数据库。它的工作模式可以用下面这张图概括：

面对应用程序，MyCat会伪装成一个数据库服务器(例如MySQL服务端)。它会根据各个数据库的通信协议，从二进制请求中根据协议进行解码，然后提取SQL，并根据配置的分库分表、读写分离规则计算出需要发送到哪个物理数据库。

随后，面对真实的数据库资源，MyCat会伪装成一个数据库客户端。它会根据通信协议将SQL语句封装成二进制流，发送请求到真实的物理资源，真实的物理数据库收到请求后解析请求并进行对应的处理，再将结果层层返回到应用程序。

这种架构的优势是它对业务代码无任何侵入性，应用程序只需要修改项目中数据库的连接配置就可以了，而且使用简单，易于推广。同时它也有劣势：

• 存在性能损耗

数据库中间件需要对应用程序发送过来的请求进行解码并计算路由，随后它还要再次对请求进行编码并转发到真实的数据库，这就增加了性能开销。

• 高度中心化，数据库中间件容易成为性能瓶颈

数据库中间件需要处理所有的数据库请求，返回结果都需要在数据库中进行聚合，虽然减少了后端数据库的压力，但中间件本身很容易成为系统的瓶颈，扩展能力受到一定制约。

• 代理层实现复杂，普适性差

数据库中间件本身的实现比较复杂，需要适配市面上各主流数据库，例如MySQL、Oracle等，通用性大打折扣。

ShardingJDBC客户端代理模式

下面我们再来看下ShardingJDBC客户端代理数据库。ShardingJDBC的工作模式如下图所示：

ShardingJDBC主要实现的是JDBC协议。实现JDBC协议，其实主要是面向java.sql.Datasource、Connection、ResultSet等对象编程。它通常以客户端Jar包的方式嵌入到业务系统中，ShardingjJDBC根据分库分表的配置信息，初始化一个ShardingJdbcDatasource对象，随后解析SQL语句来提取分库、分表字段值，再根据配置的路由规则选择正确的后端真实数据库，最后，ShardingJDBC用各种类型数据库的驱动包将SQL发送到真实的物理数据库上。

我们同样来分析一下这个方案的优缺点。

主要的优势有如下几点：

• 无性能损耗

ShardingJDBC使用的是基于客户端的代理模式，不需要对SQL进行编码解码等操作，只要根据SQL语句进行路由选择就可以了，没有太多性能损耗。

• 无单点故障、扩展性强

ShardingJDBC以Jar包的形式存在于项目中，其分布式特性随着应用的增加而增加，扩展性极强。

• 基于JDBC协议，可无缝支持各主流数据库

JDBC协议是应用程序与关系型数据库交互的业界通用标准，市面上所有关系型数据库都天然支持JDBC，故不存在兼容性问题。

当然缺点也很明显，对于分库分表，它没有一个统一的视图，运维类成本较高。举个例子，如果订单表被分成了1024个表，这时候如果你想根据订单编号去查询数据，必须人为计算出这条数据存在于哪个库的哪个表中，然后再去对应的库上执行SQL语句。

为了解决ShardingJDBC存在的问题，官方提供了ShardingSphere，其工作机制基于代理模式，与MyCat的设计理念一致，作为数据库的代理层，提供统一的数据聚合层，可以有效弥补ShardingJDBC在运维层面的缺陷，因此项目通常采用ShardingDBC的编程方式，然后再搭建一套ShardingSphere供数据查询。

在没有ShardingSphere之前，使用MyCat也有一定优势。MyCat对业务代码无侵入性，接入成本也比较低。但ShardingSphere弥补了ShardingJDBC对运维的不友好，而且它的性能损耗低、扩展性强、支持各类主流数据库，可以说相比MyCat已经占有明显的优势了。

所以如果要在实践生产中选择数据库中间件，我更加推荐ShardingJDBC。

除了上面的原因，从资源利用率和社区活跃度的角度讲，首先，MyCat的“前身”是阿里开源的Cobar，是数据库中间件的开山鼻祖，技术架构稍显古老，而ShardingJDBC在设计之初就可以规避MyCat的固有缺陷，摒弃服务端代理模式。代理模式需要额外的机器搭建MyCat进程，引入了新的进程，势必需要增加硬件资源的投入。

其次，ShardingJDBC目前已经是Apache的顶级项目，它的社区活跃度也是MyCat无法比拟的。一个开源项目社区越活跃，寻求帮助后问题得到解决的概率就会越大，越多人使用，系统中存在的Bug也更容易被发现、被修复，这就使得中间件本身的稳定性更有保障。

总结

好了，这节课就讲到这里，我们来做个小结。通过刚才的学习，我们知道了中间件的概念，它是为了解决系统中的技术需求，将技术需求与业务需求进行解耦，让我们专注于业务代码开发的一个个技术组件。中间件的存在，就是为了解决高并发、高可用性、高性能等各领域的技术难题。

在项目中，合理引用中间件能极大提升我们系统的稳定性、可用性，但同时也会提升系统维护的复杂度，对我们的技术能力提出了更高的要求，我们必须要熟练掌握项目中引用的各种中间件，深入理解其工作原理、实现细节，提高对中间件的驾驭能力，否则一旦运用不当，很可能给系统带来灾难性的故障。

为了让你对中间件有一个更加宏观的认识，我给你列举了市面最为常用的中间件。虽然现在新的中间件层出不穷，但在我看来，大都不超过我列的这几类。这节课我们重点讲了两个主流的数据库中间件，下节课，我们再来解读缓存、全文索引、分布式日志这几类中间件。

课后题

学完这节课，我也给你出两道课后题吧！

1.从数据库中间件的演变历程中，你能提炼出哪些分布式架构设计理念？

2.请你以订单业务场景，搭建一个2库2表的ShardingSphere集群，实现数据的插入、查询功能。

如果你想要分享你的修改或者想听听我的意见，可以提交一个 GitHub的push请求或issues，并把对应地址贴到留言里。我们下节课见！

02｜中间件生态（下）：同类型的中间件如何进行选型？

作者: 丁威

你好，我是丁威。

这节课，我们继续中间件生态的讲解。

缓存中间件

纵观整个计算机系统的发展历程，不难得出这样一个结论：缓存是性能优化的一大利器。

我们先一起来看一个用户中心查询用户信息的基本流程：

这时候，如果查找用户信息这个API的调用频率增加，并且在整个业务流程中，同一个用户的信息会多次被调用，那么我们可以引入缓存机制来提升性能：

也就是说，在UserService中引入一个LinkedHashMap结构的内存容器，用它存储已经查询到的数据。如果新的查询请求能命中缓存，那么我们就不需要再查询数据库了，这就降低了数据库的压力，将网络IO、磁盘IO转变为了直接访问内存，性能自然而然也提升了。

但上面这个方案实在算不上一个优秀的方案，因为它考虑得非常不全面，存在下面这几个明显的缺陷：内存容量有限、容易引发内存溢出，缓存在节点之间不一致，数据量非常庞大。

上面每一个问题都会带来巨大的影响，如果我们每做一个业务系统，都需要花这么多精力去解决这些技术问题，那这个成本也是不可估量的。为了解决与缓存相关的技术诉求，市面上也涌现出了一些非常优秀的中间件。缓存中间件经历了从本地缓存到分布式缓存的演变历程，我们先来看本地缓存中间件。

本地缓存中间件

本地缓存与应用属于同一个进程，主要的优势是没有网络访问开销，其中Ehcache、Guava Cache与Caffeine是Java领域当下比较知名的本地缓存框架。由于Ehcache比较耗磁盘空间，并且在进程宕机后容易造成缓存数据结构破坏，只能通过重建索引的方式进行修复，所以目前我们主要使用Guava Cache和Caffeine，他们之间并没有明显的优劣势。

尽管内部实现细节不同，但本地缓存中间件基本都需要包含下面三个功能。

• 支持大容量。
  

  它们基本都会采取内存+磁盘两级存储模型，其中内存存放热数据，磁盘存放全量数据。

• 过期/淘汰机制。
  

  评估缓存对性能提升程度的一个重要依据就是缓存的命中率。如果用户每次访问都无法命中缓存，相当于缓存没有起到效果，存储的数据都是“无用”的数据，只会带来存储空间的浪费。所以，必须引入缓存过期机制，删除不常用的数据。

• 基本的数据统计功能。
  

  监控数据的主要目的是检测当前缓存的工作状态是否健康，需要检测的内容包括缓存命中率、内存空间使用情况、磁盘空间使用情况等。

总的来说，本地缓存对单体应用非常友好，但对分布式应用就会显得有点浪费资源，为什么这么说呢？你可以先看看下面这张图。

在这张图中，当连续两次查询用户ID为1的用户信息时，受到负载均衡组件的影响，其中一个请求会转发到192.168.3.100，另外一个请求会转发到192.168.3.101。这样，同一个用户的信息会在两台机器上分别缓存一份数据。

而且，如果数据发生变化，也需要通知多台机器同时刷新缓存，这就造成了资源浪费。因此，本地缓存更适合存储一些变化频率极低，数据量较小的场景，诸如基础数据、配置了类型的数据缓存等。

分布式缓存中间件

本地缓存属于单进程管理的范畴，存在单点故障与资源瓶颈，无法应对数据的持续增长。为了适应分布式架构的特点，市面上也出现了一批基于内存存储的分布式存储框架。

由于分布式缓存与应用进程分属不同的进程，存在网络访问开销，所以几乎各个缓存中间件都是基于内存存储的系统，它们的存储容量受限于机器内存容量。

为了解决存储方面的瓶颈，各个分布式缓存中间件都支持集群部署。分布式缓存中间件中比较出名的非Redis与Memcached莫属。我们以Redis为例，来看一下经典的分布式缓存部署架构：

从这张图中，我们可以提取出下面几个要点。

首先，客户端通常会使用一致性哈希算法进行负载均衡，主要是为了提高节点扩容、缩容时的缓存命中率。

第二，Redis采用主从同步模式，这可以提升数据的存储可靠性。如果是像Memcache这种不能持久化的中间件，进程一旦退出，存储在内存中的数据将会丢失，就要重新从数据库加载数据，这会让大量流量在短时间内穿透到数据库，造成数据库层面不稳定。

第三，单台Redis受限于机器内存的容量限制，通常会采用集群部署，即每一个节点存储部分数据。

第四，为了提升 Redis 的 master-slave 高可用性能，降低由于master节点宕机导致的集群写入节点数量减少问题，通常会引入哨兵集群，使 master-slave 主从自动切换，进一步提升缓存中间件的高可用性。

那么，同为分布式缓存中间件，Redis和Memcached又有什么区别与联系呢？二者的共同点是，它们都是基于内存访问的高性能缓存存储系统，具有高并发、低延迟特性。

但它们的不同点也很多，我总结为了以下四点。

• 数据类型：Redis支持丰富的数据类型，不仅支持key-value的存储结构，还支持List、Set等复杂数据结构，而Memcache只支持简单的数据类型。
• 数据持久化：Redis支持基于AOF、快照两种数据持久机制，持久化带来的好处便是进程重启后数据不会丢失，能有效防止缓存被击穿的风险；Memcache不支持数据持久化。
• 分布式存储：Redis自身支持master-slave、Cluster两种分布式存储架构，而Memcache自身并不支持集群部署，需要使用一致性哈希算法来构建集群。
• 线程模型：Redis命令执行采用单线程，故Redis不适合大Value值的存储，但借助Redis单线程模型可以非常方便地实现分布式锁等功能；Memcache基于多线程运行模型，可以充分利用多核CPU的并发优势，提升资源的利用率。

讲了这么多，要一下记住可能有点难度，我给你画了两张图，总结了刚才不同中间件的差异、适用场景，你可以保存下来随时回顾：

一句话总结，缓存框架是不断在演进的，在项目中引入缓存相关的中间件技术绝对是一个明智之举。在数据量较少，并且变更不频繁时，我建议你采用本地缓存，其他情况建议使用分布式缓存。

那如何在Redis与Memcache中进行选型呢？虽然技术选型我们需要结合业务场景来看，但从上述功能的对比来看，Redis基本在各个对比项中对Memcache呈“压制”态势，所以多数情况下，我建议你使用Redis。

全文索引中间件

Elasticsearch是一个基于Apache Lucene的开源且支持全文搜索的搜索引擎。

Lucene被公认为迄今为止性能最强、功能最齐全的搜索引擎库。但Lucene只是一个类库，只提供单机版本的搜索功能，无法与分布式计算、分布式存储等协调展开工作。为了适应分布式的架构体系，Elasticsearch应运而生。

Elasticsearch提供了强大的分布式文件存储能力、分布式实时分析搜索能力、实时全文搜索能力、强大的集群扩展能力，PB级别的结构化和非结构化数据处理能力。

Elasticsearch在分布式架构中有两个最常见的应用场景，一个是宽表、解决跨库Join，另一个就是全文搜索。接下来我们分别展开介绍。

在数据库领域，如果一个表的数据量庞大，我们通常会引入分库分表技术以提高可用性。但这会带来一个新的问题，就是数据关联、报表等查询会变得无比复杂，性能也无法得到保障。

我们以订单场景为例。在一个订单中通常会包含多个商品，一个非常经典的设计策略是会创建t_order与t_order_item表，其中t_order_item是torder的子表。但如果我们使用了分库分表技术，关联查询将变得非常复杂：

看一下上面这张图片，想象一下，如果应用程序发送一条Join语句给数据库，会发生什么事情呢？

由于订单编号为1的订单信息存储在order_db_00中，但与这条订单关联的订单字表却存储在order_db_01中，而Join操作需要的笛卡尔积操作存在于不同的数据库实例中，所以我们就要将多个数据库中的数据统一加载到内存中。这就需要创建众多对象，如果需要加载的数据庞大，无疑会导致内存竞争，垃圾回收加剧，性能将直线下降。

我相信你一定能想到这个问题的解法：用ER分库思想，让具有关联性的表使用字段相同的分片算法。例如上面的示例，我们可以将t_order、t_order_item两个表的分库字段都设置为订单ID，这样一来，同一订单id的父子数据都在同一个数据库实例中，就避免了跨库Join，可以让性能得到很大提升。

但真实的应用场景比这个要复杂很多，面对的用户不同，他们的诉求也不一样。

我们还是说回订单系统。

• 从买家的角度出发，我们希望同一个买家的订单数据（父子关联表）能够采用同样的分库策略，以此保证同一个买家的订单关联数据存储在同一个库中，这样买家在查询订单时不必跨库。
• 但是如果采用这种策略，从商家的角度出发就会发现，商家在查询商家订单信息、商家日订单报表、月订单报表时要查询多个数据库，甚至可能产生跨库Join的风险。这无疑会降低性能，严重时会使整个数据库变得不可用。

用一句话概述就是，分库分表在面对多维度查询时将变得力不从心，那该如何解决呢？

我们通常会引入数据异构+宽表的设计方案：

我们需要引入Canal数据同步工具，订阅MySQL的Binglog，将增量数据同步到Elasticsearch中，实现数据访问层面的读写分离。

ElasticSearch另外一个场景就是全文搜索。

我们以电商场景为例，用户在购买商品之前通常需要输入一些关键字搜索出符合自己期望的数据，例如商品表的表结构如下图所示：

如果我们要查询关键字为“苹果电脑”，基于关系型数据库，我们通常会写出这样的SQL语句：

1

select * from goods a where a.goods_decribe like '%苹果电脑%'；

运行上述代码，如果商品数量少那倒没关系，但如果是淘宝、天猫、京东等一线电商平台，需要存储海量商品信息，在商品库中运行上述SQL，对数据库来说就是一个“噩梦”，因为上述语句并不会走索引，容易很快耗尽数据库链接而导致系统不可用。

这个时候，使用Elasticsearch就是一个非常明智的选择。因为Elasticsearch的底层是Lucene，可以对需要查找的字段建立索引，中间还会进行分词处理，进行更智能的匹配。由于Elasticsearch底层会为字段建立倒排索引，根据关键字查询可以轻松命中缓存，从而能极大提升访问性能，实现低延迟访问。

分布式日志中间件

随着微服务的兴起、业务量的增长，每一个服务在生产环境都会部署多台机器。例如，在我们公司，光是订单中心的“创建订单”服务就部署了四十多台机器。当遇到生产问题时，如果我们想要查看服务器日志，就会异常困难，因为我们根本不知道发生错误的请求具体在哪台机器上。

在机器数量较少（10台机器以内）的时候，通常我们可以使用Ansibe同时向所有需要采集的服务端执行日志检索命令，其工作示意图如下：

这种方式对于用户来说就像是操作单机模式一样，但是它的缺陷也是显而易见的。

• 基于Ansibe这种命令行等批量运维工具，需要保存目标机器的用户名与密码，安全性会受到影响。
• 如果要管理的目标机器有成百上千台，这种方式的系统开销会很大，搜索的响应时间很长，几乎是不太可能顺畅使用的。

为了进一步解决这个问题，我们通常需要采集每台服务器的日志，并将它存储在一个集中的地方，再提供一个可视化界面供用户查询。那么问题来了，市面上有这样的中间件吗？

我的回答是，必须得有，它就是大名鼎鼎的ELK。我们可以先看下这张ELK的工作架构图：

我们需要在需要进行日志采集的机器上安装一个filebeat工具，用来采集服务器的日志，并将它们存储到消息中间件中。然后，在需要采集的机器中安装Logstash进程，通过Logstash将日志数据存储到Elasticsearch服务器，用户可以通过Kibana查询存储在Elasticsearch中的日志数据，这样，我们就可以有针对性地查询所需要的日志了。

总结

好了，这节课就讲到这里。这节课，我们重点介绍了缓存、全文索引、分布式日志三类中间件。

缓存是性能优化的一柄利器，我们重点阐述了缓存技术从本地缓存到分布式缓存的演进之路，各种技术引入的背景以及解决方案，你可以根据自身情况，选择适合自己的缓存中间件。

另外，搜索相关技术也是应用系统必不可少的一环。随着微服务技术和数据库分库分表技术的兴起，数据写入效率大大提高，但与此同时，数据查询也面临更大的挑战，而基于Elasticsearch的数据异构架构方式能非常方便地解决数据查询的性能问题。

在分布式环境下，传统的应用日志查询方式也变得越来越难使用，ELK日志技术则为日志搜索带来了新气象，是分布式日志中间件的不二之选。

课后题

学完这节课，我也给你出一道课后题吧。

数据异构是一种非常经典的架构方式，请你尝试使用Canal或者Flink-CDC，将数据从MySQL同步到Elasticsearch中。

欢迎你在留言区与我交流讨论，我们下节课见。

03 | 数组与链表：存储设计的基石有哪些？

作者: 丁威

你好，我是丁威。

从这节课开始，我们就要进行基础篇的学习了。想要熟练使用中间件解决各种各样的问题，首先需要掌握中间件的基础知识。

我认为，中间件主要包括如下三方面的基础：数据结构、JUC和Netty，接下来的两节课，我们先讲数据结构。

数据结构主要解决的是数据的存储方式问题，是程序设计的基座。

按照重要性和复杂程度，我选取了数组和链表、键值对(HashMap)、红黑树、LinkedHashMap和PriorityQueue几种数据结构重点解析。其中，数组与链表是最底层的两种结构，是后续所有数据结构的基础。

我会带你分析每种结构的存储结构、新增元素和搜索元素的方式、扩容机制等，让你迅速抓住数据结构底层的特性。当然，我还会结合一些工业级实践，带你深入理解这些容器背后蕴含的设计理念。

说明一下，数据结构其实并不区分语言，但为了方便阐述，这节课我主要基于Java语言进行讲解。

数组

我们先来看下数组。

数组是用于储存多个相同类型数据的集合，它具有顺序性，并且也要求内存空间必须连续。高级编程语言基本都会提供数组的实现。

为了更直观地了解数组的内存布局，我们假设从操作系统申请了128字节的内存空间，它的数据结构可以参考下面这张图：

结合这张图我们可以看到，在Java中，数组通常包含下面几个部分。

• 引用：每一个变量都会在栈中存储数组的引用，我们可以通过引用对数组进行操作，对应上图的 array1、array2。
• 容量：数组在创建时需要指定容量，一旦创建，无法修改，也就是说，数组并不能自动扩容。
• 下标：数组可以通过下标对数组中的元素进行随机访问，例如array1[0]表示访问数组中的第一个元素，下标从0开始，其最大值为容量减一。

在后面的讲解中，你能看到很多数据结构都是基于数组而构建的。

那么数组有哪些特性呢？这里我想介绍两个我认为最重要的点：内存连续性和随机访问效率高。

我们先来看下内存连续性。

内存连续性的意思是，数组在向操作系统申请内存时，申请的必须是连续的内存空间。我们还是继续用上面这个例子做说明。我们已经创建了array1、array2两个数组，如果想要再申请一个拥有五个int元素的数组，能把这五个元素拆开，分别放在数组1的前面和后面吗？你可以看看下面这张示意图。

答案当然是不可以。

虽然当前内存中剩余可用空间为32个字节，乍一看上去有充足的内存。但是，因为不存在连续的20字节的空间，所以不能直接创建array3。

当我们想要创建20字节长度的array3时，在Java中会触发一次内存回收，如果垃圾回收器支持整理特性，那么垃圾回收器对内存进行回收后，我们就可以得到一个新的布局：

经过内存整理后就能创建数组3了。也就是说，如果内存管理不当，确实容易产生内存碎片，从而影响性能。

那我们为什么要把内存设计为连续的呢？换句话说，连续内存有什么好处呢？

这就不得不提到数组一个无可比拟的优势了：数组的随机访问性能极好。连续内存确保了地址空间的连续性，寻址非常简单高效。

举个例子，我们创建一个存放int数据类型的数组，代码如下：

1

int[] array1 = new int[10];

然后我们看下JVM中的布局：

可以看到，首先内存管理器在栈空间会分配一段空间，用它存储数组在物理内存的起始地址，这个起始地址我们用baseOffset表示。如果是64位操作系统，默认一个变量使用8字节，如果采用了指针压缩技术，可以减少到4字节。

数组能够高效地随机访问数组中的元素，主要原因是它能够根据下标快速计算出真实的物理地址，寻找算法为“baseOffset + index * size”。

其中,size为数组中单个元素的长度，是一个常量。在上面这个数组中，存储的元素是int类型的数据，所以size为4。因此，我们根据数组下标就可以迅速找到对应位置存储的数据。

数组这种高效的访问机制在中间件领域有着非常广泛的应用，大名鼎鼎的消息中间件RocketMQ在它的文件设计中就灵活运用了这个特性。

RocketMQ为了追求消息写入时极致的顺序写，会把所有主题的消息全部顺序写入到commitlog文件中。也就是说，commitlog文件中混杂着各个主题的消息，但消息消费时，需要根据主题、队列、消费位置向消息服务器拉取消息。如果想从commitlog文件中读取消息，则需要遍历commitlog文件中的所有消息，检索性能非常低下。

一开始，为了提高检索效率，RocketMQ引入了ConsumeQueue文件，可以理解为commitlog文件按照主题创建索引。

为了在消费端支持消息按tag进行消息过滤，索引数据中需要包含消息的tag信息，它的数据类型是String，索引文件遵循{topic}/{queueId}，也就是按照主题、队列两级目录存储。单个索引文件的存储结构设计如下图所示：

索引文件中，每一条消息都包含偏移量、消息长度和tag内容 3个字段。

• commitlog偏移量
  

  可以根据该值快速从commitlog文件中找到消息，这也是索引文件的意义。

• 消息长度
  

  消息的长度，知道它可以方便我们快速提取一条完整的消息。

• tag内容
  

  由于消息的tag是由用户定义的，例如tagA、createorder等，它的长度可变。在文件存储领域，一般存储可变长的数据，通常会采用“长度字段+具体内容”的存储方式。其中用来存储内容的长度使用固定长度，它是用来记录后边内容的长度。

回到消息消费这个需求，我们根据主题、消费组，消息位置(队列中存储的第N条消息)，能否快速找到消息呢？例如输入 topic:order_topic、queueId:0,offset:2，能不能马上找到第N条消息？

答案是可以找到，但不那么高效。原因是，我们根据topic、queueid，能非常高效地找到对应的索引文件。我们只需要找到对应的topic文件夹，然后在它的子目录中找到对应的队列id文件夹就可以了。但要想从索引文件中找到具体条目，我们还是必须遍历索引文件中的每一个条目，直到到达offset的条目，才能取出对应的commitlog偏移量。

那是否有更高效的索引方式呢？

当然有，我们可以将每一个条目设计成固定长度，然后按照数组下标的方式进行检索。

为了实现每一个条目定长，我们在这里不存储tag的原始字符串，而是存储原始字符串的hashCode，这样就可以确保定长了。你可以看看下面这张设计图：

基于这种设计，如果给定一个offset，我们再想快速提取一条索引就变得非常简单了。

首先，根据 offset * 20(每一个条目的长度)，定位到需要查找条目的起始位置，用startOffset表示。

然后，从startOffset位置开始读取20个字节的长度，就可以得到物理偏移量、消息长度和tag的hashCode了。

接着，我们可以通过hashCode进行第一次过滤，如果遇到hash冲突，就让客户端再根据消息的tag字符串精确过滤一遍。

这种方式，显然借鉴了数组高效访问数据的设计理念，是数组实现理念在文件存储过程中的经典运用。

总之，正是由于数组具有内存连续性，具有随机访问的特性，它在存储设计领域的应用才非常广泛，我们后面介绍的HashMap也引入了数组。

ArrayList

不过，数组从严格意义上来说是面向过程编程中的产物，而Java是一门面向对象编程的语言，所以，直接使用数组容易破坏面向对象的编程范式，故面向对象编程语言都会对数组进行更高级别的抽象，在Java中对应的就是ArrayList。

我会从数据存储结构、扩容机制、数据访问特性三个方面和你一起来探究一下ArrayList。

首先我们来看一下ArrayList的底层存储结构，你可以先看下这个示意图：

从图中可以看出，ArrayList的底层数据直接使用了数组，是对数组的抽象。

ArrayList相比数组，增加了一个特性，它支持自动扩容。其扩容机制如下图所示：

扩容的实现有三个要点。

• 扩容后的容量= 原容量 +（原容量）/ 2，以 1.5 倍进行扩容。
• 内部要创建一个新的数组，数组长度为扩容后的新长度。
• 需要将原数组中的内容拷贝到新的数组，即扩容过程中存在内存复制等较重的操作。

注意，只在当前无剩余空间时才会触发扩容。在实际的使用过程中，我们要尽量做好容量评估，减少扩容的发生。因为扩容的成本还是比较高的，存储的数据越多，扩容的成本越高。

接下来，我们来看一下ArrayList的数据访问特性。

• 顺序添加元素的效率高

ArrayList顺序添加元素，如果不需要扩容，直接将新的数据添加到elementData[size]位置，然后size加一即可（其中，size表示当前数组中存储的元素个数）。

ArrayList添加元素的时间复杂度为O(1)，也就是说它不会随着存储数据的大小而改变，是非常高效的存储方式。

• 中间位置插入/删除元素的效率低
  

在插入元素时，我们将需要插入数据的下标用 index 表示，将 index 之后的依次向后移动(复制到 index + 1)，然后将新数据存储在下标 index的位置。

删除操作与插入类似，只是一个数据是往后移，而删除动作是往前移。

ArrayList在中间位置进行删除的时间复杂度为O(n)，这是一个比较低效的操作。

• 随机访问性能高

由于ArrayList的底层就是数组，因此它拥有高效的随机访问数据特性。

LinkedList

除了ArrayList，在数据结构中，还有一种也很经典的数据结构：链表。LinkedList就是链表的具体实现。

我们先来看一下LinkedList的底层存储结构，最后再对比一下它和ArrayList的差异。

从上面这张图你可以看到，一个LinkedList对象在内存中通常由两部分组成：LinkedList对象和由Node节点组成的链条。

一个LinkedList对象在内存中主要包含3个字段。

• int size：链表中当前存在的Node节点数，主要用来判断是否为空、判断随机访问位点是否存在；
• Node first：指向链表的头节点；
• Node last：指向链表的尾节点。

再来说说由Node节点组成的链条。Node节点用于存储真实的数据，并维护两个指针。分别解释一下。

• E item：拥有存储用户数据；
• Node prev：前驱节点，指向当前节点的前一个指针；
• Node last：后继节点，指向当前节点的下一个节点。
  

由这两部分构成的链表具有一个非常典型的特征：内存的申请无须连续性。这就减少了内存申请的限制。

接下来我们来看看如何操作链表。对于链表的操作主要有两类，一类是在链表前后添加或删除节点，一类是在链表中间添加或删除数据。

当你想要在链表前后添加或删除节点时，因为我们在LinkedList对象中持有链表的头尾指针，可以非常快地定位到头部或尾部节点。也就是说，这时如果我们想要增删数据，都只需要更新相关的前驱或后继节点就可以了，具体操作如下图所示：

举个例子，如果我们向尾部节点添加节点，它的代码是这样的：

1
2
3
4
5
6

Node oldLastNode = list.last; //添加数据之前原先的尾部节点
Node newNode = new Node();
newNode.item = 4;//设置用户的值
oldLastNode.next = newNode; // 将原先尾部节点的next指针更新为新添加的节点
newNode.prev = oldLastNode; // 新添加的节点的prev指向源尾部节点，通过这两步，使新加入的节点添加到链表中
list.last = newNode; // 更新LinkedList的尾部节点为新添加节点

在链表的尾部、头部添加和删除数据，时间复杂度都是O(1)，比ArrayList在尾部添加节点效率要高。因为当ArrayList需要扩容时，会触发数据的大量复制，而LinkedList是一个无界队列，不存在扩容问题。

如果要在链表的中间添加或删除数据，我们首先需要遍历链表，找到操作节点。因为链表是非连续内存，无法像数组那样直接根据下标快速定位到内存地址。

例如，在下标index为1的后面插入新的数据，它的操作示例图如下：

我们从上往下看。插入新节点的第一步是需要从头节点开始遍历，找到下标为i=1的节点，然后在该节点的后面插入节点，最后执行插入节点的逻辑。

插入节点的具体实现主要是为了维护链表中相关操作节点的前驱与后继节点。

遍历链表、查询操作节点的时间复杂度为O(n)，然后基于操作节点进行插入与删除动作的时间复杂度为O(1)。

关于链表的知识点就讲到这里。由于链表与数组是数据结构中两种最基本的存储结构，为了让你更直观地了解二者的差异，我也给你画了一个表格，对两种数据结构做了对比：

HashMap

无论是链表还是数组都是一维的，在现实世界中有一种关系也非常普遍：关联关系。关联关系在计算机领域主要是用键值对来实现，HashMap就是基于哈希表Map接口的具体实现。

JDK1.8版本之前，HashMap的底层存储结构如下图所示：

HashMap的存储结构主体是哈希槽与链表的组合，类似一个抽屉。

我们向HashMap中添加一个键值对，用这个例子对HashMap的存储结构做进一步说明。

HashMap内部持有一个Map.Entry[]的数组，俗称哈希槽。当我们往HashMap中添加一个键值对时，HashMap会根据Key的hashCode与槽的总数进行取模，得出槽的位置(也就是数组的下标)，然后判断槽中是否已经存储了数据。如果未存储数据，则直接将待添加的键值对存入指定的槽；如果槽中存在数据，那就将新的数据加入槽对应的链表中，解决诸如哈希冲突的问题。

在HashMap中，单个键值对用一个Map.Entry结构表示，具体字段信息如下。

• K key：存储的Key，后续可以用该Key进行查找
• V value：存储的Value；
• int hash：Key的哈希值；
• Ma.Entry ：next 链表。

到这里，你可以停下来思考一下，当哈希槽中已经存在数据时，新加入的元素是存储在链表的头部还是尾部呢？

答案是放在头部。代码如下：

1
2
3
4

//假设新放入的槽位下标用 index 表示,哈希槽用 hashArray 表示
Map.Entry newEntry = new Map.Entry(key,value);
newEntry.next = hashArray[index];
hashArray[index] = newEntry;

我们将新增加的元素放到链表的头部，也就是直接放在哈希槽中，然后用next指向原先存在于哈希槽中的元素。

这种方式的妙处在于，只涉及两个指针的修改。如果我们把新增加的元素放入链表的头部，链表的复杂度为O(1)。相反，如果我们把新元素放到链表的尾部，那就需要遍历整条链表，写入复杂度会有所提高，随着哈希表中存储的数据越来越多，那么新增数据的性能将随着链表长度的增加而逐步降低。

介绍完添加元素，我们来看一下元素的查找流程，也就是如何根据Key查找到指定的键值对。

首先，计算Key的hashCode，然后与哈希槽总数进行取模，得到对应哈希槽下标。

然后，访问哈希槽中对应位置的数据。如果数据为空，则返回“未找到元素”。如果哈希槽对应位置的数据不为空，那我们就要判断Key值是否匹配了。如果匹配，则返回当前数据；如果不匹配，则需要遍历哈希槽，如果遍历到链表尾部还没有匹配到任何元素，则返回“未找到元素”。

说到这里，我们不难得出这样一个结论：如果没有发生哈希槽冲突，也就是说如果根据Key可以直接命中哈希槽中的元素，数据读取访问性能非常高。但如果需要从链表中查找数据，则性能下降非常明显，时间复杂度将从O(1)提升到O(n)，这对查找来说就是一个“噩梦”。

一旦出现这种情况，HashMap的结构会变成下面这个样子：

怎么解决这个问题呢？JDK的设计者们给出了两种优化策略。

第一种，对Hash槽进行扩容，让数据尽可能分布到哈希槽上，但不能解决因为哈希冲突导致的链表变长的问题。

第二种，当链表达到指定长度后，将链表结构转换为红黑树，提升检索性能(JDK8开始引入)。

我们先来通过源码深入探究一下HashMap的扩容机制。HashMap的扩容机制由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

   /**
     * Initializes or doubles table size.  If null, allocates in
     * accord with initial capacity target held in field threshold.
     * Otherwise, because we are using power-of-two expansion, the
     * elements from each bin must either stay at same index, or move
     * with a power of two offset in the new table.
     *
     * @return the table
     */
 &nbsp; &nbsp;final Node<K,V>[] resize() {
 &nbsp; &nbsp; &nbsp; &nbsp;Node<K,V>[] oldTab = table;
 &nbsp; &nbsp; &nbsp; &nbsp;int oldCap = (oldTab == null) ? 0 : oldTab.length;
 &nbsp; &nbsp; &nbsp; &nbsp;int oldThr = threshold;
 &nbsp; &nbsp; &nbsp; &nbsp;int newCap, newThr = 0;
 &nbsp; &nbsp; &nbsp; &nbsp;if (oldCap > 0) {
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;if (oldCap >= MAXIMUM_CAPACITY) {
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;threshold = Integer.MAX_VALUE;
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;return oldTab;
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;  }
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;else if ((newCap = oldCap << 1) < MAXIMUM_CAPACITY &&
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; oldCap >= DEFAULT_INITIAL_CAPACITY)
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;newThr = oldThr << 1; // double threshold
 &nbsp; &nbsp; &nbsp;  }
 &nbsp; &nbsp; &nbsp; &nbsp;else if (oldThr > 0) // initial capacity was placed in threshold
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;newCap = oldThr;
 &nbsp; &nbsp; &nbsp; &nbsp;else { &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; // zero initial threshold signifies using defaults
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;newCap = DEFAULT_INITIAL_CAPACITY;
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;newThr = (int)(DEFAULT_LOAD_FACTOR * DEFAULT_INITIAL_CAPACITY);
 &nbsp; &nbsp; &nbsp;  }
 &nbsp; &nbsp; &nbsp; &nbsp;if (newThr == 0) {
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;float ft = (float)newCap * loadFactor;
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;newThr = (newCap < MAXIMUM_CAPACITY && ft < (float)MAXIMUM_CAPACITY ?
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;  (int)ft : Integer.MAX_VALUE);
 &nbsp; &nbsp; &nbsp;  }
 &nbsp; &nbsp; &nbsp; &nbsp;threshold = newThr;
 &nbsp; &nbsp; &nbsp; &nbsp;@SuppressWarnings({"rawtypes","unchecked"})
 &nbsp; &nbsp; &nbsp; &nbsp;Node<K,V>[] newTab = (Node<K,V>[])new Node[newCap];
 &nbsp; &nbsp; &nbsp; &nbsp;table = newTab;
 &nbsp; &nbsp; &nbsp;//此处省略数据复制相关代码
 &nbsp;  }

为了方便你对代码进行理解，我画了一个与之对应的流程图：

总结一下扩容的要点。

• HashMap的容量并无限制，但超过2的30次幂后不再扩容哈希槽。
• 哈希槽是按倍数扩容的。
• HashMap在不指定容量时，默认初始容量为16。

HashMap并不是在无容量可用的时候才扩容。它会先设置一个扩容临界值，当HashMap中的存储的数据量达到设置的阔值时就触发扩容，这个阔值用threshold表示。

我们还引入了一个变量loadFactor来计算阔值，阔值=容量*loadFactor。其中，loadFactor表示加载因子，默认为0.75。

加载因子的引入与HashMap哈希槽的存储结构与存储算法有关。

HashMap在出现哈希冲突时，会引入一个链表，形成“数组+链表”的存储结构。这带来的效果就是，如果HashMap有32个哈希槽，当前存储的数据也刚好有32个，这些数据却不一定全会落在哈希槽中，因为可能存在hash值一样但是不同Key的数据，这时，数据就会进入到链表中。

前面我们也提到过，数据放入链表就容易引起查找性能的下降，所以，HashMap的设计者为了将数据尽可能地存储到哈希槽中，会提前进行扩容，用更多的空间换来检索性能的提高。

我们再来看一下扩容的下半部分代码。

我们先来看下这段代码：

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

@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 { // preserve order
                Node<K,V> loHead = null, loTail = null;
                Node<K,V> hiHead = null, hiTail = null;
                Node<K,V> next;
                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;
                }
            }
        }
    }
}

这段代码不难理解，就是按照扩容后的容量创建一个新的哈希槽数组，遍历原先的哈希槽(数组)，然后将数据重新放入到新的哈希槽中，为了保证链表中数据的顺序性，在扩容时采用尾插法。

除了扩容，JDK8之后的版本还有另外一种提升检索能力的措施，那就是在链表长度超过8时，将链表演变为红黑树。这时的时间复杂度为O(2lgN)，可以有效提升效率。

关于红黑树，我会在下节课详细介绍。

总结

这节课，我们介绍了数组、ArrayList、LinkedList、HashMap这几种数据结构。

数组，由于其内存的连续性，可以通过下标的方式高效随机地访问数组中的元素。

数组与链表可以说是数据结构中两种最基本的数据结构，这节课，我们详细对比了两种数据结构的存储特性。

哈希表是我们使用得最多的数据结构，它的底层的设计也很具技巧性。哈希表充分考虑到数组与链表的优劣，扬长避短，HashMap就是这两者的组合体。为了解决链表检索性能低下的问题，HashMap内部又引入了扩容与链表树化两种方式进行性能提升，提高了使用的便利性，降低了使用门槛。

课后题

最后，我也给你留两道思考题吧！

1、业界在解决哈希冲突时除了使用链表外，还有其他什么方案？请你对这两者的差异进行简单的对比。

2、HashMap中哈希槽的容量为什么必须为2的倍数？如果不是很理解，推荐你先学习一下位运算，然后在留言区告诉我你的答案。

我们下节课再见！

04 | 红黑树：图解红黑树的构造过程与应用场景

作者: 丁威

你好，我是丁威。

这节课，我们继续Java中常用数据结构的讲解。我会重点介绍TreeMap、LinkedHashMap和 PriorityQueue这三种数据结构。

TreeMap

先来看TreeMap。TreeMap的底层数据结构是一棵红黑树，这是一种比较复杂但也非常重要的数据结构。它是由树这种基础的数据结构演化而来的。

我们知道，在计算机领域，树指的就是具有树状结构的数据的集合。把它叫做“树”，是因为它看起来像一棵自上而下倒挂的树。一棵树通常有下面几个特点：

• 每个节点都只有有限个子节点或无子节点；
• 没有父节点的节点称为根节点；
• 每一个非根节点有且只有一个父节点；
• 除了根节点外，每个子节点可以分为多个不相交的子树；
• 树里面没有环路（cycle）。

如果一棵树的每个节点最多有两个子树，那它就是一棵二叉树。二叉树是“树”的一个重要分支，我们可以通过文稿中这张图来直观感受一下：

但是如果数据按照这样的结构存储，想要新增或者查找数据就需要沿着根节点去遍历所有的节点，这时的效率为O(n)，可以看出性能非常低下。作为数据结构的设计者，肯定不能让这样的事情发生。

这时候，我们就需要对数据进行排序了，也就是使用所谓的二叉排序树（二叉查找树）。它有下面几个特点：

• 若任意节点的左子树不为空，则左子树上所有节点的值均小于它的根节点的值；
• 若任意节点的右子树不为空，则右子树上所有节点的值均大于它的根节点的值；
• 没有键值相等的节点。

如果上图这棵二叉树变成一棵二叉排序树，可能长成下面这个样子：

基于排序后的数据存储结构，我们来尝试一下查找数字30：

• 从根节点37开始查找，判断出37比30大，然后尝试从37的左子树继续查找；
• 37的左子节点为26，判断出26比30小，所以需要从26的右子树继续查找；
• 26的右子节点为32，由于32比30大，所以从32的左子树继续查找；
• 32的左子节点为30，命中，结束。

你应该已经发现了，每次查找，都可以排除掉一半的数据。我们可以将它类比作二分查找算法，其时间复杂度为O(logN)，也就是对数级。所以说，二叉排序树是一种比较高效的查找算法。

不过，二叉排序树也有缺陷。一个最主要的问题就是，在查找之前我们需要按照二叉排序树的存储特点来构建它。我们还是用上面这个例子，将节点按照从小到大的顺序构建二叉排序树，构建过程如下图所示：

根据排序二叉树的构建规则，如果数据本身是顺序的，那么二叉排序树会退化成单链表，时间复杂度飙升到O(n)，我们显然不能接受这种情况。

对比这两棵二叉排序树，第一棵左右子树比较对称，两边基本能保持平衡，但第二棵严重地向右边倾斜，这会导致每遍历新的一层，都无法有效过滤一半的数据，也就意味着性能的下降。

那有没有一种办法能够自动调整二叉排序树的平衡呢？这就是红黑树要解决的问题了。

红黑树是一种每个节点都带有颜色属性（红色或黑色）的二叉查找树，它可以实现树的自平衡，查找、插入和删除节点的时间复杂度都为O(logn)。

除了要具备二叉排序树的特征外，红黑树还必须具备下面五个特性。

性质1：节点是红色或黑色。

性质2：根是黑色。

性质3：所有叶子都是黑色（叶子是NIL节点）。

性质4：每个红色节点必须有两个黑色的子节点。也就是说，从每个叶子到根的所有路径上不能有两个连续的红色节点。

性质5：从任一节点到其每个叶子的所有简单路径都包含相同数目的黑色节点。

由于插入、删除节点都有可能破坏红黑树的这些特性，所以我们需要进行一些操作，也就是通过树的旋转让它重新满足这些特点。

树的旋转又分为右旋和左旋两种：右旋指的是旋转后需要改变支点节点的右子树，左旋指的是旋转后需要改变支点节点的左节点。 这个通过旋转重新满足特性的过程就是自平衡。树越平衡，数据的查找效率越高。

为了让你直观地看到“红黑树的魅力”，我们还是沿用上面的例子，将节点按照从小到大的顺序依次插入到一棵红黑树中，最终产生的红黑树为如下图所示：

这是一棵地地道道的二叉排序树。

但是我们刚才说，在查找元素时，时间复杂度从O(n)飙升到了O(logN)，这棵树是如何做到节点顺序插入时没有退化成链表的呢？我们一起来看下红黑树的构建过程。

提前说明一下，由于从小到大排序是一种特殊情况，不能覆盖建构红黑树的多种情况，所以为了更好地说明红黑树的工作机制，我们把节点的插入顺序变更为50、37、70、35、25、30、26、80、90、100、20、18、32、75、85。

1. 按照这个顺序，首先我们连续插入节点50、节点37、节点70，其初始状态如下图所示：

2. 然后，继续插入节点35：

这个时候，新插入的节点0035的父节点(00037)和叔叔节点(0070)都是红色，所以我们需要将0035的祖父节点的颜色传递到它的两个子节点，这样也就到了图里的第二个状态。由于根节点的颜色为红色，不符合红黑树的特点，我们再将根节点的颜色变更为黑色。

3. 继续插入节点25****：

可以看到，初始状态的当前节点、父节点和祖先节点的形状为一条斜线。这时红色节点0025与0035都是红色，违背了红黑树的性质4，这种情况可以使用右旋来解决，具体操作是：

• 让当前节点(0025)的祖先节点(0037)下沉，作为当前节点的父节点(0035)的右子节点。同时，当前父节点（0025）的祖先节点（0050）的左节点指向当前节点的父节点，这样，0050的左节点就直接指向了0035。本轮操作后变成图里的第二个状态。
• 旋转之后0035节点的右子树路径多了一个黑色的节点0037，为了符合红黑树的特性，我们需要将0037父节点的颜色进行翻转，变成图里的第三个状态。

总结一下，右旋的第一个触发条件：当前节点与父亲节点为红色，并且都是左节点。

4. 继续插入节点30：

当前节点(0030)、父节点(0025)和叔叔节点(0037)都为红色，所以可以将当前节点的祖先(0035)的状态传递给子节点，变成上图第二个状态。

5. 继续插入节点26：

可以看到，现在的状态是，当前节点(0026)和父节点(0030)为红色，当前节点为左子树，父节点为右子树，并且叔叔节点并不为红色（组成一个大于号）。

这时候我们也需要右旋，以当前节点为支点，将其父节点作为当前节点的右节点，当前节点重新充当其祖父节点的右节点，状态从图一转为图二。

这是右旋的第二个触发条件：当前节点、父节点、祖父节点的形状为大于号，而且当前节点的父节点为支点。

状态变为图二之后，当前节点(0030)与父节点(0026)都是红色，并且都是右节点，所以应该执行一次左旋。以父节点0026为支点，将当前节点（0030）的祖父节点（0025）变为父节点（0026）的左子节点，经过这个动作后，状态从图二转为图三。

左旋之后，黑色节点0025变成了节点0026的左子树，左子树的黑色节点数量变多，所以我们需要将黑色传递到父节点，也就是要把节点0025变为红色，0026变为黑色，变成图中的第四个状态。

6. 我们接着插入节点80，此时不会改变红黑色特性，再插入节点90：

由于当前节点与父节点都是红色，并且都是右节点，需要执行左旋。

其实，到底什么时候需要左旋，什么时候需要右旋你没有必要死记硬背。因为左旋、右旋的最终目的是要满足树的平衡，也就是降低树的层级。只要确保旋转后的最终效果满足二叉排序树的定义（根节点比左子树大，比右子数小）就可以了。

7. 继续插入节点100、20：

到这里我们就需要说明一下了。

这一步和步骤2一样，当前节点、父节点和叔叔节点都是红色，只需要将当前节点的祖父节点的颜色传递到祖父节点的两个子节点就可以了，这就到了图中的第二个状态。

但这个时候，0026和它的父节点0035同为红色，并且叔叔节点也是红色，我们需要再像上面一样传递颜色，调整后变成图里的第三个状态。

最后，由于根节点是红色，我们需要将根节点转为黑色。

这里重点强调的是，无论是左旋、右旋还是变色，都需要再次向上递归进行验证。

8. 继续插入18、32、75、85等节点：

到这一步基本没有什么新的知识点了，按照我们前面所讲过的方法进行调整，就可以得到上面这棵红黑树了。

红黑树的构建过程就介绍到这里。红黑树的主要过程就是通过为节点引入颜色、左旋、右旋、变色等手段实现树的平衡，保证查询功能高效有序进行。

聊完数据结构，我们再来看看它的应用。其实，TreeMap在中间件开发领域的运用非常广泛，其中最出名的估计要属使用TreeMap实现一致性哈希算法了。

下面是一致性哈希算法的示意图：

其中，Node1、Node2、Node3是真实存储的有效数据，每一个节点需要存储一些关联信息，很适合key-value的存储形式。一致性哈希算法的查询规则是：查询第一个大于目标哈希值的节点。

例如，如果输入key1，key2，需要命中Node2，如果输入key3，则需要命中Node3。

这种情况其实就是需要将数据按照key进行排序，而TreeMap中的数据本身就是顺序的，所以非常适合这个场景。

在RocketMQ中，就使用了一致性哈希算法来实现消费组队列的负载均衡。

TreeMap的TailMap是返回大于等于key的子树，然后调用子树的firstKey获取TreeMap中最小的元素，符合一致性哈希算法的命中规则。又因为TreeMap是一棵排序树，所以得到最小、最大值会非常容易。

在TreeMap中实现firstkey方法时，内部会先获取TreeMap中的键值对，也就是Entry对象：

然后从根节点开始遍历，查找到节点的左子树，再一直遍历到树的最后一个左节点，时间复杂度为O(logN)。

LinkedHashMap

红黑树就介绍到这里了，接下来我们再来看一个与LRU相关的数据结构LinkedHashMap。

LinkedHashMap是LinkedList和HashMap的结合体，它内部的存储结构可以简单表示为下面这样：

LinkedHashMap内部存储的Entry在HashMap的基础上增加了两个指针：before和after。这两个节点可以对插入的节点进行链接，以此来维护顺序性。同时，链表结构为了方便插入，也会持有“头尾节点”这两个指针。

那引入链表有什么好处呢？

我认为大概有下面两个优点。

一个是降低了遍历实现的复杂度。我们对比一下，HashMap的遍历是首先遍历哈希槽，然后遍历链表；但LinkedHashMap则可以基于头节点遍历，复杂度明显降低。引入链表的第二个优点则是提供了顺序性。接下来，我们就来看看LinkedHashMap的顺序性和使用场景。

LinkedHashMap提供了两种顺序性机制：

• 按节点插入顺序，是LinkedHashMap的默认行为；
• 按节点的访问性顺序，最新访问的节点将被放到链表的末尾。

它的使用场景也很常见，有一种知名的淘汰算法叫LRU。顾名思义，LRU就是要淘汰最近没有使用的数据。在Java领域，实现LRU的首选就是LinkedHashMap，因为LinkedHashMap能够按访问性排序。

在LinkedHashMap中，如果顺行性机制选择“按访问顺序”，那么当元素被访问时，元素会默认被放到链表的尾部，并且在向LinkedHashMap添加元素时会调用afterNodeInsertion方法。这个方法的具体实现代码如下：

从代码中可以看出，如果removeEldestEntry函数返回true，则会删除LinkedHashMap中的第一个元素，这样就淘汰了旧的数据，实现了LRU的效果。removeEledestEntry方法的代码如下：

可以看到，默认返回的是false，表示LinkedHashMap并不会启用节点的淘汰机制。为了实现LRU算法，我们需要继承LinkedHashMap并重写该方法，具体实现代码如下：

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

package net.codingw.datastruct;
import java.util.LinkedHashMap;
import java.util.Map;
public class LRUCache<K,V> extends LinkedHashMap<K,V> {
 &nbsp; &nbsp;private int maxCapacity;
 &nbsp; &nbsp;protected boolean removeEldestEntry(Map.Entry<K,V> eldest) {
 &nbsp; &nbsp; &nbsp; &nbsp;//如果超过了最大容量，则启动剔除机制
 &nbsp; &nbsp; &nbsp; &nbsp;return size() >= maxCapacity;
 &nbsp;  }
 &nbsp; &nbsp;public void setMaxCapacity(int maxCapacity) {
 &nbsp; &nbsp; &nbsp; &nbsp;this.maxCapacity = maxCapacity;
 &nbsp;  }
}

LinkedHashMap就介绍到这里了，我们再来看一种特殊的队列——优先级队列，它是实现定时调度的核心数据结构。

PriorityQueue

我们知道，普通队列都是先进先出的，但优先级队列不同，它可以为元素设置优先级，优先级高的元素完全可以后进先出。

我们先来看一下PriorityQueue的类图：

优先级队列的底层结构是数组，可是怎么在数组的基础上排列优先级呢？原来，PriorityQueue的底层是基于最小堆实现的堆排序。

所谓最小堆指的是一棵经过排序的完全二叉树，根结点的键值是所有堆结点键值中最小者。无论是最大堆还是最小堆，都只固定根节点与子节点的关系，两个子节点之间的关系并不做强制要求。

我们采用数组作为最小堆的底层数据结构，将最小堆用一棵二叉树来表示，这时的数据是按照从上到下、从左到右的方式依次存储在数组中的：

这种存储方式有两个特点：

• 假设一个节点在数组中的下标为n，则它的左子节点的下标为2n+1，它的右子节点的下标为2n+2；
• 假设一个节点在数组中的下标为n，那么它的父节点下标为 (n -1) >>> 1。

有些最小堆的存储方式是将数组的第一个元素空出来，把根节点存储在下标为1的位置。如果基于这种方式，存储有下面两个特点：

• 假设一个节点在数组中的下标为n，则它的左子节点的下标为2n，它的右子节点的下标为2n+1；
• 假设一个节点在数组中的下标为n，则它的父节点下标为 (n) >>> 1。

但在实践场景中，数据不可能按顺序插入，既然如此，要实现优先级队列，该怎么对这棵树进行排序呢？

PriorityQueue队列的实现中采用了堆排序。我们还是用图解的方式来看一下构建规则。

首先我们连续插入节点 500，600，700，800，其内部结构如下图所示：

由于首先插入了根节点为500，后续600，700比根节点都小，所以600和700可以直接成为根节点的左右子树。

继续插入800，由于比根节点大，同时比600大，则直接放入到600的子节点即可。

继续插入490，插入过程如下图所示：

解释一下。我们首先将新元素插入到数组的最后，下标为n=5，队列是图中的第一个状态。

根据公式 n >>> 1 ，可以算出它的父节点的下标为2，比较两者的大小，如果新插入的节点比父节点少，那么交换两者的值，变化到图中的第二个状态。

这时候，我们再通过公式n>>>1算出父节点的下标为1，比较两者的值，发现子节点的值比父节点的值低，则继续交换两者的值，成为图中的第三个状态。

要实现上面的步骤，我们相应的代码是：

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

private void siftUp(int k, E x) {
    if (comparator != null)
        siftUpUsingComparator(k, x);
    else
        siftUpComparable(k, x);
}
@SuppressWarnings("unchecked")
private void siftUpComparable(int k, E x) {
    Comparable<? super E> key = (Comparable<? super E>) x;
    while (k > 0) {
        int parent = (k - 1) >>> 1;
        Object e = queue[parent];
        if (key.compareTo((E) e) >= 0)
            break;
        queue[k] = e;
        k = parent;
    }
    queue[k] = key;
}

在这段代码里，我们首先使用while(k>0)实现递归，因为最小堆是将新插入的节点放在叶子结点，然后不断与其父节点进行比较，直到到达根节点。

然后，我们要根据当前节点的序号，计算其父节点的序号(这里的算法与图解方式不一样，是因为PriorityQueue是将根节点的下标定为0)，然后比较大小：

• 如果当前节点比父节点的值大，则跳出循环，符合最小堆的要求；
• 如果当前节点比父节点的值小，则交换两者的值，将k的值赋值为父节点(k = parent)，然后继续向上递归做判断。

构建好堆之后，我们再来看看怎么从堆中获取数据。要注意的是，访问数据只能从堆的根节点开发方法，具体做法就是删除根节点，并将根节点的值返回。我们先尝试删除根节点490：

删除根节点和删除其他任何节点的算法是一样的：

• 首先，我们将待删除的位置的值清除，状态为图一。
• 然后，将数组最后的元素移动到待删除位置，我们用下标n表示。删除根节点，n为0，状态转为图二。
• 接下来，根据下标算法分别算出其子节点的下标为2n、2n+1，从左右节点中挑选最小值，如图三。
• 用父节点的值与左右子节点中最小的值进行对比，如图四。如果父节点比最小子节点大，则交换两者的值，如图五。
• 我们要一直往下递归，直到节点没有子节点，或者没有父节点比子节点小为止。

结合这张图，我们同样来看一下PriorityQueue中删除元素的代码：

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 siftDown(int k, E x) {
    if (comparator != null)
        siftDownUsingComparator(k, x);
    else
        siftDownComparable(k, x);
}
@SuppressWarnings("unchecked")
private void siftDownComparable(int k, E x) {
    Comparable<? super E> key = (Comparable<? super E>)x;
    int half = size >>> 1;        // loop while a non-leaf
    while (k < half) {
        int child = (k << 1) + 1; // assume left child is least
        Object c = queue[child];
        int right = child + 1;
        if (right < size &&
            ((Comparable<? super E>) c).compareTo((E) queue[right]) > 0)
            c = queue[child = right];
        if (key.compareTo((E) c) <= 0)
            break;
        queue[k] = c;
        k = child;
    }
    queue[k] = key;
}

可以看到，在这段代码中，我们设定half为size的一半，如果下标大于half，则下标对应的位置不会再有子节点，可以跳出循环。

代码的第12行是计算左右节点下标的公式，我们可以按照公式算出左右节点的下标，并比较两者的大小，挑选更小的值与父节点进行对比。

最后，我们再来看一下优先级队列的应用场景。其实，JUC中的定时调度线程池ScheduledExecutorService的底层就使用了优先级队列。

定时任务调度线程池的基本实现原理是：

• 在将调度任务提交到线程池之前，首先计算出下一次需要执行的时间戳，通过时间戳来计算优先级，将其存入最小堆中，这样就确保了最先需要执行的调度任务位于最小堆的顶部(也就是根节点)。

• 然后开一个定时任务，拿队列中第一个元素和当前时间进行比较： - 如果下一次执行时间大于等于当前时间，则将队列中第一个元素(调度任务)从队列中移除，投入线程池中执行。

  • 如果下一次执行时间小于当前时间，则不处理，因为队列中最小的待执行任务都还没有到执行时间，其他任务一定也是这样。

可以看到，定时调度场景的关键是找到第一个需要触发的任务，类似SQL中的min语义，重在优先二字，而优先级队列的实现原理同样注重优先。理念上的契合让定时任务调度和优先级队列经常绑定在一起出现。

总结

好了，这节课就讲到这里。内容比较多，但是把脉络拎出来，其实我们主要讲了三种数据结构。

其中，树是数据结构中比较难但同时也非常常见的一种数据结构。我们从二叉排序树的优劣势出发，引出了红黑树，并用图解的方式详细介绍了红黑树的构建过程，介绍了红黑树的左旋、右旋、变色方法，还列举了红黑树的经典应用场景。

紧接着我们介绍了LinkedHashMap，它是链表与HashMap的结合体。LinkedHashMap既拥有HashMap快速的检索能力，还引入了节点顺序性，可以基于它实现LRU缓存淘汰算法。

最后，我们还通过图解认识了优先级队列，看到了用数组存储树的高阶用法，以及堆排序的工作机制和应用场景。

希望你能够借这个机会再巩固一下自己的基础知识，有所收获。同时，我也建议你在学完这些数据结构基本原理之后，有针对性地阅读一下源码，提炼出自己的学习方法。

课后题

最后我还是照例给你留两道课后题吧！

1、请你根据红黑树的特性，实现一棵红黑树（插入、删除、查找）。

2、红黑树和最小堆之间有什么区别，各自适用于什么场景？

欢迎你在留言区与我交流讨论，我们下节课见！

05 | 多线程：多线程编程有哪些常见的设计模式？

作者: 丁威

你好，我是丁威。

从这节课开始，我们开始学习Java多线程编程。

多线程是很多人在提升技术能力的过程中遇到的第一个坎，关于这部分的资料在网络上已经很多了，但是这些资料往往只重知识点的输出，很少和实际的生产实践相挂钩。但是我不想给你机械地重复“八股文”，接下来的两节课，我会结合这些年来在多线程编程领域的经验，从实际案例出发，带你掌握多线程编程的要领，深入多线程的底层运作场景，实现理解能力的跃升。

如何复用线程？

线程是受操作系统管理的最核心的资源，反复创建和销毁线程会给系统层面带来比较大的开销。所以，为了节约资源，我们需要复用线程，这也是我们在多线程编程中遇到的第一个问题。那怎么复用线程呢？

我们先来看一小段代码：

1

Thread t = new Thread(new UserTask());

请你思考一下，这段代码会创建一个操作系统线程吗？

答案是不会。这段代码只是创建了一个普通的Java对象，要想成为一个真实的线程，必须调用线程的start方法，让线程真正受操作系统调度。而线程的结束和run方法的执行情况有关，一旦线程的run方法结束运行，线程就会进入消亡阶段，相关资源也会被操作系统回收。

所以要想复用线程，一个非常可行的思路就是，不让run方法结束。

通常我们会想到下面这种办法：

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

class Task implements Runnable {
        @Override
        public void run() {
            while(true) {
                if( shouldRun() ) {// 符合业务规则就运行
                    doSomething();
                } else {
                    try {
                        //休眠1s,继续去判断是否可运行
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                
            }
        }
        private void doSomething() {
        }
        private boolean shouldRun() {
            //根据具体业务规则进行判断
            return false;
        }
    }

通过一个while(true)死循环确保run方法不会结束，然后不断地判断当前是否可以执行业务逻辑；如果不符合执行条件，就让线程休眠一段时间，然后再次进行判断。

这个方法确实可以复用线程，但存在明显的缺陷。因为一旦不满足运行条件，就会进行反复无意义的判断，造成CPU资源的浪费。另外，在线程处于休眠状态时，就算满足执行条件，也需要等休眠结束后才能触发检测，时效性会大打折扣。

那我们能不能一有任务就立马执行，没有任务就阻塞线程呢？毕竟，如果线程处于阻塞状态，就不会参与CPU调度，自然也就不会占用CPU时间了。

答案当然是可以的，业界有一种非常经典的线程复用模型：while循环+阻塞队列，下面是一段示范代码：

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

class Task implements Runnable {
        private LinkedBlockingQueue taskQueue = new LinkedBlockingQueue();
        private AtomicBoolean running = new AtomicBoolean(true);
​
        public void submitTask(Object task) throws InterruptedException {
            taskQueue.put(task);
        }
​
        @Override
        public void run() {
            while(running.get()) {
                try {
                    Object task = taskQueue.take(); // 如果没有任务，会使线程阻塞，一旦有任务，会被唤醒
                    doSomething(task);
                } catch (Throwable e) {
                    e.printStackTrace();
                }
            }
        }
​
        public void shutdown() {
            if(running.compareAndSet(true, false)) {
                System.out.println(Thread.currentThread() + " is stoped");
            }
        }
​
        private void doSomething(Object task) {
        }
    }

我们来解读一下。这里，我们用AtomicBoolean变量来标识线程是否在运行中，用while(running.get())替换while(true)，方便优雅地退出线程。

线程会从阻塞队列中获取待执行任务，如果当前没有可执行的任务，那么线程处于阻塞状态，不消耗CPU资源；一旦有任务进入到阻塞队列，线程会被唤醒执行任务，这就很好地保证了时效性。

那怎么停止一个线程呢？调用线程的shutdown方法一定能停止线程吗？

答案是不一定。 如果任务队列中没有任务，那么线程会一直处于阻塞状态，不能被停止。而且，Java中Thread对象的stop方法被声明为已过期，直接调用并不能停止线程。那怎么优雅地停止一个线程呢？

原来，Java中提供了中断机制，在Thread类中与中断相关的方法有三个。

• public void interrupt()：Thread实例方法，用于设置中断标记，但是不能立即中断线程。
• public boolean isInterrupted()：Thread实例方法，用于获取当前线程的中断标记。
• public static boolean interrupted()：Thread静态方法，用于获取当前线程的中断标记，并且会清除中断标记。

如果调用线程对象的interrupt()方法，会首先设置线程的中断位，这时又会出现两种情况：

• 如果线程阻塞在支持中断的方法上，会立即结束阻塞，并向外抛出InterruptedException(中断异常)；
• 如果线程没有阻塞在支持中断的方法上，则该方法不能立即停止线程。

不过要说明的是，JUC类库中的所有阻塞队列、锁、Object的wait等方法都支持中断。

通常，我们需要在代码中添加显示的中断检测代码，我还是用前面的例子给出示例代码，你可以看一下：

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

static class Task implements Runnable {
        private LinkedBlockingQueue taskQueue = new LinkedBlockingQueue();
        private AtomicBoolean running = new AtomicBoolean(true);
​
        public void submitTask(Object task) throws InterruptedException {
            taskQueue.put(task);
        }
​
        @Override
        public void run() {
            while(running.get()) {
                try {
                    Object task = taskQueue.take(); // 如果没有任务，会使线程阻塞，一旦有任务，会被唤醒
                    doSomething(task);
                    
                    if(Thread.currentThread().isInterrupted()) {
                        //线程被中断，跳出循环，线程停止
                        break;
                    }
                    
                    //这是一个耗时很长的方法
                    doSomething2(task);
                    
                } catch (Throwable e) {
                    e.printStackTrace();
                }
            }
        }
​
        public void shutdown() {
            if(running.compareAndSet(true, false)) {
                System.out.println(Thread.currentThread() + " is stoped");
            }
        }
​
        private void doSomething(Object task) {
        }
        
        private void doSomething2(Object task) {
        }
        
    }

我们继续说回线程的复用。JUC框架提供了线程池供我们使用。关于线程池相关的基础知识，你可以参考我之前的文章《如何评估一个线程池需要设置多少个线程》，这里我就不过多展开了。接下来，我就结合自己的工作经验分享一下怎么在实战中使用线程池。

我非常不建议你直接使用Executors相关的API来创建线程池，因为通过这种方式创建的线程池内部会默认创建一个无界的阻塞队列，一旦使用不当就会造成内存泄露。

我更推荐你使用new的方式创建线程，然后给线程指定一个可阅读的名称：

1
2
3
4
5
6
7
8
9
10

ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 5, 0, TimeUnit.MILLISECONDS, 
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;new LinkedBlockingQueue<>(), new ThreadFactory() {
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;private AtomicInteger threadNum = new AtomicInteger(0);
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;@Override
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;public Thread newThread(Runnable r) {
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;Thread t = new Thread(r);
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;t.setName("pull-service-" + threadNum.incrementAndGet());
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;return t;
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;  }
 &nbsp; &nbsp; &nbsp;  });

这样，当系统发生故障时，如果我们想要分析线程栈信息，就能很快定位各个线程的职责。例如，RocketMQ的消费线程我就会以“ConsumeMessageThread_”开头。

使用线程池另一个值得关注的问题是怎么选择阻塞队列，是使用无界队列还是有界队列。

通常，我们可以遵循这样的原则：对于Request-Response等需要用户交互的场景，建议使用有界队列，避免内存溢出；对于框架内部线程之间的交互，可以根据实际情况加以选择。

我们通过几个例子来看一下具体的场景。

项目开发中通常会遇到文件下载、DevOps的系统发布等比较耗时的请求，这类场景就非常适合使用线程池。基本的工作方式如图：

在与用户交互的场景中，如果几十万个文件下载请求同时提交到线程池，当线程池中的所有线程都在处理任务时，无法及时处理的请求就会存储到线程池中的阻塞队列中。这就很容易使内存耗尽，从而触发Full-GC，导致系统无法正常运作。

因此，这类场景我建议使用有界队列，直接拒绝暂时处理不了的请求，并给用户返回一条消息“请求排队中，请稍后再试”，这就保证了系统的可用性。

在一个线程或多个线程向一个阻塞队列中添加数据时，通常也会使用有界队列。记得我在开发数据同步产品时，为了实现源端与目标端线程，就采用了阻塞队列，下面是一张示意图：

为了实现MySQL增量同步，Canal线程源源不断地将MySQL数据写入到阻塞队列，然后目标端线程从队列中读取数据并写入到MQ。如果写入端的写入速度变慢，阻塞队列中的数据就变得越来越多，一旦不加以控制就可能导致内存溢出。所以，为了避免由于写入端性能瓶颈造成的整个系统的不可用，这时候需要引入有界阻塞队列。这样，队列满了之后，我们就能让源端线程处于阻塞状态，从而对源端进行限流。

但在选择阻塞队列时还可能有另外一种情况，那就是一个线程对应多个阻塞队列，这时我们一般会采用无界阻塞队列+size的机制，实现细粒度限流。当时，我设计的RocketMQ消费模型是下面这样：

一个拉取线程轮流从Broker端队列(q0、q1)中拉取消息，然后根据队列分别放到不同的阻塞队列中，每一个阻塞队列会单独分配单个或多个线程去处理。

这个时候，采用有界队列可能出现问题。如果我们采用有界队列，一旦其中一个阻塞队列对应的下游消费者处理性能降低，阻塞队列中没有剩余空间存储消息，就会阻塞消息发送线程，最终造成另外一个任务也无法拉取新的消息。显然，这会让整体并发度降低，影响性能。

那如果采用无界队列呢？单纯使用无界队列容易导致内存泄露，触发更严重的后果，好像也不是一个好的选择。

其实我们可以在无界队列的基础上额外引入一个参数，用它来控制阻塞队列中允许存放的消息条数。当阻塞队列中的数据大于允许存放的阔值时，新的消息还可以继续写入队列，不会阻塞消息发送线程。但我们需要给消息拉取线程一个反馈，暂时停止从对应队列中拉取消息，从而实现限流。

阻塞队列是多线程协作的核心纽带，除了清楚它的使用方法，我们还应该理解它的使用原理，也就是 “锁 + 条件等待与唤醒”。我们来看一下LinkedBlockingQueue 的put的实现代码：

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

public void put(E e) throws InterruptedException {
        if (e == null) throw new NullPointerException();
​
        int c = -1;
 &nbsp; &nbsp; &nbsp; &nbsp;Node<E> node = new Node<E>(e);
 &nbsp; &nbsp; &nbsp; &nbsp;final ReentrantLock putLock = this.putLock;
 &nbsp; &nbsp; &nbsp; &nbsp;final AtomicInteger count = this.count;
 &nbsp; &nbsp; &nbsp; &nbsp;putLock.lockInterruptibly(); // @1
 &nbsp; &nbsp; &nbsp; &nbsp;try {
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;while (count.get() == capacity) { // @2
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;// &nbsp;  private final ReentrantLock putLock = new ReentrantLock();
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;// &nbsp;  private final Condition notFull = putLock.newCondition();
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;notFull.await(); &nbsp;// @3
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;  }
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;enqueue(node); &nbsp;// @4
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;c = count.getAndIncrement();
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;if (c + 1 < capacity)
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;notFull.signal();
 &nbsp; &nbsp; &nbsp;  } finally {
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;putLock.unlock();
 &nbsp; &nbsp; &nbsp;  }
 &nbsp; &nbsp; &nbsp; &nbsp;if (c == 0)
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;signalNotEmpty();
 &nbsp;  }

这里，我重点解读一下关键代码。

第8行：我们要申请锁，获取队列内部数据存储结构(LinkedBlockingQueue底层结构为链表)的修改控制权，也就是让一个阻塞队列同一时刻只能操作一个线程。

第10行：判断队列中元素的数量是否等于其最大容量，如果是，则线程进入到条件等待队列(第13行)，调用put的线程会释放锁进入到阻塞队列。当队列中存在空闲空间时，该线程会得到通知，从而结束阻塞状态进入到可调度状态。

队列中有可用空间之后，线程被唤醒，但是不能立即执行代码（第15行），它需要重新和其他线程竞争锁，获得锁后将数据存储到底层数据结构中。关于锁的底层原理，我们会在下节课详细介绍。

这里也请你思考一下：为什么上面的代码我们要采用while(count.get() == capacity)而不使用if(count.get() == capacity)呢？

多线程编程常用的设计模式

如果你刚开始学习多线程编程，可能会觉得这个问题很难。不过不用担心，业界大佬早就总结出了很多和多线程编程相关的设计模式。接下来，我就带你一起看看其中应用最广的几个。

Future模式

多线程领域一个非常经典的设计模式是Future模式。它指的是主线程向另外一个线程提交任务时，无须等待任务执行完毕，而是立即返回一个凭证，也就是Future。这时主线程还可以做其他的事情，不会阻塞。等到需要异步执行结果时，主线程调用Future的get方法，如果异步任务已经执行完毕，则立即获取结果；如果任务还没执行完，则主线程阻塞，等待执行结果。

Future模式的核心要领是将多个请求进行异步化处理，并且可以得到返回结果。我们来看一个示例：

当一个请求在处理时，需要发起多个远程调用，并且返回多个请求，再根据结果进行下一步处理。它的伪代码如下：

1
2
3

Object result1 = sendRpcToUserCenter(); // @1
Object result2 = sendRpcToOrgCenter(0); // @2
Object result = evalBusiness(result1,result2);

说明一下，在不使用Future模式的情况下，两个远程RPC调用是串行执行的。例如，第一个请求需要1s才能返回，第二个请求需要1.5s才能返回，这两个过程就需要2.5s。为了提高性能，我们可以将这两个请求进行异步处理，然后分别得到处理结果。这就到了Future模式发挥作用的时候了。

业务开发领域通常会采用线程池的方式来实现Future模式，你可以看下具体的实现代码：

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

package net.codingw.jk02;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
public class FutureTask {
 &nbsp; &nbsp;static class Rpc2UserCenterTask implements Callable<Object> {
 &nbsp; &nbsp; &nbsp; &nbsp;@Override
 &nbsp; &nbsp; &nbsp; &nbsp;public Object call() throws Exception {
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;return sendRpcToUserCenter();
 &nbsp; &nbsp; &nbsp;  }
 &nbsp; &nbsp; &nbsp; &nbsp;private Object sendRpcToUserCenter() {
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;// 具体业务逻辑省略
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;return new Object();
 &nbsp; &nbsp; &nbsp;  }
 &nbsp;  }
​
 &nbsp; &nbsp;static class Rpc2OrgCenterTask implements Callable<Object> {
 &nbsp; &nbsp; &nbsp; &nbsp;@Override
 &nbsp; &nbsp; &nbsp; &nbsp;public Object call() throws Exception {
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;return sendRpcToOrgCenter();
 &nbsp; &nbsp; &nbsp;  }
 &nbsp; &nbsp; &nbsp; &nbsp;private Object sendRpcToOrgCenter() {
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;// 具体业务逻辑省略
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;return new Object();
 &nbsp; &nbsp; &nbsp;  }
 &nbsp;  }
 &nbsp; &nbsp;
 &nbsp; &nbsp;public static void main(String[] args) throws Exception {
 &nbsp; &nbsp; &nbsp; &nbsp;// 生产环境建议使用 new ThreadPoolExecutor方式创建线程池
 &nbsp; &nbsp; &nbsp; &nbsp;ExecutorService executorService = Executors.newFixedThreadPool(5);
 &nbsp; &nbsp; &nbsp; &nbsp;// 发起
 &nbsp; &nbsp; &nbsp; &nbsp;Future<Object> userRpcResultFuture = executorService.submit(new Rpc2UserCenterTask()); //异步执行
 &nbsp; &nbsp; &nbsp; &nbsp;Future<Object> orgRpcResultFuture = executorService.submit(new Rpc2OrgCenterTask()); &nbsp; // 异步执行
 &nbsp; &nbsp; &nbsp; &nbsp;Object userRpcResult = userRpcResultFuture.get(); // 如果任务未执行完成，则该方法会被阻塞，直到处理完成
 &nbsp; &nbsp; &nbsp; &nbsp;Object orgRpcResult = orgRpcResultFuture.get(); &nbsp; // 如果任务未执行完成，则该方法会被阻塞，直到处理完成
 &nbsp; &nbsp; &nbsp; &nbsp;doTask(userRpcResult, orgRpcResult);
 &nbsp;  }
​
 &nbsp; &nbsp;private static void doTask(Object userRpcResult, Object orgRpcResult) {
 &nbsp; &nbsp; &nbsp; &nbsp;// doSomeThing
 &nbsp;  }
}

我们还是解读一下这段代码的要点。

• 首先，我们需要创建一个线程池。
• 接着，要将需要执行的具体任务进行封装，并实现java.util.concurrent.Callable接口(如上述代码中的Rpc2UserCenterTask)，并重写其Call方法。
• 然后将一个具体的任务提交到线程池中去执行，返回一个Future对象。
• 在想要获取异步执行结果时，可以调用Future的get方法。如果任务已经执行成功，则直接返回；否则就会进入阻塞状态，直到任务执行完成后被唤醒。

因为线程池是一个较重的资源，而中间件领域的开发追求极致的性能，所以在中间件开发领域通常不会直接使用线程池来实现Future模式。

RocketMQ会使用CountDownLatch来实现Future模式，它的设计非常精妙，我们先一起来看一下它的序列图：

可以看到，SendMessageThread会首先创建一个GroupCommitRequest请求对象，并提交到刷盘线程，然后发送线程阻塞，等待刷盘动作完成。刷盘线程在执行具体刷盘逻辑后，会调用request的通知方法，唤醒发送线程。

乍一看，主线程提交刷盘任务之后并没有返回一个Future，那为什么说这是Future模式呢？这就是RocketMQ的巧妙之处了。它其实是把请求对象当作Future来使用了。我们来看一下GroupCommitRequest的实现代码：

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

public static class GroupCommitRequest {
        private final long nextOffset;
        private final CountDownLatch countDownLatch = new CountDownLatch(1); 
        private volatile boolean flushOK = false;
​
        public GroupCommitRequest(long nextOffset) {
            this.nextOffset = nextOffset;
        }
​
        public long getNextOffset() {
            return nextOffset;
        }
​
        public void wakeupCustomer(final boolean flushOK) {
            this.flushOK = flushOK;
            this.countDownLatch.countDown();
        }
​
        public boolean waitForFlush(long timeout) { 
            try {
                this.countDownLatch.await(timeout, TimeUnit.MILLISECONDS);
                return this.flushOK;
            } catch (InterruptedException e) {
                e.printStackTrace();
                return false;
            }
        }
    }

在这里，GroupCommitRequest 中的 waitForFlush 方法相当于 Future 的 get 方法。具体实现是，调用CountDownLatch的await方法使自己处于阻塞状态，然后当具体的刷盘线程完成刷盘之后，通过调用wakeupCustomer方法，实际上调用了CountDownLatch的countDown方法，实现唤醒主线程的目的。

基于CountDownLatch实现的Future模式非常巧妙，更加得轻量级，性能也会更好。不过要说明的是，在业务开发领域，直接使用线程池将获得更高的开发效率和更低的使用成本。

生产者-消费者模式

Future模式就说到这里，我们再来看看多线程编程领域中最常见的设计模式：生产者-消费者模式。

程序设计中一个非常重要的思想是解耦合，在Java设计领域也有一条重要的设计原则就是要职责单一。基于这些原则，通常一个功能需要多个角色相互协作才能正常完成。

生产者-消费者模式正是这种思想的体现，它的理论也很简单，我们这里不会深入介绍。但我想用RocketMQ举一个实操的例子。

在RocketMQ消费线程模型中，应用程序在启动消费者时，首先需要根据负载算法进行队列负载，然后消息拉取线程会根据负载线程计算的结果有针对性地拉取消息。交互流程如下图所示：

Rebalace线程作为生产者，会根据业务逻辑生成消息拉取任务，然后Pull线程作为消费者会从队列中获取任务，执行对应的逻辑；如果当前没有可执行的逻辑，Pull线程则会阻塞等待，当生产者将新的任务存入到阻塞队列中后，Pull线程会再次被唤醒。

系统的运行过程中会存在很多意料之外的突发事件，在高并发领域更是这样。所以在进行系统架构设计时，我们必须具备底线思维，对系统进行必要的兜底，防止最坏的情况发生，这里最常见的做法就是采用限流机制。

所以在这节课的最后，我们一起来看看并发编程领域如何实现限流。

线程池自带一定的限流效果，因为工作线程数量是一定的，线程池允许的最大并发也是确定的。一旦达到最大并发，新的请求就会进入到阻塞队列，或者干脆被拒绝。不过这节课我想给你介绍另一种限流的方法：使用信号量。

我们先来看一个具体的示例：

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

public static void main(String[] args) {
        Semaphore semaphore = new Semaphore(10);
 &nbsp; &nbsp; &nbsp; &nbsp;for(int i = 0; i < 100; i++) {
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;Thread t = new Thread(new Runnable() {
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;@Override
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;public void run() {
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;doSomething(semaphore);
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;  }
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;  });
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;t.start();
 &nbsp; &nbsp; &nbsp;  }
 &nbsp;  }
 &nbsp; &nbsp;private static &nbsp;void doSomething(Semaphore semaphore) {
 &nbsp; &nbsp; &nbsp; &nbsp;boolean acquired = false;
 &nbsp; &nbsp; &nbsp; &nbsp;try {
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;acquired = semaphore.tryAcquire(3000, TimeUnit.MILLISECONDS);
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;if(acquired) {
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;System.out.println("执行业务逻辑");
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;  } else {
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;System.out.println("信号量未获取执行的逻辑");
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;  }
 &nbsp; &nbsp; &nbsp;  } catch (Throwable e) {
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;e.printStackTrace();
 &nbsp; &nbsp; &nbsp;  } finally {
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;if(acquired) {
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;semaphore.release();
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;  }
 &nbsp; &nbsp; &nbsp;  }
 &nbsp;  }

这段代码非常简单，其实就是通过信号量来控制doSomething方法的并发度，使用了信号量的两个主要的方法。

• tryAcquire：这种方法是尝试获取一个信号，如果当前没有剩余的许可，过了指定等待时间之后会返回false，表示未获取许可；
• release：归还许可，该方法必须在tryAcquire方法返回true时调用，不然会发生“许可超发”。

但是如果场景再复杂一点，比如doSomething是一个异步方法，前面这段代码的效果就会大打折扣了。如果doSomething的分支非常多，或者遇到异步调用等复杂情况下，归还许可将变得非常复杂。

因为在使用信号量时，如果多次调用release，应用程序实际的并发数量会超过设置的许可值。所以避免重复调用release方法显得非常关键。RocketMQ给出的解决方案如下：

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

public class SemaphoreReleaseOnlyOnce {
  private final AtomicBoolean released = new AtomicBoolean(false);
  private final Semaphore semaphore;
​
  public SemaphoreReleaseOnlyOnce(Semaphore semaphore) {
    this.semaphore = semaphore;
  }
  public void release() {
    if (this.semaphore != null) {
      if (this.released.compareAndSet(false, true)) {
        this.semaphore.release();
      }
    }
  }
  public Semaphore getSemaphore() {
    return semaphore;
  }
}

这套方案的核心思想是对Semaphore进行一次包装，然后将包装对象（SemaphoreReleaseOnlyOnce）传到业务方法中。就像上面这段代码，其中的doSomething方法无论调用release多少次都可以保证底层的Semaphore只会被释放一次。

SemaphoreReleaseOnlyOnce的release方法引入了CAS机制，如果release方法被调用，就使用CAS将released设置为true。下次其他线程再试图归还许可时，由于状态为true，所以不会再次调用Semaphore的release方法，这样就可以有效控制并发数量了。

总结

好了，这节课就讲到这里。

这节课一开始，我们就讲了一个大家在多线程编程中常会遇到的问题：如何复用线程？我们重点介绍了线程池这种复用方法。它的内部的原理是采用 while + 阻塞队列的机制，确保线程的run方法不会结束。在有任务执行时运行任务，无任务运行时则通过阻塞队列阻塞线程。我们还顺便讲了讲怎么通过中断技术优雅地停止线程。

使用线程池时，还有一个常见的问题就是怎么选择阻塞队列，我总结了下面三个小窍门：

• Request-Response等需要用户交互的场景，建议使用有界队列，避免内存溢出；
• 如果一个线程向多个队列写入消息，建议使用“无界队列+size”机制，不阻塞队列；
• 如果一个线程向一个队列写入消息，建议使用有界队列，避免内存溢出。

这节课的后半部分，我们详细介绍了多线程领域Future模式、生产者-消费者模式的工作原理和使用场景。我还提到了高并发架构设计中的底线思维：限流机制。基于信号量来实现限流，在多线程环境中避免信号量的超发可以防止你踩到很多坑。

课后题

在课程的最后，我还是照例给你留两道思考题。

1. 你是怎么理解Future模式的？又会怎么实现它呢？
2. 场景题：有一家主要生产面包的工厂，但是工厂的仓库容量非常有限。一旦仓库存满面包，就没法生产新的面包了。顾客来购买面包后，仓库容量会得到释放。请你用Java多线程相关的技术实现这个场景。

完成这个场景可以让我们迅速理解多线程编程的要领，所以请你一定要重视第二题。如果你想要分享你的修改或者想听听我的意见，可以提交一个 GitHub的push请求或issues，并把对应地址贴到留言里。我们下节课见！

06 | 锁：如何理解锁的同步阻塞队列与条件队列？

作者: 丁威

你好，我是丁威。

这节课，我们重点介绍并发编程中的基石：锁。

锁的基本存储结构

我们先通过一个简单的场景来感受一下锁的使用场景。一家三口在一起生活，家里只有一个卫生间，大家早上起床之后都要去厕所。这时候，一个人在卫生间，其他人就必须排队等待。

这个场景用IT术语可以表述为下面两点。

• 洗手间作为一个资源在同一时间只能被一个人使用，它具备排他性。
• 一个人用完洗手间（资源）之后会归还锁，然后排队者重新开始竞争洗手间的使用权。

我们可以对这个场景进行建模。

• 资源：更准确地说是公共资源或共享资源需要被不同的操作者使用，但它不能同时被使用。
• 资源使用者：共享资源的使用者。
• 锁：用来保护资源的访问权。锁对象的归属权为共享资源，但当资源使用者向资源申请操作时，可以将锁授予资源使用者。这时候，资源使用者叫做锁的占有者，在此期间它有权操作资源。操作者不再需要操作资源之后，主动将锁归还。
• 排队队列：我们可以更专业地称之为阻塞队列，它可以存储需要访问资源但还没获取锁的资源使用者，其归属权通常为锁对象。

这里我之所以强调归属权，主要是因为它可以帮助我们理解锁的基本结构和资源的关系。

那锁的结构是什么呢？我们通过上节课的课后题来理解这个问题。

上节课的第二道课后题是问你怎么用多线程实现面包厂的生产和销售。我在这里也给你写了一段示例代码：

面包仓库的职能是为面包厂存储面包，它需要提供两个基本的方法：存储面包和获取面包。面包仓库内部使用ArrayList来存储面包，但是因为ArrayList是一个线程不安全的存储容器，它不允许多个使用者同时存储数据，所以我们需要对资源进行保护。体现在代码上，我们可以通过 synchronized（资源对象）来创建一把锁，保护多线程对资源对象的串行访问。

我们结合put方法的流程来看一下锁的基本存储结构。

假设 t1、t2两个生产者（线程）同时调用Bakery的put方法。那么synchronized(breads)在编译的时候，就会在资源breads对象上创建锁相关的结构，即锁对象。

t1，t2在执行synchronized(breads)时，只有一个线程可以获取锁，另外一个线程需要等待，所以这里需要引入一个存储结构（通常为队列）来存储这些排队的线程，我们通常会使用阻塞队列。

首先获取到锁的线程t1在向仓库中存放面包之前需要先进行判断，如果存储空间足够，执行上图中的代码step2。但是如果仓库没有足够的空间存储面包，就要执行代码step3，调用锁对象的wait方法，让获得锁的线程t1阻塞，并且释放锁。

但是，被阻塞的t1和t2还是有所不同。因为t1被阻塞的原因是条件不满足，当面包仓库有额外的存储空间时，t1就会被唤醒。所以我们还要引入一个条件队列，用来存放因条件不满足而被阻塞的线程。

t1线程如果因为条件不满足而存储在条件等待队列，当存在剩余空间后，就能被其他线程唤醒继续执行后续的代码了。在这里是将面包存储到ArrayList，那此时面包工厂中存储了面包，需要通知那些因为仓库中没有面包而阻塞的线程，调用锁的notifyAll方法唤醒在等待的线程。

线程t1因为存储空间不足在step3被阻塞，进入到条件等待队列。等到面包被卖出，仓库有足够的容量之后，t1线程将被唤醒。

这里我想给你提个问题，t1线程可以立马继续执行step3之后的代码step4吗？

答案是不能，它需要先去尝试竞争锁，成功获得锁之后才能开始执行step4，否则就会进入到阻塞队列。

从上面这个过程中，我们可以归纳出锁的基本存储结构，它包括锁的持有者线程、锁的重入次数、阻塞队列和条件等待队列四个部分。

锁的底层实现机制-AQS实现原理剖析

在Java中使用锁通常有两种编程方式。一种是使用JVM虚拟机（Java规范）层面提供的synchronized关键字；另一种是使用JUC类库，也就是大名鼎鼎的AbstractQueuedSynchronizer，简称 AQS。

其中，synchronized是在JVM虚拟机层面实现的，涉及很多底层知识，直接研读源码难度太大。相比较而言，JUC并发编程遵从JSR-166规范，提供了锁的另外一种实现方式，也就是大家所熟知的AQS类库，更加常用和易学。

接下来，我会基于JUC框架，带你从代码层面近距离观摩锁的实现原理，掌握锁的本质。

在JUC框架中，ReentrantLock对标synchronized，它实现了可重入互斥锁的全部语义。语义主要包括两个方面：一个是lock（申请锁）和unlock（释放锁）；另一个是条件等待，对标Object的wait/notify。

我们先来看下ReentrantLock和AQS的类图：

简单介绍一下类图中各个类的含义。

1. AbstractQueuedSynchronizer

它是AQS体系的核心基类，使用的是类模版设计模式。这个类实现了锁的基本存储结构，定义了锁的基本行为。AQS的内部数据结构为链表，持有链表的头尾节点，每一个节点用Node表示，可以实现阻塞队列和条件等待队列。其中，Node prev、next用于构建阻塞队列，而Node nextWatier用于构建条件等待队列。

AQS方法的修饰符也很有规律，其中，使用protected修饰的方法为抽象方法，通常需要子类去实现，从而实现不同特性的锁（例如互斥、共享锁、读写锁等）；而用public修饰的方法基本可以认为是模板方法，不建议子类直接覆盖。

AQS还额外提供了很多有用的方法，我给你列了个表格，方便你在有需要的时候随时查看。

2. AbstractOwnableSynchronizer

它是AQS核心基类的父类，用于记录锁当前的持有者线程。

3. ReentrantLock

可重入互斥锁的具体实现。由于Java不支持多继承，所以由ReentrantLock继承抽象类Lock，用内部类的方式继承AQS。所以说，ReentrantLock在具体实现锁时基本都是委托内部类Sync，而Sync又继承自AQS。Sync内部有两个子类，分别是FairSync（公平锁）与NonfairSync（非公平锁）。

锁的申请

接下来我们结合ReentrantLock的部分关键源码来看看怎么实现锁的申请与释放。先看锁的申请。

ReentrantLock支持带超时时间的锁申请，具体实现方法是tryLock，时序图如下：

AQS的tryAcquireNanos代码如下图所示，该方法是在AQS中定义的。

解读一下核心要点。

如果线程的中断位标记为true，表示应用方主动放弃锁的申请，可以直接抛出中断异常，结束锁的申请。

否则，调用Sync的tryAcquire尝试获取锁。如果返回true，表示成功获取锁，可以直接返回；不然就调用doAcquireNanos，进入锁等待队列。

Sync的tryAcquire方法，代码如下：

尝试实现锁有几个要点。

首先我们要确保获取当前申请锁的线程。

我们还要获取锁的当前状态，也就是state值（state字段的含义是当前锁的重入次数，如果state为0，表示当前锁并没有被占用）。这又包括两种情况。

情况一：如果state的值为0，表示当前锁并没有被占用。根据申请锁的公平与否，会有不同的处理逻辑。

• 如果是公平锁，那么我们需要判断阻塞队列中有没有其他线程在排队。如果有，公平锁此时无法竞争锁，返回false，尝试获取锁失败。这个线程最终会调用doAcquireNanos，进入到同步阻塞队列。
• 但是如果是非公平锁，则会首先和阻塞队列中的线程竞争，如果竞争成功，可以直接获取锁，如果竞争失败，则同样进入到阻塞队列。
  

竞争锁的代码使用的是CAS机制，尝试更新state的值为acquires，如果更新成功，则占有锁。成功占有锁之后，需要设置锁的拥有者为当前线程。

情况二：如果state的值不为0，表示锁已经被占用。我们需要判断当前线程是不是锁的持有者。如果是，则只需要更新state的值（ReentrantLock支持可重入）；否则就进入阻塞队列，排队获取锁。

为什么在竞争锁时需要使用CAS呢？什么是CAS呢？

我们知道，申请锁时要先查询state的值，然后更新state的值。但这两步在多线程环境中并不是一个安全的操作。如下图所示：

这很容易导致t1，t2都获取到了锁。根本原因是这个步骤包括两个CPU指令，无法做到原子更新。

为了解决这个问题，操作系统提供了一个新的CPU指令（CAS），用它来实现“比较-和-更新”。具体的原理是在更新一个值之前，首先比较这个值是否发生了变化，如果确实发生了变化，那么就会更新失败，否则更新成功。

如果没有成功获取锁，当前申请锁的线程还会继续调用AQS的doAcquireNanos：

这是AQS机制中非常重要的一个方法，它的实现比较复杂。我们先来看一张流程图：

我们可以把这个流程归结为五步。

第一步：判断获取锁是不是已经超时，如果是，返回false（ReentrantLock支持锁获取超时）。

第二步：调用addWaiter方法把当前节点加入到阻塞队列中。

第三步：获取节点的前驱节点。

第四步：如果节点的前驱节点为头节点，再次调用tryAcquire方法尝试获取锁。如果成功获取锁，则将当前节点设置为Head，表示当前它是锁的持有者。

第五步：如果当前节点不是头节点，或者没有成功获取锁，调用shouldParkAfterFailedAcquire判断当前线程是否需要阻塞，如果需要阻塞，则调用LockSupport.parkNanos阻塞线程。

接下来，我们对上面流程中的关键代码进行详细的解读。

先看第二步里addWaiter的具体实现：

因为AQS内部不管是阻塞队列还是条件等待队列都是基于链表实现的，所以入队列的实现比较容易理解，这里主要关注三点。

1. 需要创建一个Node节点，将线程对象存储在Node节点中，方便后续对线程进行阻塞或唤醒。
2. 链表在多线程环境中操作并不安全，所以在更新链表相关指针时要引入CAS机制。首先将 if和CAS组合进行一次测试，如果更新成功，直接结束操作；不然就要使用 for和CAS的组合进行多次重试，一直到更新成功为止。这背后的原理是，多线程在更新Head或者Tail时，只有一个能更新成功，如果更新失败，则重新获取Head或者Tail再进行更新，直到节点安全地加入链表为止。
3. 在入队的过程中，如果队列为空时，会创建一个空的Node节点，但是不持有任何线程信息。

等到节点成功加入到阻塞队列后，需要判断节点的前驱节点是否为头节点，如果是，表示成功获取锁。成功获得锁的线程对应的节点将成为头节点，设置头节点的代码如下：

头节点持有的线程对象为什么为空呢？

这是因为锁的持有者被记录在了AbstractOwnableSynchronizer的Thread exclusiveOwnerThread属性中。这样做的好处是，我们可以认为头节点是锁的持有者，但头节点却并不维护线程对象。在实现非公平锁时，如果锁被新线程抢占，不需要更新头节点。

相反，如果节点的前驱节点不是头节点，则需要判断申请锁的线程是否需要阻塞。我们可以通过shouldParkAfterFailedAcquire方法来实现它：

解读一下，如果前驱节点的状态是Node.SIGNAL，则当前线程直接进入到阻塞队列，排队获取锁。

这里再对Node.SIGNAL补充说明一下：Node.SIGNAL的含义是节点需要一个信号来唤醒自己，如果前驱节点的状态为Node.SIGNAL，说明前驱节点在等待被唤醒，那作为前驱节点的后继节点，自然而然也需要等待被唤醒。

如果前驱节点的状态大于0，需要删除当前节点之前连续的节点。因为当前节点的状态只有Node.CANCELLED大于0，所以如果前驱节点的状态大于0说明是已取消的节点，需要被删除。示例图如下：

这里以当前节点为基准（状态为-1）向前删除。注意，只删除连续的1，也就是说遇到非取消节点立即停止删除。基于分段思想，我们不会删除前面所有的已取消节点，因为删除节点的方向是从后向前的，而且shouldParkAfterFailedAcquire这个方法会在多个线程获取锁之后被多个线程调用，但后续的节点在执行删除时，遇到当前线程，会被切割成段，段与段之间并不会有多线程执行，从而可以安全地操作各自的段。

如果前驱节点的状态为0或Propagate，需要尝试把前驱节点的状态变更为Node.SIGNAL。也就是说，不阻塞线程，而是再次试图获取锁相关的逻辑。

如果需要阻塞线程，先判断本次获取锁的剩余时间是否大于等于spinForTimeoutThreshold，如果是，则通过自旋方式进行循环，否则将使线程阻塞。其中spinForTimeoutThreshold默认为1s，这样做的目的主要是如果本次锁申请距超时还剩不到1s，就没有必要再阻塞线程了，避免线程切换带来的额外开销。

如果需要阻塞线程，我们可以调用LockSupport.parkNanos方法使线程阻塞，这个方法同样支持设置超时时间。

锁的释放

申请完锁之后，我们还会面临锁的释放。我们可以通过ReentrantLock的unlock方法释放锁，并最终调用AQS的模版方法：release方法，代码如图所示：

在详细地介绍具体的方法之前，我们先来看一张整体的时序图，理解一下释放锁的实现机制。

锁的释放主要包含如下几个步骤：

第一步：释放锁，必须先判断当前线程是否是锁的持有者，如果不是，抛出IllegalMonitorStateException异常。

第二步：判断锁的剩余占有次数，如果为0，表示锁已释放，需要唤醒阻塞队列中的其他排队线程。

我们看一下释放锁的关键代码。具体定义在ReentrantLock$Sync的tryRelease中：

这段代码有两个要点。

• 如果当前锁的占有者不是申请释放锁的线程，那就不能释放锁，只有持有者线程才能释放锁。这个时候需要抛出监视器错误。
• 如果一个锁被同一个线程重入n次，那对应也要释放n次。当持有次数为0时，表示可以释放锁。

尝试释放锁后，返回“成功”，接下来要做的是唤醒阻塞队列中的下一个线程。当然，如果你使用的是非公平锁，新来的线程在这个时候是可以直接获取锁的，这样唤醒的线程如果没能获取锁，就又会进入到阻塞队列中。

从阻塞队列中查找下一个待唤醒的线程需要使用AQS的unparkSuccessor方法，代码如下图所示：

这个过程主要包括四个要点，分别对应上图的step1、step2、step3和step4。

step1：因为这个方法的参数是头节点，头节点是当前锁的持有者，所以在释放锁时，我们要找头节点的下一个未取消的节点。

step2：确认头节点的状态。如果头节点的状态不为0，则更新为0。

step3：从链表的尾部开始寻找，找到头节点后面的第一个非取消节点。这里说明一下，因为我们在维护节点的前驱节点时使用了CAS，通过前驱节点遍历是可靠的，不会遗漏节点。

step4：找到对应的节点，调用LockSupport.unpark方法唤醒线程。线程被唤醒后会继续去竞争锁。这里唤醒的是申请锁时用LockSupport.park阻塞的线程，因为这样可以让锁的申请和释放形成闭环通道。

锁的条件等待队列

理解了锁的申请和释放，接下来我们再来看看ReentrantLock是怎么实现Object.wait和Object.notify语义的，这是线程之间协作的基石。

线程调用锁对象的wait方法时会进入到条件等待队列，而线程调用锁对象的notify方法，会唤醒条件队列中的一个线程，具有下面三个特征。

1. Object 的 wait与notify必须在临界区中调用。
2. Object 的wait和notify的使用场景为条件等待。例如，一个线程获取锁后，需要等待某一个条件满足后才能继续执行。这时，为了节省CPU资源，线程可以调用锁的wait方法使自己阻塞，等待条件满足后被别的线程唤醒。
3. Object的wait方法会释放当前锁。

在AQS中，实现Object的notify和wait功能的主要类为Condition，类图如下：

Condition的接口对标Object的wait与notify方法，底层的存储结构为一个链表（条件阻塞队列）。链表中的节点为Node，条件阻塞队列为单链表，链表通过Node nextWaiter指针维护链表。

因为前面在介绍lock语义的时候我们用的是带超时时间的方法，所以为了覆盖更多的AQS方法，这里我们就变一变，用不带超时时间的方法来解读await语言。不过这两者在本质上并没有差别。

为了帮助你更快掌控await的整体实现思路，可以先看一下时序图：

Condition的wait()方法对标Object.wait()，我们来看一下它的具体实现逻辑：

我们结合Object.wait的语义来体会一下await方法中最关键的六个步骤。

step1：如果当前线程被中断，要直接抛出中断异常。

step2：将节点加入条件等待队列中。

step3：释放锁，并保存释放之前锁的状态，等到条件满足线程被唤醒后，需要重新申请指定数量的锁。

step4：如果节点存在于条件队列而不在阻塞队列中，说明未收到signal信号，线程会被阻塞；如果线程被中断，就结束条件队列的等待。

step5：再次尝试申请锁，并检查唤醒的原因，看看是因为收到signal信号而被唤醒，还是因为收到了中断信号。

step6：如果先收到signal信号，再收到中断信号，那就要重新设置线程中断位，等待下一次中断检查点；如果是先收到中断信号，后收到signal信号，就直接抛出中断异常；如果正常收到signal信号，await方法结束阻塞，则继续执行后续逻辑。

其中，第二步中的加入条件队列，具体的代码实现是将节点接入到链表的尾部，如果有取消的节点就把它删除。这里线程是安全的，因为执行await方法的前提条件是要获取锁。

第四步是用await方法阻塞和唤醒线程的关键。核心代码如下图所示：

我们来看一下怎么判断线程是否在同步队列中（用isOnSyncQueue方法实现）。

• 如果节点的状态为Node.CONDITION，或者node.prev为空，表示线程在等待条件被触发。为什么节点的前驱节点不为空就可以认为线程在同步阻塞队列中呢？这是因为进入同步队列时是用CAS机制来更新前驱节点的。
• 如果Node的next指针不为空，表示线程在同步阻塞队列中，返回true。
• 如果不满足上述条件，则从尾部节点再查找一遍，如果能找到，返回true，否则返回false。

因为节点如果在条件等待队列中，说明条件不满足，线程需要阻塞并等待条件触发。线程可以通过下面几种方式被唤醒：

• 由于正常收到signal信号被唤醒；
• 先收到signal信号，然后收到中断信号；
• 先收到中断信号，再收到signal信号。

那怎么判断唤醒方式呢？

我们可以通过checkInterruptWhileWaiting来实现它。也就是检测线程的中断标志位，如果线程并没有设置中断位，则返回0。如果检测到了中断位，则用transferAfterCancelledWait方法来判断中断信号和signal的先后顺序。

transferAfterCancelledWait的核心实现逻辑是，如果成功将节点的状态从Node.CONDITION更新为0，就表示先收到了中断标记，否则就是先收到了signal信号。因为如果是先收到signal信号，节点的状态应该是NODE.SIGNAL，而且节点会进入同步阻塞队列。这样做可以有效避免signal信号丢失。

线程被唤醒后需要重新申请锁，调用acquireQueued方法来实现，这一步和前面我们提到的申请流程类似，这里就不再重复了。

当条件满足后，线程被唤醒，这时候我们就要用到Condition的signal()方法了。signal方法的时序图如下：

这部分就是从条件队列中找到第一个没有取消的节点，然后唤醒它。实现transferForSignal方法的具体代码如下：

这个方法有三个要点。

第一，要使用CAS尝试将节点状态从CONDITION转化为0。如果更新失败，说明节点已取消，需要返回false，继续通知下一个等待线程。

第二，将线程从条件阻塞队列放入到同步阻塞队列，这一步非常关键，可以防止signal信号丢失。

第三，如果线程加入同步队列后，其前置节点的状态为已取消，或者将其设置为signal失败，则直接唤醒线程。

signal方法内部的实现机制就是确保线程要么在同步队列中，要么在条件等待队列中。这样可以有效防止通知信号丢失，避免线程一直被阻塞。

到这里，Condition的await和signal方法就都介绍完了。

总结

这节课，我们首先通过一个简单的生活场景，并结合生产者-消费者模型引出了锁的基本结构，它包括：锁要保护的资源、锁的拥有者、同步阻塞队列和条件等待队列。

紧接着，我们以源码分析为主要手段，辅助流程图、时序图，一步一步地实现了锁的申请和释放。

同步阻塞队列存放的都是竞争锁失败的线程，主要表征的是线程之间的竞争、互斥，而条件等待队列中存储的是因为某一个条件不满足而需要阻塞的线程，通常需要被其他线程主动唤醒，主要表征的是线程协作。

我们可以使用LockSupport.parkNanos来阻塞线程，并通过LockSupport.unpark方法来唤醒线程。

如果你对中间件感兴趣，对锁的语义的理解必不可少。它虽然有一定难度，但是只要攻下了源码，读懂AQS，对锁的理解与认知能力会有一个质的提升，对多线程协作开发大有裨益。

JUC的体系非常庞大，这节课不能全面覆盖，但是只要掌握了AQS，后面再去学习CountDownLatch、信号量、CAS等知识会变得非常容易。如果你有兴趣，也可以读一读我写过的和锁相关的文章：《锁的优化思路》和《disruptor无锁化设计实践》，应该可以给你更多的启发。

课后题

最后，还是给你留一道课后题。

请你尝试写一篇文章，分析JUC读写锁的源码，重点剖析读锁的申请与释放还有写锁的申请与释放。我也给你提供一篇文章供你参考：《Java并发锁ReentrantReadWriteLock读写锁源码分析》。

关于这节课，如果你还有不理解的问题，也欢迎你在留言区留言。我们下节课再见！

07 | NIO：手撸一个简易的主从多Reactor线程模型

作者: 丁威

你好，我是丁威。

中间件是互联网发展的产物，而互联网有一个非常显著的特点：集群部署、分布式部署。当越来越多的服务节点分布在不同的机器上，高效地进行网络传输就变得更加迫切了。在这之后，一大批网络编程类库如雨后春笋般出现，经过不断的实践表明，Netty框架几乎成为了网络编程领域的不二之选。

接下来的两节课，我们会通过对NIO与Netty的详细解读，让你对网络编程有一个更直观的认识。

NIO和BIO模型的工作机制

NIO是什么呢？简单来说，NIO就是一种新型IO编程模式，它的特点是同步、非阻塞。

很多资料将NIO中的“N”翻译为New，即新型IO模型，既然有新型的IO模式，那当然也存在中老型的IO模型，这就是BIO，同步阻塞IO模型。

定义往往是枯燥的，我们结合实际场景看一下BIO和NIO两种IO通讯模式的工作机制，更直观地感受一下它们的差异。

MySQL的客户端(mysql-connector-java)采用的就是BIO模式，它的工作机制如下图所示：

我们模拟场景，向MySQL服务端查询表中的数据，这时会经过四个步骤。

第一步，应用程序拼接SQL，然后mysql-connector-java会将SQL语句按照MySQL通讯协议编码成二进制，通过网络API将数据写入到网络中进行传输。底层最终是使用Socket的OutputStream的write与flush这两个方法实现的。

第二步，调用完write方法后，再调用Socket的InputStream的read方法，读取服务端返回数据，此时会阻塞等待。

第三步，服务端在收到请求后会解析请求，从请求中提取出对应的SQL语句，然后按照SQL抽取数据。服务端在处理这些业务逻辑时，客户端阻塞，不能做其他事情，我把这个阶段称之为等待数据阶段。

第四步，服务端执行完指定逻辑，抽取到合适的数据后，会调用Socket的OutputStream的write将响应结果通过网络传递到客户端。此时，客户端用read方法从网卡中把数据读取到应用程序的内存中，此阶段我称之为数据传输阶段。

BIO的IO模型在等待数据阶段、数据传输阶段都会阻塞。其实，“IO模型”的名称基本就是这两个阶段的特质决定的。

在等待数据阶段，如果发起网络调用后，在服务端数据没有准备好的情况下客户端会阻塞，我们称为阻塞IO；如果数据没有准备好，但网络调用会立即返回，我们称之为非阻塞IO。

在数据传输阶段，如果发起网络调用的线程还可以做其他事情，我们称之为异步，否则称之为同步。

这样看来，BIO的完整名称叫做“同步阻塞IO”也就不足为奇了。

从JDK1.4开始，Java又引入了另外一种IO模型：NIO。

虽然MySQL客户端主要使用的是BIO模型，但是我们可以演示一下MySQL Client采用NIO与MySQL服务端通信的样子：

NIO与BIO的不同点在于，在调用read方法时，如果服务端没有返回数据，该方法不会阻塞当前调用线程，read方法的返回值会为本次网络调用实际读取到的字节数量。也就是说，客户端调用一次read方法，如果本次没有读取到数据，线程可以继续处理其他事情，然后在需要数据的时候再次调用，但是在数据返回的过程中同样会阻塞线程。这也是NIO全名的由来：同步非阻塞IO。

NIO提供了在数据等待阶段的灵活性，但如果需要客户端反复调用读相关的API进行测试，编程体验也极不友好，为了改进NIO网络模型的缺陷，又引入了“事件就绪选择机制”。

事件就绪选择机制指的是，应用程序只需要在通道（网络连接）上注册感兴趣的事件（如网络读事件），客户端向服务端发送请求后，无须立即调用read方法去尝试读取响应结果，而是等服务端将准备好的数据通过网络传输到客户端的网卡。这时，操作系统会通知客户端“数据已到达”，此时客户端再调用读取API，从中读取响应结果。其实我们现在说NIO，说的就是“NIO + 事件就绪选择”的合体。

NIO和BIO模型的使用场景

那BIO与NIO相比，有什么优劣势呢？它们对应的使用场景是什么？为了直观地展示两种编程模型的优缺点，我们用网络游戏这个场景来举例。

一个简易的网络游戏分为服务端与客户端（玩家）两个端口，我们一起来思考一下，如果游戏服务端分别使用BIO技术和NIO技术进行架构设计，结果会是怎样的。

BIO领域一种经典的设计范式是每个请求对应一个线程。我们就用这种思想设计一下游戏的服务端，设计图如下：

游戏服务端后端的设计思想是：采用长连接模式。每当一个客户端上线，服务端就会为请求创建一个线程，在独立的线程中和客户端进行网络读写，一旦客户端下线，就会关闭对应的线程。

但是一台服务器能创建的线程个数是有限的，所以基于BIO模式构建的优秀服务端一个非常明显的弊端：在线用户数越多，需要创建的线程数就越多，支撑的并发在线用户数量受到明显制约。更加严重的问题是，服务端与其中某些客户端并不是一直在通信，大量线程的网络连接处于阻塞状态，线程资源无法得到有效利用。

为了防止因为线程急剧膨胀、线程资源耗尽影响到服务端的设计，这时候我们通常会引入线程池。因为引入线程池就相当于是在限流，超过线程池规定的线程数量，服务器就会拒绝连接。

对于需要支持大量在线并发用户（连接）的服务器来说，BIO的网络IO模型绝对不是一个好的选择。

我们再来看下NIO模式。基于NIO模式设计的游戏服务端模型如下图所示：

基于NIO，在业界有一种标准的线程模型Reactor，在这节课的后半部分我们还会详细介绍，这里我们先说明一下NIO的优势。

首先，服务端会创建一个线程池专门处理网络读写，我们称之为IO线程池。IO线程池内会内置NIO的事件选择器。当游戏服务端监听到一个客户端时，会从IO线程池中根据负载均衡算法选择一个IO线程，将其注册到事件选择器中。

事件选择器会定时进行事件轮询，挑选=出数据进行传输（读取或写入），执行事件选择，然后在IO线程中按连接分别读取数据。在将请求解码后，丢到业务线程中执行对应的业务逻辑，它的主要功能是分担IO线程的压力，做到尽量不阻塞IO线程。

使用NIO可以做到用少量线程来服务大量连接，哪怕客户端连接数增长，也不会造成服务端线程膨胀。这个优势的关键点在于，基于事件选择机制，IO线程都在进行有效的读写，而不像BIO那样，在没有数据传输时还得占用线程资源。

也正是因此，NIO非常适合需要同时支持大量客户端在线的场景。在NIO模型下，单个请求的数据包建议不要太大。

值得注意的是，一个IO线程在一次事件就绪选择可能会有多个网络连接具备了读或写的准备，但此时对这些网络通道是串行执行的，所以如果每一个网络通道需要读或写的数据比较大，这就必然导致其他连接的延时。

既然NIO这么优秀，那为什么MySQL数据访问客户端还是采用BIO模式呢？为啥不改造成NIO呢？

其实在进行技术选型时，并不是越新的技术就越好，我们还是要结合具体问题具体分析。

我们再回过头来看MySQL客户端的场景。目前在应用层面，我们会为每一个应用配置一个数据库连接池。当业务线程需要进行数据库操作时，它会尝试从数据库连接池获取一个数据库连接（底层是一条TCP连接，负责与服务端进行网络的读与写），然后使用这条连接发送SQL语句并获取SQL结果。任务结束之后，业务线程会把数据库连接归还给连接池，从而实现数据库连接的复用。

与此同时，我们为了保证数据库服务端的可用性，通常需要强制限制客户端能使用的连接数量。这就注定了数据库客户端没有需要支持大量连接的诉求，在这个场景下，客户端使用阻塞型IO对保护数据库服务端更有优势。

简单说明一下。假设业务代码存在缺陷，导致需要执行一条SQL语句来获取大量数据。这时，我们要尝试从数据库连接池中获取连接，并通过这个连接向MySQL服务端发送SQL语句。由于这条SQL语句的执行性能很差，这条连接在客户端一直被阻塞，无法继续发送更多的SQL。另外如果数据库连接池中没有空闲连接，再尝试获取连接时还需要等待连接被释放，服务器缓慢的执行速度确保了客户端不能持续发送新的请求，对保护数据库服务器大有裨益。

这种情况下如果使用NIO模型，客户端会无节制地用一条连接发送大量请求，导致服务端出现完全不可用的情况。

总结一下就是，NIO模型更适合需要大量在线活跃连接的场景，常见于服务端；BIO模型则适合只需要支持少量连接的场景，常常用于客户端，这也是MySQL数据访问客户端会在网络IO模型方面使用BIO的原因。

Reactor线程模型

学习NIO的理论知识非常枯燥，而且很难做到透彻地理解，我们需要一个实例来深入进去。结合我的学习经验，我觉得学习Reactor经典线程模型，尝试编写一个Reactor线程模型对提升NIO的理解非常有帮助。

为什么这么说呢？因为在编写网络通信相关的功能模块时，建立一套线程模型是非常重要的一环，经过各位前辈不断的实践，Reactor线程模型已成为NIO领域的事实标准，无论是网络编程类库NIO，还是Kafka、Dubbo等主流中间件的底层网络模型都是直接或间接受到了Reactor模型的影响。

那什么是Reactor线程模型？怎么使用NIO来实现Reactor模型？这两个问题，就是我们这节课后半部分的重点。

什么是Reactor线程模型？

Reactor主从多Reactor模型的架构设计如下图所示：

说明一下各个角色的职责。

• Acceptor：请求接收者，作用是在特定端口建立监听。
• Main Reactor Thread Pool：主Reactor模型，主要负责处理OP_ACCEPT事件（创建连接），通常一个监听端口使用一个线程。在具体实践时，如果创建连接需要进行授权校验（Auth）等处理逻辑，也可以直接让Main Reactor中的线程负责。
• NIO Thread Group（ IO 线程组）：在Reactor模型中也叫做从Reactor，主要负责网络的读与写。当Main Reactor Thread 线程收到一个新的客户端连接时，它会使用负载均衡算法从NIO Thread Group中选择一个线程，将OP_READ、OP_WRITE事件注册在NIO Thread的事件选择器中。接下来这个连接所有的网络读与写都会在被选择的这条线程中执行。
• NIO Thread：IO线程。负责处理网络读写与解码。IO线程会从网络中读取到二进制流，并从二进制流中解码出一个个完整的请求。
• 业务线程池：通常IO线程解码出的请求将转发到业务线程池中运行，业务线程计算出对应结果后，再通过IO线程发送到客户端。

我们再通过一个网络通信图进一步理解Reactor线程模型。

网络通信的交互过程通常包括下面六个步骤。

1. 启动服务端，并在特定端口上监听，例如，web 应用默认在80端口监听。
2. 客户端发起TCP的三次握手，与服务端建立连接。这里以 NIO 为例，成功建立连接后会创建NioSocketChannel对象。
3. 服务端通过 NioSocketChannel 从网卡中读取数据。
4. 服务端根据通信协议从二进制流中解码出一个个请求。
5. 根据请求执行对应的业务操作，例如，Dubbo 服务端接受了请求，并根据请求查询用户ID为1的用户信息。
6. 将业务执行结果返回到客户端（通常涉及到协议编码、压缩等）。

线程模型需要解决的问题包括：连接监听、网络读写、编码、解码、业务执行等，那如何运用多线程编程优化上面的步骤从而提升性能呢？

主从多Reactor模型是业内非常经典的，专门解决网络编程中各个环节问题的线程模型。各个线程通常的职责分工如下。

1. Main Reactor 线程池，主要负责连接建立（OP_ACCEPT），即创建NioSocketChannel后，将其转发给SubReactor。
2. SubReactor 线程池，主要负责网络的读写（从网络中读字节流、将字节流发送到网络中），即监听OP_READ、OP_WRITE，并且同一个通道会绑定一个SubReactor线程。

编码、解码和业务执行则具体情况具体分析。通常，编码、解码会放在IO线程中执行，而业务逻辑的执行会采用额外的线程池。但这不是绝对的，一个好的框架通常会使用参数来进行定制化选择，例如 ping、pong 这种心跳包，直接在 IO 线程中执行，无须再转发到业务线程池，避免线程切换开销。

怎么用NIO实现Reactor模型？

理解了Reactor线程模型的内涵，接下来就到了实现这一步了。

我建议你在学习这部分内容时，同步阅读一下《Java NIO》这本电子书的前四章。这本书详细讲解了NIO的基础知识，是我学习Netty的老师，相信也会给你一些帮助。

我们先来看一下Reactor模型的时序图，从全局把握整体脉络：

这里核心的流程有三个。

• 服务端启动，会创建MainReactor线程池，在MainReactor中创建NIO事件选择器，并注册OP_ACCEPT事件，然后在指定端口监听客户端的连接请求。
• 客户端向服务端建立连接，服务端OP_ACCEPT对应的事件处理器被执行，创建NioSocketChannel对象，并按照负载均衡机制将其转发到SubReactor线程池中的某一个线程上，注册OP_READ事件。
• 客户端向服务端发送具体请求，服务端OP_READ对应的事件处理器被执行，它会从网络中读取数据，然后解码、转发到业务线程池执行具体的业务逻辑，最后将返回结果返回到客户端。

我们解读下核心类的核心代码。

NioServer的代码如下：

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

private static class Acceptor implements Runnable {
        // main Reactor 线程池，用于处理客户端的连接请求
        private static ExecutorService mainReactor = Executors.newSingleThreadExecutor(new ThreadFactory() {
            private AtomicInteger num = new AtomicInteger(0);
            @Override
            public Thread newThread(Runnable r) {
                Thread t = new Thread(r);
                // 为线程池中的名称进行命名，方便分析线程栈
                t.setName("main-reactor-" + num.incrementAndGet());
                return t;
            }
        });
        public void run() {
            // NIO中服务端对应的Channel
            ServerSocketChannel ssc = null;
            try {
              // 通过静态方法创建一个ServerSocketChannel对象
                ssc = ServerSocketChannel.open();
              //设置为非阻塞模式
                ssc.configureBlocking(false);
              //绑定端口
                ssc.bind(new InetSocketAddress(SERVER_PORT));
​
                //转发到 MainReactor反应堆
                dispatch(ssc);
                System.out.println("服务端成功启动。。。。。。");
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        private void dispatch(ServerSocketChannel ssc) {
            mainReactor.submit(new MainReactor(ssc));
        }
    }

启动服务端会创建一个Acceptor线程，它的职责就是绑定端口，创建ServerSocketChannel，然后交给MainReactor去处理接收连接的逻辑。

MainReactor的具体实现如下：

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

public class MainReactor implements Runnable{
    // NIO 事件选择器
    private Selector selector;
    // 子ReactorThreadGroup 即IO线程池
    private SubReactorThreadGroup subReactorThreadGroup;
    // IO线程池默认线程数量
    private static final int DEFAULT_IO_THREAD_COUNT = 4;
    // IO线程个数
    private int ioThreadCount = DEFAULT_IO_THREAD_COUNT;
  
    public MainReactor(ServerSocketChannel channel) {
        try {
            // 创建事件选择器
            selector = Selector.open();
            // 为通道注册OP_ACCEPT 事件，客户端发送数据后，服务端通过该事件进行数据的读取
            channel.register(selector, SelectionKey.OP_ACCEPT);
        } catch (IOException e) {
            e.printStackTrace();
        }
        // IO线程池，里面包含负载均衡算法
        subReactorThreadGroup = new SubReactorThreadGroup(ioThreadCount);
    }
    public void run() {
        System.out.println("MainReactor is running");
        while (!Thread.interrupted()) {
            Set<SelectionKey> ops = null;
            try {
                // 进行事件选择
                selector.select(1000);
                // 经过事件选择后已经就绪的事件
                ops = selector.selectedKeys(); 
            } catch (IOException e) {
                e.printStackTrace();
            }
            // 处理相关事件
            for (Iterator<SelectionKey> it = ops.iterator(); it.hasNext();) {
                SelectionKey key = it.next();
                it.remove();
                try {
                     // 如果有客户端尝试建立连接
                    if (key.isAcceptable()) {
                        System.out.println("收到客户端的连接请求。。。");
                        //获取服务端的ServerSocketChannel对象， 这里其实，可以直接使用ssl这个变量
                        ServerSocketChannel serverSc = (ServerSocketChannel) key.channel();
                        // 调用ServerSocketChannel的accept方法，创建SocketChannel
                        SocketChannel clientChannel = serverSc.accept();
                        // 设置为非阻塞模式
                        clientChannel.configureBlocking(false);
                        // 转发到IO线程，由对应的IO线程去负责网络读写
                        subReactorThreadGroup.dispatch(clientChannel); // 转发该请求
                    }
                } catch (Throwable e) {
                    e.printStackTrace();
                    System.out.println("客户端主动断开连接。。。。。。。");
                }
            }
        }
    }
}

SubReactorThreadGroup内部包含一个SubReactorThread数组，并提供负载均衡机制，供MainReactor线程选择具体的SubReactorThread线程，具体代码如下：

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

public class SubReactorThreadGroup {
    private static final AtomicInteger requestCounter = new AtomicInteger();  //请求计数器
    // 线程池IO线程的数量
    private final int ioThreadCount;  
    // 业务线程池大小
    private final int businessTheadCout; 
    private static final int DEFAULT_NIO_THREAD_COUNT;
    // IO线程池数组
    private SubReactorThread[] ioThreads;
    //业务线程池
    private ExecutorService businessExecutePool; 
​
    static {
        DEFAULT_NIO_THREAD_COUNT = 4;
    }
​
    public SubReactorThreadGroup() {
        this(DEFAULT_NIO_THREAD_COUNT);
    }
    public SubReactorThreadGroup(int ioThreadCount) {
        if(ioThreadCount < 1) {
            ioThreadCount = DEFAULT_NIO_THREAD_COUNT;
        }
        //暂时固定为10
        businessTheadCout = 10;
        //初始化代码
        businessExecutePool = Executors.newFixedThreadPool(businessTheadCout, new ThreadFactory() {
            private AtomicInteger num = new AtomicInteger(0);
            @Override
            public Thread newThread(Runnable r) {
                Thread t = new Thread(r);
                t.setName("bussiness-thread-" + num.incrementAndGet());
                return t;
            }
        });
        this.ioThreadCount = ioThreadCount;
        this.ioThreads = new SubReactorThread[ioThreadCount];
        for(int i = 0; i < ioThreadCount; i ++ ) {
            this.ioThreads[i] = new SubReactorThread(businessExecutePool);
            this.ioThreads[i].start(); //构造方法中启动线程，由于nioThreads不会对外暴露，故不会引起线程逃逸
        }
        System.out.println("Nio 线程数量：" + ioThreadCount);
    }
    public void dispatch(SocketChannel socketChannel) {
         //根据负载算法转发到具体IO线程
        if(socketChannel != null ) {
            next().register(new NioTask(socketChannel, SelectionKey.OP_READ));
        }
    }
    protected SubReactorThread next() {
        return this.ioThreads[ requestCounter.getAndIncrement() %  ioThreadCount ];
    }

SubReactorThread IO线程的具体实现如下：

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
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136

public class SubReactorThread extends Thread{
  // 事件选择器
    private Selector selector;
  //业务线程池
    private ExecutorService businessExecutorPool;
  //任务列表
    private List<NioTask> taskList = new ArrayList<NioTask>(512);
  // 锁
    private ReentrantLock taskMainLock = new ReentrantLock();
    /**
     * 业务线程池
     * @param businessExecutorPool
     */
    public SubReactorThread(ExecutorService businessExecutorPool) {
        try {
            this.businessExecutorPool = businessExecutorPool;
            //创建事件选择器
            this.selector = Selector.open();
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
    /**
     * 接受读写任务
     *
     */
    public void register(NioTask task) {
        if (task != null) {
            try {
                taskMainLock.lock();
                taskList.add(task);
            } finally {
                taskMainLock.unlock();
            }
        }
    }
    /**
     * 此处的reqBuffer处于可写状态
     * @param sc
     * @param reqBuffer
     */
    private void dispatch(SocketChannel sc, ByteBuffer reqBuffer) {
        businessExecutorPool.submit( new Handler(sc, reqBuffer, this)  );
    }
​
​
​
    public void run() {
        while (!Thread.interrupted()) {
            Set<SelectionKey> ops = null;
            try {
                //执行事件选择
                selector.select(1000);
              // 获取已就绪的事件集合
                ops = selector.selectedKeys();
            } catch (IOException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
                continue;
            }
            // 处理相关事件
            for (Iterator<SelectionKey> it = ops.iterator(); it.hasNext();) {
                SelectionKey key = it.next();
                it.remove();
                try {
                     // 通道写事件就绪，说明可以继续往通道中写数据
                    if (key.isWritable()) { 
                        SocketChannel clientChannel = (SocketChannel) key.channel();
                        // 获取上次未写完的数据
                        ByteBuffer buf = (ByteBuffer) key.attachment();
                        // 将其写入到通道中。
                        // 这里实现比较粗糙，需要采用处理taskList类似的方式，因为此时通道缓冲区有可能已写满
                        clientChannel.write(buf);
                        System.out.println("服务端向客户端发送数据。。。");
                        // 重新注册读事件
                        clientChannel.register(selector, SelectionKey.OP_READ);
                    } else if (key.isReadable()) { // 接受客户端请求
                        System.out.println("服务端接收客户端连接请求。。。");
                        SocketChannel clientChannel = (SocketChannel) key.channel();
                        ByteBuffer buf = ByteBuffer.allocate(1024);
                        System.out.println(buf.capacity());
                        /**
                         * 这里其实实现的非常不优雅，需要对读取处理办关闭，而且一次读取，并不一定能将一个请求读取
                         * 一个请求，也不要会刚好读取到一个完整对请求，
                         * 这里其实是需要编码，解码，也就是通信协议  @todo
                         * 这里如何做，大家可以思考一下，后面我们可以体验netty是否如何优雅处理的。
                         */
                        int rc = clientChannel.read(buf);//解析请求完毕
                        //转发请求到具体的业务线程；当然，这里其实可以向dubbo那样，支持转发策略，如果执行时间短，
                        //，比如没有数据库操作等，可以在io线程中执行。本实例，转发到业务线程池
                        dispatch(clientChannel, buf);
                    }
                } catch (Throwable e) {
                    e.printStackTrace();
                    System.out.println("客户端主动断开连接。。。。。。。");
                }
            }
​
            // 处理完事件后，我们还需要处理其他任务，这些任务通常来自业务线程需要IO线程执行的任务
            if (!taskList.isEmpty()) {
                try {
                    taskMainLock.lock();
                    for (Iterator<NioTask> it = taskList
                            .iterator(); it.hasNext();) {
                        NioTask task = it.next();
                        try {
                            SocketChannel sc = task.getSc();
                            if(task.getData() != null ) { // 写操作
                                ByteBuffer byteBuffer = (ByteBuffer)task.getData();
                                byteBuffer.flip();
                                // 如果调用通道的写函数，如果写入的字节数小于0，并且待写入还有剩余空间，说明缓存区已满
                                // 需要注册写事件，等缓存区空闲后继续写入
                                int wc = sc.write(byteBuffer);
                                System.out.println("服务端向客户端发送数据。。。");
                                if(wc < 1 && byteBuffer.hasRemaining()) { // 说明写缓存区已满，需要注册写事件
                                    sc.register(selector, task.getOp(), task.getData());
                                    continue;
                                } 
                                byteBuffer = null;//释放内存
                            } else {
                                sc.register(selector, task.getOp());
                            }
                        } catch (Throwable e) {
                            e.printStackTrace();// ignore
                        }
                        it.remove();
                    }
​
                } finally {
                    taskMainLock.unlock();
                }
            }
        }
    }
}

IO线程负责从网络中读取二进制并将其解码成具体请求，然后转发到业务线程池执行。

接下来，业务线程池会执行业务代码并将响应结果通过IO线程写入到网络中，我们对业务进行简单的模拟：

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

public class Handler implements Runnable{
​
  // 模拟业务处理
    private static final byte[] b = "hello,服务器收到了你的信息。".getBytes(); // 服务端给客户端的响应
    // 网络通道
    private SocketChannel sc;
    // 请求报文
    private ByteBuffer reqBuffer;
    // IO线程
    private SubReactorThread parent;
​
    public Handler(SocketChannel sc, ByteBuffer reqBuffer,
                   SubReactorThread parent) {
        super();
        this.sc = sc;
        this.reqBuffer = reqBuffer;
        this.parent = parent;
    }
​
    public void run() {
        System.out.println("业务在handler中开始执行。。。");
        // TODO Auto-generated method stub
        //业务处理
        reqBuffer.put(b);
      // 业务处理完成后，通过向IO线程提交任务
        parent.register(new NioTask(sc, SelectionKey.OP_WRITE, reqBuffer));
        System.out.println("业务在handler中执行结束。。。");
    }
}

我们再来看一下客户端创建连接的代码：

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

public class NioClient {
    public static void main(String[] args) {
        // socket
        SocketChannel clientClient;
        // 事件选择器
        Selector selector = null;
        try {
            // 创建网络通道
            clientClient = SocketChannel.open();
            // 设置为非阻塞模型
            clientClient.configureBlocking(false);
            selector = Selector.open();
            // 注册连接成功事件，在与服务端通过tcp三次握手建立连接后可以收到该事件
            clientClient.register(selector, SelectionKey.OP_CONNECT);
          //建立连接，该方法会立即返回
            clientClient.connect(new InetSocketAddress("127.0.0.1",9080));
            Set<SelectionKey> ops = null;
            while(true) {
                try {
                   // 执行事件选择
                    selector.select();
                    ops = selector.selectedKeys();
                    for (Iterator<SelectionKey> it = ops.iterator(); it.hasNext();) {
                        SelectionKey key = it.next();
                        it.remove();
                        if(key.isConnectable())  //连接事件
                            System.out.println("client connect");
                            SocketChannel sc =  (SocketChannel) key.channel();
                            // 判断此通道上是否正在进行连接操作。
                            // 完成套接字通道的连接过程。
                            if (sc.isConnectionPending()) {
                                sc.finishConnect();
                                System.out.println("完成连接!");
                              
                                // 完成连接后，向服务端发送请求包
                                ByteBuffer buffer = ByteBuffer.allocate(1024);
                                buffer.put("Hello,Server".getBytes());
                                buffer.flip();
                                sc.write(buffer);
                            }
                            // 注册读事件，等待服务端响应包到达
                            sc.register(selector, SelectionKey.OP_READ);
                        } else if(key.isWritable()) {
                            System.out.println("客户端写");
                            SocketChannel sc = (SocketChannel)key.channel();
                            //这里是NIO ByteBuffer的基本API
                            ByteBuffer buffer = ByteBuffer.allocate(1024);
                            buffer.put("hello server.".getBytes());
                            buffer.flip();
                            sc.write(buffer);
                        } else if(key.isReadable()) {
                            System.out.println("客户端收到服务器的响应....");
                            SocketChannel sc = (SocketChannel)key.channel();
                            ByteBuffer buffer = ByteBuffer.allocate(1024);
                            int count = sc.read(buffer);
                            if(count > 0 ) {
                                buffer.flip();
                                byte[] response = new byte[buffer.remaining()];
                                buffer.get(response);
                                System.out.println(new String(response));
                            }
                            // 再次发送消息，重复输出
                            buffer = ByteBuffer.allocate(1024);
                            buffer.put("hello server.".getBytes());
                            buffer.flip();
                            sc.write(buffer);
                        }
                    }
                } catch(Throwable e) {
                    e.printStackTrace();
                }
            }
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
}

这样，一个Reactor模型就搭建好了。如果你想完整地学习这个Reactor模型的详细代码，可以到我的GitHub上查看。

总结

好了，这节课就讲到这里。

这节课，我们先结合场景介绍了BIO与NIO两种网络编程模型和它们的优缺点。

根据等待数据阶段和数据传输阶段这两个阶段的特质，我们可以得到BIO的全称同步阻塞IO，还有NIO的全称同步非阻塞IO。NIO模型更适合需要大量在线活跃连接的场景，常见于服务端；BIO模型则适合只需要支持少量连接的场景。

我们还了解了一个业内非常经典的线程模型：主从多Reactor模型。它的核心设计理念是让线程分工明确，相互协作。Main Reactor 线程池主要负责连接建立，SubReactor 线程池主要负责网络的读写，而编码、解码和业务执行则需要具体情况具体分析。

最后，我还带你使用NIO技术实现了主从多Reactor模型，给你推荐了一本学习NIO必备的电子书《Java NIO》，这本书非常详细介绍了NIO的三大金刚：缓存、通道和选择器的各类基础知识。我建议你在阅读完本电子书后，再来反复看看这个Reactor示例，相信可以在你进修NIO的基础上助你一臂之力。

课后题

学完这节课，我也给你出两道课后题。

1. 为什么NIO不适合请求体很大的场景？
2. 请你详细阅读《Java NIO》这本书中Reactor模型的示例子代码，尝试实现一个简易的RPC Request-Response模型。
   

例如，模拟Dubbo服务调用需要传入基本的参数：包名、方法名，参数。客户端发送这些数据后，服务端根据接收的数据，在服务端要正确打印包名、方法名、参数，并向客户端返回 “hello,收到请求” + 包名 + 方法名。

欢迎你在留言区留下你的思考结果，我们下节课见！

08 | Netty：如何优雅地处理网络读写，制定网络通信协议？

作者: 丁威

你好，我是丁威。

上一节课，我们介绍了中间件领域最经典的网络编程模型NIO，我也在文稿的最后给你提供了用NIO模拟Reactor线程模型的示例代码。如果你真正上手了，你会明显感知到，如果代码处理得过于粗糙，只关注正常逻辑却对一些异常逻辑考虑不足，就不能成为一个生产级的产品。

这是因为要直接基于NIO编写网络通讯层代码，需要开发者拥有很强的代码功底和丰富的网络通信理论知识。所以，为了降低网络编程的门槛，Netty框架就出现了，它能够对NIO进行更高层级的封装。

从这之后，开发人员只需要关注业务逻辑的开发就好了，网络通信的底层可以放心交给Netty，大大降低了网络编程的开发难度。

这节课，我们就来好好谈谈Netty。

我会先从网络编程中通信协议、线程模型这些网络编程框架的共性问题入手，然后重点分析Netty NIO的读写流程，最后通过一个Netty编程实战，教会你怎么使用Netty解决具体问题，让你彻底掌握Netty。

通信协议

如果你不从事中间件开发工作，那估计网络编程对你来说会非常陌生，为了让你对它有一个直观的认知，我给你举一个例子。

假如我们在使用Dubbo构建微服务应用，Dubbo客户端在向服务提供者发起远程调用的过程中，需要告诉服务提供者服务名、方法名和参数。但这些参数是怎么在网络中传递的呢？服务提供者又怎么识别出客户端的意图呢？

你可以先看看我画的这张图：

客户端在发送内容之前需要先将待发送的内容序列化为二进制流。例如，上图发送了两个包，第一个包的二进制流是0110，第二个包的二进制流是00110011。这时，服务端读取数据的情形可能有两种。

1. 经过多次读取：在上面这张图中，服务端调用了3次read方法才把数据全部读取出来，分别读取到的包是011、000、110011。
2. 调用一次read就读取到所有数据：例如011000110011。

这里我插播一个小知识，一次read方法能读取到的数据量，要取决于网卡中可读数据和接收缓冲区的大小。

那服务端是如何正确识别出0110就是第一个请求包，00110011是第二个请求包的呢？它为什么不会将011当成第一个请求包，000当成第二个请求包，110011当成第三个包，或者直接将011000110011当成一个请求包呢？其实这种现象叫做粘包。

常用的解决方案是客户端与服务端共同制定一个通信规范（也称通信协议），用它来定义请求包/响应包的具体格式。这样，客户端发送请求之前，需要先将内容按照通信规范序列化成二进制流，这个过程称之为编码；同样，服务端会按照通信规范将收到的二进制流进行反序列化，这个过程称之为解码。

从这里你也可以看出，网络编程中通常涉及编码、往网络中写数据（Write）、从网络中读取数据（Read）、解码、业务逻辑处理、发送响应结果和接受响应结果等步骤，你可以看下下面这张图，加深理解：

那如何制订通信协议呢？

通信协议的制订方法有很多，有的是采用特殊符号来标记一个请求的结束，但如果请求体中也包含这个分隔符就会使协议破坏，还有一种方法是使用固定长度来表述一个请求包，定义一个请求包固定包含多少字节，如果请求体内存不足，就使用填充符合进行填充，但这种方式会造成空间的浪费。

业界最为经典的协议设计方法是协议头+Body的设计理念，如图所示：

这里有几个关键点，你需要注意一下：

• 协议头的长度是固定的，通常为识别出一个业务的最小长度；
• 协议头中会包含一个长度字段，用来标识一个完整包的长度，用来表示长度字段的字节位数直接决定了一个包的最大长度；
• 消息体中存储业务数据。

为了更直观地给你展示，我直接以一个简单的 RPC 通信场景为例，实现类似Dubbo服务远程服务调用，通信协议设计如下图所示：

这里我们演示的是基于Header+Body的设计模式，接受端从网络中读取到字节后解码的流程。接受端将读取到的数据存储在一个接收缓冲区，在Netty中称为累积缓冲区。

首先我们要判断累积缓存区中是否包含一个完整的Head，例如上述示例中，一个包的Header 的长度为 6 个字节，那首先判断累积缓存中可读字节数是否大于等于 6，如果不足 6 个字节，跳过本次处理，等待更多数据到达累积缓存区。

如果累积缓存区中包含一个完整的Header，就解析头部，并且提取长度字段中存储的数值，即包长度，然后判断累积缓存区中可读字节数是否大于或等于整个包的长度。如果累积缓存区不包含一个完整的数据包，则跳过本次处理，等待更多数据到达累积缓存区。如果累积缓存区包含一个完整的包，则按照通信协议的格式按顺序读取相关的内容。

通过上面这种方式，我们就可以完美解决粘包问题了。

我们前面也说了，网络编程中包含编码、解码、网络读取、业务逻辑等多个步骤，所以如何使用多线程提升并发度，合理处理多线程之间的高效协作就显得尤为重要，接下来我们来看一下Netty的线程模型是怎么做的。

Netty的线程模型采取的是业界的主流线程模型，也就是主从多Reactor模型：

它的设计重点主要包括下面这几个方面。

• Netty Boss Group线程组

主要处理OP_ACCEPT事件，用于处理客户端链接，默认为1个线程。当Netty Boss Group线程组接收到一个客户端链接时，会创建NioSocketChannel对象，并封装成Channel对象，在Channel对象内部会创建一个缓冲区。这个缓冲区可以接收需要通过这个通道写入到对端的数据，然后从Netty Work Group线程组中选择一个线程并注册读事件。

• Nettty Work Group线程组

主要处理OP_READ、OP_WRITE事件，处理网络的读与写，所以也称为IO线程组，线程组中线程个数默认为CPU的核数。由于注册了读事件，所以当客户端发送请求时，读事件就会触发，从网络中读取请求，进入请求处理流程。

• 扩展机制采用责任链设计模式

编码、解码等功能对应一个独立的Handler，这些Handler默认在IO线程中执行，但Netty支持将Handler的执行放在额外的线程中执行，实现与IO线程的解耦合，避免IO线程阻塞。

• Business Thread Group

经过解码后得到一个完整的请求包，根据请求包执行业务逻辑，通常会额外引入一个独立线程池，执行业务逻辑后会将结果再通过IO线程写入到网络中。

业务线程在处理完业务逻辑后，通过调用通道将数据发送到目标端。但它并不能当下直接发送，而是要将数据放入到Channel中的写缓存区，并向IO线程提交一个写入任务。这里涉及到线程切换，因为所有的读写操作都需要在IO线程中执行（即一个通道的IO操作都是同一个线程触发的），避免了多线程编程的复杂性。

说到这里，我建议你停下来，尝试用NIO实现Netty的线程模型，检验一下自己对NIO的掌握程度。

理解了Netty的线程模型，接下来我们继续学习Netty是怎样处理读写流程的。在进入下面的学习之前，我有几个问题希望你先思考一下：

1. 如何处理连接半关闭？
2. 什么时候应该注册读事件？
3. 写数据之前一定要先注册写事件吗？

Netty如何处理网络读写事件？

Netty IO 读事件由 AbstractNioByteChannel 内部类 AbstractNioUnsafe 的 read 方法实现，接下来我们就来重点剖析一下这个方法，从中窥探 Netty 是如何实现 IO 读事件的。

由于AbstractNioUnsafe的read方法代码很长，我们分步进行解读。

第一步，如果没有开启自动注册读事件，在每一次读时间处理过后会取消读事件，代码片段如下：

1
2
3
4
5
6

final ChannelConfig config = config();
 if (!config.isAutoRead() && !isReadPending()) {
      // ChannelConfig.setAutoRead(false) was called in the meantime
      removeReadOp();
      return;
 }

这段代码背后蕴含的知识点是，事件注册是一次性的。例如，为通道注册了读事件，然后经事件选择器选择触发后，选择器不再监听读事件，再出来完成一次读事件后需要再次注册读事件。Netty中默认每次读取处理后会自动注册读事件，如果通道没有注册读事件，则无法从网络中读取数据。

第二步，为本次读取创建接收缓冲区，临时存储从网络中读取到的字节，代码片段如下：

1
2
3
4
5
6

final ByteBufAllocator allocator = config.getAllocator();
  final int maxMessagesPerRead = config.getMaxMessagesPerRead();
  RecvByteBufAllocator.Handle allocHandle = this.allocHandle;
  if (allocHandle == null) {
    this.allocHandle = allocHandle = config.getRecvByteBufAllocator().newHandle();
  }

创建接收缓存区需要考虑的问题是，该创建多大的缓存区呢？如果缓存区创建大了，就容易造成内存浪费；如果分配少了，在使用过程中就可能需要进行扩容，性能就会受到影响。

Netty在这里提供了扩展机制，允许用户自定义创建策略，只需实现RecvByteBufAllocator接口就可以了。它又包括两种实现方式：

• 分配固定大小，待内存不够时扩容；
• 动态变化，根据历史的分配大小，动态调整接收缓冲区的大小。

第三步，循环从网络中读取数据，代码片段如下：

1
2
3
4
5
6

do {
    byteBuf = allocHandle.allocate(allocator);
    int writable = byteBuf.writableBytes();
    int localReadAmount = doReadBytes(byteBuf);
    // 省略代码
  } while (++ messages < maxMessagesPerRead);

为什么要循环读取呢？为什么不一次性把通道中需要读取到的数据全部读完再继续下一个通道呢？

其实，这主要是为了避免单个通道占用太多时间，导致其他链接没有机会去读取数据。所以Netty会限制在一次读事件处理过程中调用底层读取API的次数，这个次数默认为16次。

接下来我们进行第四步。这里要提醒一下，第四步和第五步都是位于第三步的循环之中的。

第四步，调用底层SokcetChannel的read方法从网络中读取数据，代码片段如下：

1
2
3
4
5
6
7
8
9
10
11

byteBuf = allocHandle.allocate(allocator);
 int writable = byteBuf.writableBytes();
 int localReadAmount = doReadBytes(byteBuf);
 if (localReadAmount <= 0) {
    // not was read release the buffer
    byteBuf.release();
    byteBuf = null;
    close = localReadAmount < 0;
    break;
 }
pipeline.fireChannelRead(byteBuf);

解释一下，首先用writable存储接收缓存区可写字节数，然后通过调用底层NioSocketChannel从网络中读取数据，并返回本次读取的字节数。

那在什么情况下读取的字节数小于0呢？原来，TCP是全双工通信模型，任意一端都可以关闭接收或者写入，如果对端连接调用了关闭（半关闭），那么我们尝试从网络中读取字节时就会返回-1，跳出循环。

然后，我们要将读取到的内容传播到事件链中，事件链中各个事件处理器会依次对这些数据进行处理。

如果你也在使用Netty进行应用代码开发，请特别注意byteBuf的释放问题。自定义的事件处理器中要尽量继续调用fireChannelRead方法，Netty内置了一个HeadContext，它在实现时会主动释放ByteBuf。但如果自定义的事件处理器阻断了事件传播，请记得一定要释放ByteBuf，否则会造成内存泄露。

第五步，判断是否要跳出读取：

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

if (totalReadAmount >= Integer.MAX_VALUE - localReadAmount) {
    // Avoid overflow.
    totalReadAmount = Integer.MAX_VALUE;
    break;
  }
  totalReadAmount += localReadAmount;
  // stop reading
  if (!config.isAutoRead()) {
    break;
  }
  if (localReadAmount < writable) {
    // Read less than what the buffer can hold,
    // which might mean we drained the recv buffer completely.
    break;
  }

这里需要关注的一个点是，本次读取到的字节数如果小于接收缓冲区的可写大小，说明通道中已经没有数据可读了，结束本次读取事件的处理。

第六步，完成网络IO读取后，进行善后操作。具体代码片段如下：

1
2
3
4
5
6

pipeline.fireChannelReadComplete();
  allocHandle.record(totalReadAmount);
  if (close) {
    closeOnRead(pipeline);
    close = false;
  }

操作结束后，会触发一次读完成事件，并向整个事件链传播。这时候如果对端已经关闭了，则主动关闭链接。

就像我们在上节课提到的，事件机制触发后将失效，需要再次注册，所以Netty支持自动注册读事件。在每一次读事件完成后会主动调用下面这段代码实现读事件的自动注册，具体实现在HeadContext的fireChannelReadComplete方法中，代码片段如下：

1
2
3
4

public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
  ctx.fireChannelReadComplete();
  readIfIsAutoRead();//该方法最终会调用Channel的read方法，注册读事件
}

这里还涉及另一问题，那就是Netty的channelRead、channelReadComplete等事件是怎么传播的呢？我建议你查看我的另一篇文章《Netty4 事件处理传播机制》。

Netty网络读流程就讲到这里了，我们用一张流程图结束网络读取部分的讲解：

接下来，我们一起看看Netty的网络写入流程。

基于Netty网络模型，通常会使用一个业务线程池来执行业务操作，业务执行完成后，需要通过网络将响应结果提交给对应的IO线程，再通过IO线程将数据返回给客户端，其过程大致如下：

那在代码实现层面，业务线程与IO线程是怎么协作的呢？我们带着这个问题，继续深入研究Netty的网络写入流程。

在Netty中，一眼就能看到写事件的处理入口，也就是NioEventLoop（IO线程）的processSelectedKey方法，代码片段如下所示：

1
2
3
4
5

// Process OP_WRITE first as we may be able to write some queued buffers and so free memory.
if ((readyOps & SelectionKey.OP_WRITE) != 0) {
    // Call forceFlush which will also take care of clear the OP_WRITE once there is nothing left to write
    ch.unsafe().forceFlush();
}

查看processSelectedKey方法的调用链，我们看到这个方法最终会调用AbstractUnsafe的flush0方法，代码片段如下所示：

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

protected void flush0() {
		if (inFlush0) {
    	return;
    }
    final ChannelOutboundBuffer outboundBuffer = this.outboundBuffer; // @1
    if (outboundBuffer == null || outboundBuffer.isEmpty()) {
    		return;
    }
    inFlush0 = true;
    // Mark all pending write requests as failure if the channel is inactive.
    if (!isActive()) { // @2
        try {
             if (isOpen()) {
                outboundBuffer.failFlushed(FLUSH0_NOT_YET_CONNECTED_EXCEPTION, true);
             } else {
                outboundBuffer.failFlushed(FLUSH0_CLOSED_CHANNEL_EXCEPTION, false);
             }
        } finally {
             inFlush0 = false;
        }
        return;
     }
     try {
         doWrite(outboundBuffer);  // @3
     } catch (Throwable t) {
         if (t instanceof IOException && config().isAutoClose()) {
              close(voidPromise(), t, FLUSH0_CLOSED_CHANNEL_EXCEPTION, false);
         } else {
              outboundBuffer.failFlushed(t, true);
         }
     } finally {
         inFlush0 = false;
     }
}

flush0方法的核心要点主要包括下面三点。

• 获取写缓存队列。如果写缓存队列为空，则跳过本次写事件。每一个通道Channel内部维护一个写缓存区，其他线程调用Channel向网络中写数据时，首先会写入到写缓存区，等到写事件被触发时，再将写缓存区中的数据写入到网络中。
• 如果通道处于未激活状态，需要清理写缓存区，避免数据污染。
• 通过调用 doWrite 方法将写缓存中的数据写入网络通道中。

这里的doWrite方法比较重要，我们重点介绍一下。

doWrite方法主要使用NIO完成数据的写入，具体由NioSocketChannel的doWrite实现，由于这一方法代码较长，我们还是分段来进行讲解。

第一步，如果通道的写缓存区中没有可写数据，需要取消写事件，也就是说，这时候不必关注写事件。具体代码如下：

1
2
3
4
5
6

int size = in.size();
if (size == 0) {
		// All written so clear OP_WRITE
    clearOpWrite();
    break;
}

这背后的逻辑是，如果注册写事件，每次进行事件就绪选择时，只要底层TCP连接的写缓存区不为空，写就会就绪，它会继续通知上层应用程序可以往通道中就绪了。但这种情况下，如果上层应用无数据可写，写事件就绪就变得没有意义了。所以，为了避免出现这种情况，如果没有数据可写，建议直接取消写事件。

第二步，尝试将缓存区数据写入到网络中：

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

switch (nioBufferCnt) {
		case 0:
      	super.doWrite(in);
        return;
    case 1:
        ByteBuffer nioBuffer = nioBuffers[0];
        for (int i = config().getWriteSpinCount() - 1; i >= 0; i --) {
           final int localWrittenBytes = ch.write(nioBuffer);
           if (localWrittenBytes == 0) {
               setOpWrite = true;
                break;
           }
           expectedWrittenBytes -= localWrittenBytes;
           writtenBytes += localWrittenBytes;
           if (expectedWrittenBytes == 0) {
               done = true;
               break;
           }
        }
        break;
     default:
        for (int i = config().getWriteSpinCount() - 1; i >= 0; i --) {
             final long localWrittenBytes = ch.write(nioBuffers, 0, nioBufferCnt);
             if (localWrittenBytes == 0) {
                   setOpWrite = true;
                   break;
             }
             expectedWrittenBytes -= localWrittenBytes;
             writtenBytes += localWrittenBytes;
             if (expectedWrittenBytes == 0) {
                  done = true;
                  break;
             }
         }
        break;
 }

也就是说，这一步要根据缓存区中的数据进行区分写入，各个分支的情况有所不同：

• 如果缓存区nioBufferCnt的个数为0，说明待写入数据为FileRegion（Netty零拷贝实现关键点），需要调用父类NIO相关方法完成数据写入。
• 如果数据是Buffer类型，且只有1个，则直接调用父类的doWrite方法，它的底层逻辑是基于NIO通道写入数据。
• 如果数据是Buffer类型而且有多个，那就要使用NIO Gather机制了，这可以避免数据复制。

写入端的处理逻辑也是差不多的。我们可以通过底层NIO SocketChannel的write方法将数据写入到Socket缓存区，有三种情况需要分别考虑。

• 如果返回值为０，表示Socket底层的缓存区已满，需要暂停写入。具体做法是，注册写事件，等Socket底层写缓存区空闲后再继续写入。
• 如果写缓存区的数据写入到了网络，那就需要取消注册写事件，避免毫无意义的写事件就绪。
• 如果写缓存区中的数据很大，为了避免单个通道对其他通道的影响，默认设置单次写事件最多调用底层 NIO SocketChannel的write方法的次数为16。

第三步，如果底层缓存区已写满，重新注册写事件；如果需要写入的数据太多，则需要创建一个Task放入到IO线程中，待就绪事件处理完毕后继续处理。代码片段如下：

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

if (!done) {
    // Did not write all buffers completely.
    incompleteWrite(setOpWrite);
    break;
}
protected final void incompleteWrite(boolean setOpWrite) {
        // Did not write completely.
        if (setOpWrite) {
            setOpWrite();
        } else {
            // Schedule flush again later so other tasks can be picked up in the meantime
            Runnable flushTask = this.flushTask;
            if (flushTask == null) {
                flushTask = this.flushTask = new Runnable() {
                    @Override
                    public void run() {
                        flush();
                    }
                };
            }
            eventLoop().execute(flushTask); 
        }
    }

注意，这里是处理写入的第二个触发点。将写入请求添加到IO线程的任务列表中，就可以继续执行数据写入。也就是说，并不一定要注册写事件才能进行写入。

Task的触发点在NioEventLoop的run方法，代码片段如下：

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

if (ioRatio == 100) {
   try {
     processSelectedKeys();
   } finally {
     // Ensure we always run tasks.
     runAllTasks();
  }
} else {
  final long ioStartTime = System.nanoTime();
  try {
     processSelectedKeys();
  } finally {
     // Ensure we always run tasks.
     final long ioTime = System.nanoTime() - ioStartTime;
     runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
  }
}

其中，processSelectedKeys就是NIO事件的就绪执行入口。IO线程在执行完事件就绪选择后，会继续执行任务列表中的任务。

在实际开发中，通常是在完成业务逻辑后，往网络中写入数据，调用Channel的writeAndFlush方法。在Channel内部会分别调用write和flush方法。write方法是将数据写入到通道（Channel）对象的写缓存区，而调用flush方法是将通道缓存中的数据写入到网络（Socket 的写缓存区），继而通过网络传输到接收端。

Netty为了避免IO线程与多个业务线程之间的并发问题，业务线程不能直接调用IO线程的数据写入方法，只能是向IO线程提交写入任务，具体代码定义在AbstractChannelHandlerContext的write方法中。

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

private void write(Object msg, boolean flush, ChannelPromise promise) {
        AbstractChannelHandlerContext next = findContextOutbound();
        EventExecutor executor = next.executor();
        if (executor.inEventLoop()) {
            if (flush) {
                next.invokeWriteAndFlush(msg, promise);
            } else {
                next.invokeWrite(msg, promise);
            }
        } else {
            AbstractWriteTask task;
            if (flush) {
                task = WriteAndFlushTask.newInstance(next, msg, promise);
            }  else {
                task = WriteTask.newInstance(next, msg, promise);
            }
            safeExecute(executor, task, promise, msg);
        }
    }

为了方便你深入阅读Netty相关源码，我还给你整理了Netty写入的流程图：

Netty编程实战

好了，关于Netty网络读写的理解就介绍到这里了，但是只有理论是不行的，在这节课的最后，我们来看一个Netty的实战案例。

Netty通常会用在中间件开发、即时通信（IM）、游戏服务器、高性能网关服务器等领域，阿里巴巴的高性能消息中间件RocketMQ就是用Netty进行网络层开发的。

为了方便你学习，我将RocketMQ网络层代码单独抽取成了一个网络编程框架，并上传到了GitHub，你可以拷贝下来跟我一起操作。

在深入RocketMQ网络层具体实践之前，我们先来看一下RocketMQ的网络交互流程：

基于Netty进行网络编程，我们通常需要编写客户端代码、服务端代码和通信协议。

我们先来看Netty客户端编程的通用示例：

这里我们需要注意五个关键点。

1. 需要创建Handler执行线程池，让IO线程只负责网络读写，而且创建线程池一定要使用线程工厂，同时要记得为线程命名。
2. 使用Bootstrap的Group方法指定Work线程组。
3. 通过Option方法设置网络参数。
4. 通过Handler方法创建事件调用链。
5. 将编码、解码、业务逻辑处理相关的事件处理器加入到事件执行链条。

再来看一下客户端如何创建连接，其代码片段如下：

客户端通过调用Bootstrap的connect方法尝试与服务端建立连接，该方法会立即返回一个Future而无须等待连接建立，所以该方法调用结束后并不一定成功创建了连接。但是连接只有在创建成功之后才能被用来发送和读取数据，所以这里我们需要再调用Future对象的awaitUninterruptibly方法等待连接成功建立。

客户端与服务端建立连接后，就可以通过连接向服务端发送请求了：

这里主要有四个实现要点。

1. 为每一个请求创建一个唯一的请求序号。也就是为每一个请求创建一个响应结果Future，并建立RequestId到响应结果的映射Map，这样在收到服务端响应结果时，就可以准确地知道具体是哪一个请求的结果了。这是多线程共同使用单一连接发送请求的核心要点。为了更进一步理解，你可以再看一下这张示意图：

2. 通过调用Channel的writeAndFlush方法，将数据写入到网络中。也就是说，不需要在发送数据之前先注册写事件。然后基于Future模式添加事件监听器，在收到返回结构后，ResponseFuture中的状态会被更新。
3. 同步发送的实现模板，通过调用ResponseFuture获取等待结果，如果使用异步发送模式，就在第三步执行用户定义的回调函数。
4. 处理完一个请求后，删除requestId-ResponseFuture的映射关系。

介绍完客户端编程范例后，接下来我们看一下如何使用Netty编写服务端程序。

首先，创建相关线程组，代码片段如下：

这里分别创建了3个线程组。

• eventLoopGroupBoss线程组，默认使用1个线程，对应Netty线程模型中的主Reactor。
• eventLoopGroupSelector线程组，对应Netty线程模型中的从Reactor组，俗称IO线程池。
• defaultEventExecutorGroup线程组，在Netty中，可以为编解码等事件处理器单独指定一个线程池，从而使IO线程只负责数据的读取与写入。

下一步，使用Netty提供的ServerBootstrap对象创建Netty服务端，示例代码如下所示：

上面的代码基本都是模版代码，少数不同点就是需要自己实现编码器、解码器和业务处理Handler。其中，编码器、解码器其实就是实现通信协议，而ServerHandler就是服务端业务处理器。

再下一步，服务端在指定接口建立监听，等待客户端连接：

ServerBootstrap的bind方法是一个非阻塞方法，调用sync()方法会变成阻塞方法，它需要等待服务端启动完成。

最后一步就是编写服务端业务处理Handler了：

服务端处理器需要接收客户端请求，所以通常需要实现channelRead事件。通常业务Handler是在解码器之后执行，所以业务Handler中channelRead方法接收到的参数已经是通信协议中定义的具体模型，也就是请求对象了。后面就是根据该请求对象中的内容，执行对应的业务逻辑了。业务Handler会在defaultEventExecutorGroup线程组中执行，为了提高解码的性能，避免业务逻辑与IO操作相互影响，通常会将业务执行派发到业务线程池。

总结

好了，这节课就讲到这里。

这节课，我们从一个简单的RPC请求-响应模式说起，串起了网络编程中编码、网络写、网络传输、网络读取、解码、业务逻辑执行等步骤，并引出了网络粘包问题，最终通过制定通信协议解决了粘包问题。

通信协议看似是一个非常高大上的名词，它其实是一种发送端和接收端共同制定的通信格式。我在这里介绍了一种通用的设计方法：Header(请求头) + Body（消息体）的经典设计方法。

接下来，我们还讲解了Netty的线程模型，也是主从多Reactor模型。但我们要知道业务Handler默认是在IO线程池中执行的，我们改变这种行为，让Handler在一个独立的线程池中执行，主要是为了提升IO线程的执行效率。

在讲解Netty读写流程之前，我给你提了下面几个问题。只有真正理解了这些问题，才能算是真正理解了NIO编程。在这里，我也给出我的答案，你可以对照思考一下。

• 如何处理连接半关闭？
  

在调用SocketChannel方法的read方法时，如果其返回值为-1，则表示对端已经关闭了连接，接受端也需要同样关闭连接，释放相关资源。

• 什么时候应该注册读事件？
  

接受端通常在创建好NioSocketChannel后就应该注册读事件。这样才能接受发送端的数据，如果服务端感觉到有压力时，可以暂时取消关注读事件，达到限流的效果。

• 写数据之前一定要先注册写事件吗？
  

写数据之前不需要注册写事件，写事件一般是底层NioSocketChannel的底层缓存区满了，无法再往网络中写入数据时，再注册通道的写事件，等待缓冲区空闲时通知应用程序继续将剩余数据写入到网络中。

在课程的最后，我们以消息中间件RocketMQ是如何使用Netty开发网络通信模块，进行Netty网络编程实战，做到理论与实践相结合。

Netty是一个庞大的体系，如果你想进一步提升高并发编程能力，我建议你体系化地学习一下它，我也非常推荐这本《Netty 源码分析与实战-网络通道篇》，希望可以让你在学习Netty的过程中少走一些弯路。

课后题

学完这节课，你应该已经掌握了NIO的读写处理过程，那我也给你留一道课后题。

请你尝试重构上节课的代码：实现一个简易的RPC Request-Response模型，确保这个模型支持同步请求、异步请求两种请求发送模式。

如果你想要分享你的代码想听听我的意见，可以提交一个 GitHub的push请求或issues，并把对应地址贴到留言里。我们下节课见！

09｜技术选型：如何选择微服务框架和注册中心？

作者: 丁威

你好，我是丁威。

从这节课开始，我们正式进入微服务领域中间件的学习。我们会从微服务框架的诞生背景、服务注册中心的演变历程还有Dubbo微服务框架的实现原理出发，夯实基础。然后，我会结合自己在微服务领域的实践经验，详细介绍Dubbo网关的设计与落地方案，以及蓝绿发布的落地过程。

这节课，我们先从基础学起。

微服务框架的诞生背景

分布式架构体系是伴随着互联网的发展而发展的，它经历了单体应用和分布式应用两个阶段。记得我在2010年入职了一家经营传统行业的公司，公司主要负责政府采购和招投标系统的开发与维护工作，那是我第一次真正见识了庞大的单体应用架构的样子。

当时公司的架构体系是下面这个样子：

所有的业务组件、业务模块都耦合在一个工程里，最终部署的时候会打成一个统一的War包然后部署在一台Web容器中，所有的业务模块都访问同一个数据库。

在传统行业，这种架构的优势也很明显。因为部署结构单一，所以管理非常方便，而且一般情况下，政府采购等行为的流量变化不大，不会像互联网那样，随着平台的搭建造成业务体量的指数型增长。

我们设想一下，如果某一天国家发布政策，想要做一个全国的统一的政府采购平台，假设这家公司中标了，他们会怎么改造系统呢？通常的做法就是对系统进行拆分，单独部署和扩展各个子系统，拆分后的系统架构如下图所示：

由于单个子系统只部署一个节点已经无法满足要求了，所以他们需要部署多个进程，并且需要根据业务的体量进行动态的增加与减少，这样维护调用关系就会变得非常复杂而且容易出错。

在上面这张架构图中，基础资料子系统被其他所有模块调用，如果我们想要增加新的部署节点，或者由于一些机器老化需要更换设备，导致服务对应的IP地址发生变化，这时候应该怎么维护信息呢？

你可能会说这不就是负载均衡吗。我们可以通过Nginx来实现负载均衡，而调用方不需要维护调用者列表。它的架构是下面这样：

没错，通过引入Nginx可以实现负载均衡，并且在节点发生变化时，只需要修改Nginx的配置，不需要去修改调用方的代码。但是一旦部署了新的节点，我们还是需要手动在Nginx中添加路由信息，也就是说，这个操作只能是人工完成的。随着系统的膨胀，路由配置会变得越来越不可维护，容易出错甚至引发严重的故障。

这个问题代表着一系列与微服务相关的共性需求，如服务注册与自动发现机制、高性能RPC调用、服务治理等。

为了解决这些共性需求，很多微服务中间件如雨后春笋般涌现出来，其中要数Dubbo和Spring Cloud最为突出。

如何选择微服务框架？

Dubbo和Spring Cloud是什么？怎么在Dubbo和SpringCloud之间进行选择呢？

Dubbo是阿里巴巴开源的优秀的微服务框架，它开源之后迅速成为了互联网程序员们的首选微服务框架，我认为Dubbo有下面几个核心优势。

• 易用性

微服务框架通常包含服务注册与自动发现、高性能的RPC远程调用、服务治理等众多复杂的功能需求，框架内部非常复杂。但用户操作这种框架却非常简单，不需要太多专业知识，仅仅是通过Dubbo提供的dubbo:service、dubbo:reference、dubbo:registry等几个配置命令就可以轻松构建自己的微服务体系。

而且，这些配置命令拥有众多配置参数（涵盖服务发现、服务治理、性能调优等维度），而且都根据经验提供了默认值，用户几乎不需要对任何参数进行调优，就能保证项目的稳定运行。

• 可扩展机制

Dubbo通过SPI提供了高度灵活的扩展机制，Dubbo内部几乎所有的核心特性都提供了扩展点，Dubbo官方文档中给出的SPI扩展点有下面这些：

• 高性能

Dubbo RPC协议运行在传输层，并基于TCP协议实现了私有协议栈，支持多种序列化协议，包含protocuf、kryo等高性能序列化协议。

Dubbo的易用性、可扩展机制和高性能让它在一段时间内备受拥护，但也许是Dubbo发展得已经非常成熟了，又或者是阿里巴巴在部署其他的战略，Dubbo竟然“断更了”。我们知道持续迭代、持续创新是开源项目的生命源泉，停止更新的Dubbo也就无法继续高歌猛进了。这也给了其他微服务框架更多的生存空间，SpringCloud技术栈就在这个时候崛起了。

Spring Cloud技术栈由各个不同的子项目构成，每一个项目解决微服务架构领域的一个问题，我把SpringCloud和微服务架构相关的技术组件列了个表格：

SpringCloud技术栈和Dubbo都是非常优秀的微服务框架，并且随着互联网分布式架构正式拥抱云原生，Dubbo也顺应云原生发展浪潮，重新开始维护。那这两个框架我们该如何选择呢？

技术选项要考虑框架本身的特性，同时也需要结合公司的技术栈、使用的开发语言等因素综合考虑，这节课我们重点从框架本身这个维度来考量，也会顺便提一提如何结合公司自身的情况去进行选型。

从功能的丰富程度上讲，SpringCloud体系更占优势，但并不是说使用Dubbo来构建微服务体系就无法实现链路监控、服务网关这些功能。Dubbo的设计理念是职责分明，链路跟踪功能完全可以选择业界主流的链路跟踪开源项目，所以从功能维度我也给你列了一张表格，分别对比了用Spring Cloud和Dubbo搭建的微服务架构体系采用的技术栈：

从表格中我们也能看出，在微服务架构必备的注册中心、服务调用、负载均衡、熔断等基础功能上，Dubbo都是内置的，不需要用户关注太多技术细节，而Spring Cloud需要单独进行学习，入门成本偏高。

Dubbo的设计理念是提供对应的扩展点，供用户根据需要自行扩展。而Spring Cloud中各个技术组件都是单独发展的，最终SpringBoot体系将第三方的开源项目进行了整合，省去了用户的整合成本。

从性能的角度，Dubbo要明显优于SpringCloud。

Spring Cloud的RPC调用是基于HTTP协议开发的，它处于网络模型的应用层，而Dubbo的RPC调用的底层是TCP协议，它处于网络模型的传输层。所以说，在底层网络通讯方面，Dubbo就天然地占据了优势。

由于Dubbo是基于TCP编程的，这就比直接使用HTTP进行数据传输具有更大的灵活度。直接基于TCP网络进行编程，对网络通讯中各个环节可以灵活进行定制化开发，例如Dubbo在序列化、反序列化、IO线程、业务线程等方面的设置具有高度配置化，性能的提升非常明显，而Spring Cloud在这方面显得就有些吃力了。阿里、腾讯、美团、拼多多等一线互联网企业的微服务框架都是基于TCP来构建的。

Dubbo、SpringCloud都是主流的微服务，你可以根据实际情况加以选择。不过，结合目前我所处的行业和公司的技术栈，我倾向于采用Dubbo来构建微服务架构体系。

如何选择微服务注册中心？

在这节课的最后，我想结合生产中遇到的一个故障，和你聊聊注册中心的选型问题。

在微服务架构体系相当长的一段发展时间里，ZooKeeper都占领着微服务注册中心的头把交椅，几乎成为注册中心唯一的选择。这是为什么呢？接下来我们就重点解读一下ZooKeeper的CP设计理念。下节课，我们还会对微服务注册中心的设计理念做详细介绍。

ZooKeeper是一个分布式协调组件，符合CAP分布式理论中的CP。

CAP理论指的是，在一个分布式集群中存储同一份数据，无法同时实现C（一致性）、A(可用性)和P(持久性)，只能同时满足其中两个。由于P在数据存储领域是必须要满足的，所以通常需要在C与A之间做权衡。ZooKeeper是保住了一致性和持久性，选择性地牺牲了可用性。

ZooKeeper的数据写入流程如下：

在ZooKeeper集群中，首先会进行Leader选举，根据ZAB协议选举出一个Leader节点用来处理写请求，然后将数据复制给从节点：

• 当集群内超过半数节点写入成功，则返回“数据写入成功”；
• 如果集群内还没有成功选举出Leader，则ZooKeeper集群无法向外提供数据写入与读取服务。

在Leader选举期间，集群是不可用的（牺牲了可用性）。但在正常生产实践过程中，ZooKeeper集群内部选举Leader节点的耗时在毫秒级别，并不会影响使用。然而，一旦遇到异常情况就很难说了。

我在生产过程中就出现了由于ZooKeeper集群内存溢出导致频繁Full GC的情况。当时的情况是，公司内部的Dubbo专用ZooKeeper地址被业务方用做分布式锁，但他们在使用过程中频繁创建节点，加上遇到Bug，节点数据没有及时删除，这就导致占用的内存越来越大，最终频繁Full GC，使得ZooKeeper会话超时，所有注册在ZooKeeper注册中心的服务全部被删除，所有客户端服务调用都出现“No Provider”警告，酿成一场严重的生产级故障。

经过这次故障，我也开始重新审视ZooKeeper和CP模式的合理性。注册中心是微服务体系的大脑，一旦出现问题会带来不可估量的损失，其可用性尤为重要。

也正是因为CP模型存在严重的可用性问题，以AP为设计思想的注册中心开始逐渐涌现出来。AP的核心指导思想是容忍分布式集群中多个节点之间的数据短暂不一致，但最终能达到一致性。EureKa就是典型的基于AP的注册中心。

由于基于AP的注册中心不需要保证强一致性，所以集群内节点的地位通常都是平等的。客户端在同一时间与集群中一个节点保持长连接，当出现错误后，客户端再从注册中心集群中选择另外一个节点，并且客户端可以向集群中任何一个节点写入数据后立即返回“写入成功”，然后让数据异步在集群内部复制，最终实现数据的一致性。EureKa集群的写入流程如下：

由于集群内部节点的地位是平等的，客户端在其中一个节点不可用时，可以快速切换到另外的节点，这样可用性就得到了保障。那么问题来了，节点之间路由信息不一致会带来什么问题呢？这些问题我们可不可以接受？

在回答这个问题之前我们不妨来看看一个注册中心各个节点数据不一致的例子，如下所示：

在这里，由于某种异常，Eureka集群中各个节点存储的数据并不一致，在节点1和2中关于/user/saveUser接口有三个服务提供者，但在节点3中只有两个服务提供者。但无论是三个服务提供者也好，还是两个服务提供者也好，都会造成负载不均衡，如果节点出现类似Full GC的问题，节点无法对外提供服务，这时候客户端会从集群中选择其他节点重试，并不会对系统带来致命影响。

综合来看，服务注册中心这种场景，AP模式显然比CP模式更佳。这也是为什么现在很多原先使用CP模式的注册中心都开始尝试向AP转化，而像Eureka、Nacos这种注册中心基本都同时提供了AP和CP两种工作模式，用户可以按照场景进行选择。

总结

好了，这节课就讲到这里。这节课我们主要从微服务框架诞生背景、微服务框架选型和注册中心框架的演变三个方面介绍了微服务。

微服务框架的基本诉求主要包括：服务注册与自动发现机制、高性能RPC调用和服务治理，它致力于让分布式架构中的服务治理变得简单高效。

我们还分析了市面上两种最主流的微服务研发框架：Dubbo和Spring Cloud，Dubbo具有易用性、灵活的扩展机制和更好的性能，Spring Cloud则具有更加丰富的功能。你可以根据实际情况加以选择，结合目前我所处的行业，公司的技术栈，我倾向于采用Dubbo来构建微服务架构体系。

最后，我还结合自己在实践过程中发生的一起故障，介绍了注册中心从CP向AP架构演进的原因。总的来说，以Eureka和Nacos为代表的注册中心，正在逐渐取代采用CP模式的ZooKeeper，成为注册中心的优先选项。

思考题

最后，我也给你留一道思考题。

我们刚才讲了一个我在生产实践中经历的一次事故。基于Zookeeper搭建的Dubbo服务注册中心，由于ZooKeeper节点的内存使用不当导致频繁触发Full GC，最终导致ZooKeeper会话超时，在注册中心的服务提供者会全部被删除，所有的消费者调用都感知不到服务提供者，进而导致服务调用雪崩。这时候我们应该怎么做呢？难道要重启所有服务提供者，让他们重新注册吗？你有什么快速恢复的方法？

欢迎你在评论区留下自己的看法，我们下节课见！

10｜设计原理：Dubbo核心设计原理剖析

作者: 丁威

你好，我是丁威。

这节课，我们来剖析一下Dubbo中一些重要的设计理念。这些设计理念非常重要，在接下来的11和12讲Dubbo案例中也都会用到，所以希望你能跟上我的节奏，好好吸收这些知识。

微服务架构体系包含的技术要点很多，我们这节课没法覆盖Dubbo的所有设计理念，但我会带着你梳理Dubbo设计理念的整体脉络，把生产实践过程中会频繁用到的底层原理讲透，让你轻松驾驭Dubbo微服务。

我们这节课的主要内容包括服务注册与动态发现、服务调用、网络通信模型、高度灵活的扩展机制和泛化调用五个部分。

服务注册与动态发现

我们首先来看一下Dubbo的服务注册与动态发现机制。

Dubbo的服务注册与发现机制如下图所示：

Dubbo中主要包括四类角色，它们分别是注册中心（Registry）、服务调用者&消费端（Consumer）、服务提供者（Provider）和监控中心（Monitor）。

在实现服务注册与发现时，有三个要点。

1. 服务提供者(Provider)在启动的时候在注册中心(Register)注册服务，注册中心(Registry)会存储服务提供者的相关信息。
2. 服务调用者(Consumer)在启动的时候向注册中心订阅指定服务，注册中心将以某种机制（推或拉）告知消费端服务提供者列表。
3. 当服务提供者数量变化（服务提供者扩容、缩容、宕机等因素）时，注册中心需要以某种方式(推或拉)告知消费端，以便消费端进行正常的负载均衡。

Dubbo官方提供了多种注册中心，我们选择使用最为普遍的ZooKeeper进一步理解注册中心的原理。

我们先来看一下Zookeeper注册中心中的数据存储目录结构。

可以看到，它的目录组织结构为 /dubbo/{ServiceName}，其中，ServiceName表示一个具体的服务，通常用包名+类名表示，在每一个服务名下又会创建四个目录，它们分别是：

• providers，服务提供者列表；
• consumers，消费者列表；
• routers，路由规则列表（一个服务可以设置多个路由规则）；
• configurators，动态配置条目。

要说明的是，在Dubbo中，我们可以在不重启消费者、服务提供者的前提下动态修改服务提供者、服务消费者的配置，配置信息发生变化后会存储在configurators子节点中。此时，服务提供者、消费者会动态监听配置信息的变化，变化一旦发生就使用最新的配置重构服务提供者和服务消费者。

基于Zookeeper注册中心的服务注册与发现有下面三个实现细节。

1. 服务提供者启动时会向注册中心进行注册，具体是在对应服务的providers目录下增加一条记录（临时节点），记录服务提供者的IP、端口等信息。同时服务提供者会监听configurators节点的变化。
2. 服务消费者在启动时会向注册中心订阅服务，具体是在对应服务的consumers目录下增加一条记录（临时节点），记录消费者的IP、端口等信息，同时监听 configurators、routers 目录的变化，所谓的监听就是利用ZooKeeper提供的watch机制。
3. 当有新的服务提供者上线后， providers 目录会增加一条记录，注册中心会将最新的服务提供者列表推送给服务调用方（消费端），这样消费者可以立刻收到通知，知道服务提供者的列表产生了变化。如果一个服务提供者宕机，因为它是临时节点，所以ZooKeeper会把这个节点移除，同样会触发事件，消费端一样能得知最新的服务提供者列表，从而实现路由的动态注册与发现。

服务调用

接下来我们再来看看服务调用。Dubbo的服务调用设计十分优雅，其实现原理图如下：

服务调用重点阐述的是客户端发起一个RPC服务调用时的所有实现细节，它包括服务发现、故障转移、路由转发、负载均衡等方面，是Dubbo实现灰度发布、多环境隔离的理论指导。

刚才，我们已经就服务发现做了详细介绍，接下来我们重点关注负载均衡、路由、故障转移这几个方面。

客户端通过服务发现机制，能动态发现当前存活的服务提供者列表，接下来要考虑的就是如何从服务提供者列表中选择一个服务提供者发起调用，这就是所谓的负载均衡（LoadBalance）。

Dubbo默认提供了随机、加权随机、最少活跃连接、一致性Hash等负载均衡算法。

值得注意的是，Dubbo不仅提供了负载均衡机制，还提供了智能路由机制，这是实现Dubbo灰度发布的重要理论基础。

所谓路由机制，是指设置一定的规则对服务提供者列表进行过滤。负载均衡时，只在经过了路由机制的服务提供者列表中进行选择。为了更好地理解路由机制的工作原理，你可以看看下面这张示意图：

我们为查找用户信息服务设置了一条路由规则，即“查询机构ID为102的查询用户请求信息将被发送到新版本（192.168.3.102）上。具体的做法是，在进行负载均衡之前先执行路由选择，按照路由规则对原始的服务提供者列表进行过滤，从中挑选出符合要求的提供者列表，然后再进行负载均衡。

接下来，客户端就要向服务提供者发起RPC请求调用了。远程服务调用通常涉及到网络等因素，因此并不能保证100%成功，当调用失败时应该采用什么策略呢？

Dubbo提供了下面五种策略：

• failover，失败后选择另外一台服务提供者进行重试，重试次数可配置，通常适合实现幂等服务的场景；
• failfast，快速失败，失败后立即返回错误；
• failsafe，调用失败后打印错误日志，返回成功，通常用于记录审计日志等场景；
• failback，调用失败后，返回成功，但会在后台定时无限次重试，重启后不再重试；
• forking，并发调用，收到第一个响应结果后返回给客户端。通常适合实时性要求比较高的场景。但这一策略浪费服务器资源，通常可以通过forks参数设置并发调用度。

如果将服务调用落到底层，就不得不说说网络通信模型了，这部分包含了很多性能调优手段。

网络通信模型

我们先看看Dubbo的网络通信模型，如下图所示：

Dubbo的网络通信模型主要包括网络通信协议和线程派发机制（Dispatcher）两部分。

网络传输通常需要自定义通信协议，我们常用的协议设计方式是 Header + Body,其中Header 长度固定，包含一个长度字段，用于记录整个协议包的大小。

同时，为了提高传输效率，我们一般会对传输数据也就是Body的内容进行序列化与压缩处理。

Dubbo支持目前支持 java、compactedjava、nativejava、fastjson、fst、hessian2、kryo等序列化协议，生产环境默认为hessian2。

网络通信模型的另一部分是线程派发机制。Dubbo中会默认创建200个线程处理业务，这时候就需要线程派发机制来指导IO线程与业务线程如何分工。

Dubbo提供了下面几种线程派发机制：

• all，所有的请求转发到业务线程池中执行（IO读写、心跳包除外，因为在Dubbo中这两种请求都必须在IO线程中执行，不能通过配置修改）；
• message，只有请求事件在线程池中执行，其他请求在IO线程上执行；
• connection ，求事件在线程池中执行，连接和断开连接的事件排队执行（含一个线程的线程池）；
• direct，所有请求直接在IO线程中执行。

为什么线程派发机制有这么多种策略呢？其实这主要是考虑到线程切换带来的开销问题。也就是说，我们希望通过多种策略让线程切换带来的开销小于多线程处理带来的提升。

我举个例子，Dubbo中的心跳包都必须在IO线程中执行。在处理心跳包时，我们只需直接返回PONG包（OK）就可以了，逻辑非常简单，处理速度也很快。如果将心跳包转换到业务线程池，性能不升反降，因为切换线程会带来额外的性能损耗，得不偿失。

网络编程中需要遵循一条最佳实践：IO线程中不能有阻塞操作，通常将阻塞操作转发到业务线程池异步执行。

与网络通信协议相关的参数定义在dubbo:protocol，关键的设置属性如下。

• threads，业务线程池线程个数，默认为200。
• queues，业务线程池队列长度，默认为0，表示不支持排队，如果线程池满，则直接拒绝。该参数与threads配合使用，主要是对服务端进行限流，一旦超过其处理能力，就拒绝请求，快速失败，引导客户端重试。
• iothreads：默认为CPU核数再加一，用于处理网络读写。在生产实践中，通常的瓶颈在于业务线程池，如果业务线程无明显瓶颈（jstack日志查询到业务线程基本没怎么干活），但吞吐量已经无法继续提升了，可以考虑调整iothreads，增加IO线程数量，提高IO读写并发度。该值建议保持在“2*CPU核数”以下。
• serialization：序列化协议，新版本支持protobuf等高性能序列化机制。
• dispatcher：线程派发机制，默认为all。

高度灵活的扩展机制

Dubbo出现之后迅速成为微服务领域最受欢迎的框架，除操作简单这个原因外，还有扩展机制的功劳。Dubbo高度灵活的扩展机制堪称“王者级别的设计”。

Dubbo的扩展设计主要是基于SPI设计理念，我们来看下具体的实现方案。

Dubbo所有的底层能力都通过接口来定义。用户在扩展时只需要实现对应的接口，定义一个统一的扩展目录（META-INF.dubbo.internal）存放所有的扩展定义即可。要注意的是，目录下的文件名是需要扩展的接口的全名，像下图这样：

在初次使用对应接口实例时，可以扫描扩展目录中的文件，并根据文件中存储的key-value初始化具体的实例。

我们以RPC模块为例看一下Dubbo强悍的扩展能力。众所周知，目前gRPC协议以优异的性能表现正在逐步成为RPC领域的王者，很多人误以为gRPC是来革Dubbo的“命”的。其实不然，我们可以认为Dubbo是微服务体系的完整解决方案，而RPC只是微服务体系中的重要一环，Dubbo完全可以吸收gRPC，让gRPC成为Dubbo的远程调用方式。

具体的做法只需要在dubbo-rpc模块中添加一个dubbo-rpc-grpc模块，然后使用gRPC实现org.apache.dubbo.rpc.protocol接口，并将其配置在扩展目录中：

面对gRPC这么强大的功能扩展机制，绝大部分人应该和我一样，都是作为中间件的应用人员，不需要使用模块级别的扩展机制。我们通常只是结合应用场景来进行功能扩展。

Dubbo在业务功能级别的扩展可以通过Filter机制来实现。Filter的工作机制如下：

这里，过滤器链的执行时机是在服务消费者发起远程RPC请求之前。最先执行的是消费端的过滤器链，每一个过滤器可以设置执行顺序。服务端在解码之后、执行业务逻辑之前，也会首先调用过滤器链。

在专栏的最后一讲，我还会通过一个全链路压测方案讲解如何利用Filter机制来解决实际问题。

泛化调用

在这节课的最后，我们再来介绍一下Dubbo的泛化调用机制，它也是实现Dubbo网关的理论基础。

我们在开发Dubbo应用时通常会包含API、Consumer、Provider三个子模块。

其中API模块通常定义统一的服务接口，而Consumer、Provider模块都需要显示依赖API模块。这种设计理念虽然将Provider与Consumer进行了解耦合，但对API模块形成了强依赖，如果API模块发生改变，Provider和Consumer必须同时改变。也就是说，一旦API模块发生变化，服务调用方、服务消费方都需要重新部署，这对应用发布来说非常不友好。特别是在网关领域，几乎是不可接受的，如下图所示：

公司的微服务在不停地演进，如果网关需要跟着API模块不停地发布新版本，网关的可用性和稳定性都将受到极大挑战。怎么解决这个问题呢？

这就要说到Dubbo的机制了。泛化调用具体实现原理如下：

当服务消费端发生调用时，我们使用Map来存储一个具体的请求参数对象，然后传输到服务提供方。由于服务提供方引入了模型相关的Jar，服务提供方在执行业务方法之前，需要将Map转化成具体的模型对象，然后再执行业务逻辑。

Dubbo的泛化调用在服务提供方的转化是通过Filter机制统一处理的，服务端并不需要关注消费方采取何种方式进行调用。

通过泛化调用机制，客户端不再需要依赖服务端的Jar包，服务端可以不断地演变，而不会影响客户端已有服务的运行。

总结

好了，这节课就讲到这里。我们这节课主要介绍了Dubbo的服务注册与发现、服务调用、网络通信模型、扩展机制还有泛化调用等核心工作机制，了解这些内容可以指导我们更好实践微服务。

另外，Dubbo框架算是阿里巴巴开源的所有框架中文档最为齐全的框架了，非常值得我们深入学习与研究。如果你想要进一步掌握Dubbo，建议你看看Dubbo官方文档。

课后题

1. 我们将在下节课和你一起聊聊Dubbo的网关设计方案，其中泛化调用是其理论设计基础，所以我们的第一道课后题就是，请你试着先编写一个Dubbo泛化调用的示例。

提示一下，Dubbo提供了dubbo-demo模块，你可以在官方提供的示例中进行泛化调用编写，节省搭建基础项目的时间。

2. 请你尝试通过dubbo-admin运维管理工具动态修改参数，看看它是否可以动态生效。你知道它背后是如何实现的么？

欢迎你在留言区与我交流讨论，我们下节课再见！

11｜案例：如何基于Dubbo进行网关设计？

作者: 丁威

你好，我是丁威。

这节课我们通过一个真实的业务场景来看看Dubbo网关（开放平台）的设计要领。

设计背景

要设计一个网关，我们首先要知道它的设计背景。

2017年，我从传统行业脱身，正式进入物流行业。说来也非常巧，我当时加入的是公司的网关项目组，主要解决泛化调用与协议转换代码的开发问题。刚进公司不久，网关项目组就遇到了技术难题。快递物流行业的业务量可以比肩互联网，从那时候开始，我的传统技术思维开始向互联网技术思维转变。

当时网关项目组的核心任务就是确保能够快速接入各个电商平台。我来简单说明一下具体的场景。

解释一下上面这个图。

物流公司内部已经基于Dubbo构建了订单中心微服务域，其中创建订单接口的定义如下：​

外部电商平台众多，每一家电商平台内部都有自己的标准，并不会遵循统一的标准。例如在淘宝中，当用户购买商品后，淘宝内部会定义一个统一的订单外派接口。它的请求包可能是这样的：

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

{
  "seller_id":189,
  "buyer":"dingwei",
  "order":[
    {
      "goods_name":"华为笔记本",
      "num":1,
      "price":500000
    },
    {
      "goods_name":"华为手表",
      "num":1,
      "price":200000
    }
  ]
}

但拼多多内部定义的订单外派接口，它的请求包可能是下面这样的：

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

<order>
 &nbsp;<seller_uid>189</seller_uid>
 &nbsp;<buyer_uid>dingwei</buyer_uid>
 &nbsp;<order_items>
 &nbsp; &nbsp;<order_item>
 &nbsp; &nbsp; &nbsp;<goods_name>华为笔记本</goods_name>
 &nbsp; &nbsp; &nbsp;<num>1</num>
 &nbsp; &nbsp; &nbsp;<price>500000</price>
 &nbsp; &nbsp;</order_item>
 &nbsp; &nbsp;<order_item>
 &nbsp; &nbsp; &nbsp;<goods_name>华为手表</goods_name>
 &nbsp; &nbsp; &nbsp;<num>1</num>
 &nbsp; &nbsp; &nbsp;<price>200000</price>
 &nbsp; &nbsp;</order_item>
 &nbsp;</order_items>
</order>

当电商的快递件占据快递公司总业务量的大半时，电商平台的话语权是高于快递公司的。也就是说，电商平台不管下游对接哪家物流公司，都会下发自己公司内部定义的订单派发接口，适配工作需要由物流公司自己来承担。

那站在物流公司的角度，应该怎么做呢？总不能每接入一个电商平台就为它们开发一套下单服务吧？那样的话，随着越来越多的电商平台接入，系统的复杂度会越来越高，可维护性将越来越差。

设计方案

正是在这样的背景下，网关平台被立项开发出来了。这个网关平台是怎么设计的呢？在设计的过程中需要解决哪些常见的问题？

我认为，网关的设计至少需要包括三个方面，分别是签名验证、服务配置和限流。

先说签名验证。保证请求的安全是系统设计需要优先考虑的。业界有一种非常经典的通信安全校验机制：验证签名。

这种机制的做法是，客户端与服务端会首先采用HTTPS进行通信，确保传输过程的私密性。

客户端在发送请求时，先将请求参数按参数名称进行排序，然后按顺序拼接成字符串，格式为key1=a & key2=b。接下来，客户端使用一个约定的密钥对拼接出来的参数字符串进行签名，生成签名字符串（我们用sign表示签名字符串）并追加到URL。通常，还会在URL中追加一个发送时间戳（时间戳不参与签名验证）。

服务端在接收到客户端的请求后，先从请求中解析出所有的参数，同样按照参数名对参数进行排序，然后使用同样的密钥对参数进行签名。得到的签名字符串需要与客户端计算的签名字符串进行对比，如果两者不同，则请求无效。与此同时，通常我们还需要将服务端当前的时间戳与客户端时间戳进行对比，如果相差超过一定的时间，同样认为请求无效，这个操作主要是为了避免使用同一个连接对网络进行连续攻击。

这整个过程里有一个非常重要的点，就是密钥自始至终并没有在网络上进行过传播，它的安全性可以得到十足的保证。签名验证的流程大概可以用下面这张图表示：

如果要对验证签名进行产品化设计，我们通常需要：

1. 为不同的接入端（电商平台）创建不同的密钥，并通过安全的方式告知他们；
2. 为不同的接入端（电商平台）配置签名算法。

在确保能够安全通信后，接下来就是网关设计最核心的部分了：服务接口配置化。它主要包括两个要点：微服务调用协议（Dubbo服务描述）和接口定义与参数映射。

我们先来看一下微服务调用协议的配置，设计的原型界面如下图所示：

将所有的微服务（细化到方法级名称）维护到网关系统中，网关应用就可以使用Dubbo提供的编程API，根据这些元信息动态构建一个个消费者（服务调用者），进而通过创建的服务调用客户端发起RPC远程调用，最终实现网关应用的Dubbo服务调用。

基于这些元信息构建消费者对象的关键代码如下：

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

public static GenericService getInvoker(String serviceInterface, String version, List<String> methods, int retry, String registryAddr ) {
 &nbsp; &nbsp; &nbsp; &nbsp;ReferenceConfig referenceConfig = new ReferenceConfig();
 &nbsp; &nbsp; &nbsp; &nbsp;// 关于消费者通用参数，可以从配置文件中获取，本示例取消
 &nbsp; &nbsp; &nbsp; &nbsp;ConsumerConfig consumerConfig = new ConsumerConfig();
 &nbsp; &nbsp; &nbsp; &nbsp;consumerConfig.setTimeout(3000);
 &nbsp; &nbsp; &nbsp; &nbsp;consumerConfig.setRetries(2);
 &nbsp; &nbsp; &nbsp; &nbsp;referenceConfig.setConsumer(consumerConfig);
 &nbsp; &nbsp; &nbsp; &nbsp;//应用程序名称
 &nbsp; &nbsp; &nbsp; &nbsp;ApplicationConfig applicationConfig = new ApplicationConfig();
 &nbsp; &nbsp; &nbsp; &nbsp;applicationConfig.setName("GateWay");
 &nbsp; &nbsp; &nbsp; &nbsp;referenceConfig.setApplication(applicationConfig);
 &nbsp; &nbsp; &nbsp; &nbsp;// 注册中心
 &nbsp; &nbsp; &nbsp; &nbsp;RegistryConfig registry = new RegistryConfig();
 &nbsp; &nbsp; &nbsp; &nbsp;registry.setAddress(registryAddr);
 &nbsp; &nbsp; &nbsp; &nbsp;registry.setProtocol("zookeeper");
 &nbsp; &nbsp; &nbsp; &nbsp;referenceConfig.setRegistry(registry);
 &nbsp; &nbsp; &nbsp; &nbsp;// 设置服务接口名称
 &nbsp; &nbsp; &nbsp; &nbsp;referenceConfig.setInterface(serviceInterface);
 &nbsp; &nbsp; &nbsp; &nbsp;// 设置服务版本
 &nbsp; &nbsp; &nbsp; &nbsp;referenceConfig.setVersion(version);
 &nbsp; &nbsp; &nbsp; &nbsp;referenceConfig.setMethods(new ArrayList<MethodConfig>());
 &nbsp; &nbsp; &nbsp; &nbsp;for(String method : methods) {
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;MethodConfig methodConfig = new MethodConfig();
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;methodConfig.setName(method);
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;referenceConfig.getMethods().add(methodConfig);
 &nbsp; &nbsp; &nbsp;  }
 &nbsp; &nbsp; &nbsp; &nbsp;referenceConfig.setGeneric("true");// 开启dubbo的泛化调用
 &nbsp; &nbsp; &nbsp; &nbsp;return (GenericService) referenceConfig.get();
 &nbsp;  }

通过getInvoker方法发起调用远程RPC服务，这样，网关应用就成为了对应服务的消费者。

因为网关应用引入服务规约（API包）不太现实，所以这里使用的是泛化调用，这样方便网关应用不受约束地构建消费者对象。

值得注意的是，ReferenceConfig实例很重，它封装了与注册中心的连接以及所有服务提供者的连接，需要被缓存起来。因此，在真实的生产实践中，我们需要将ReferenceConfig对象存储到缓存中。否则，重复生成的ReferenceConfig可能造成性能问题并伴随着内存和连接泄漏。

除了ReferenceConfig，其实getInvoker生成对象也可以进行缓存，缓存的key通常为接口名称、版本和注册中心。

那如果配置信息动态发生了变化，例如需要添加新的服务，这时候网关应用如何做到动态感知呢？我们通常可以用基于MQ的方式来解决这个问题。具体的解决方案如下：

也就是说，用户如果在网关运营平台上修改原有服务协议（Dubbo服务）或者添加新的服务协议，变动后的协议会首先存储到数据库中，然后运营平台发送一条消息到MQ，紧接着Gateway的后台进程以广播模式进行订阅。这样，所有后台网关进程都可以感知。

如果是对已有服务协议进行修改，在具体实践时有一个小细节请你一定注意。我们先看看这段代码：

1
2
3
4
5
6

Map<String /* 缓存key */,GenericService> invokerCache;
GenericService newInvoker = getInvoker(...);//参数省略
GenericService oldInvoker = invokerCache.get(key);
invokerCache.put(newInvoker);//先缓存新的invoker
// 然后再销毁旧的invoker对象
oldInvoker.destory();

如果已经存在对应的Invoker对象，为了不影响现有调用，应该先用新的Invoker对象去更新缓存，然后再销毁旧的Invoker对象。

上面的方法解决了网关调用公司内部的Dubbo微服务问题，但还有另外一个非常重要的问题，怎么配置服务接口相关参数呢？

联系这节课前面的场景，我们需要在页面上配置公司内部Dubbo服务与外部电商的接口映射。

为此，我们专门建立了一条参数映射协议：

参数映射设计的说明如下。

• 请求类型：主要分为请求参数与响应参数；
• 字段名称：Dubbo服务对应的字段名称；
• 字段类型：Dubbo服务对应字段的属性；
• 字段所属类：Dubbo服务对应字段所属类型；
• 节点名称：外部请求接口对应的字段名称；
• 显示顺序：排序字段。

由于网关采取了泛化调用，在编写转换代码时，主要是遍历传入的参数，根据每一个字段查询对应的转换规则，然后转换为Map，返回值则刚好相反，是将Map转换为XML或者JSON。

在真正请求调用时，根据映射规则构建出请求参数Map后，通过Dubbo的泛化调用执行真正的调用：

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

GenericService genericService = (GenericService) invokeBean;
Map invokerPams;//省略转换过程
// 参数类型数组
String[] paramTypes = new String[1];
paramTypes[0]="java.util.Map";
// 参数值数组
Object[] paramValues = new Object[1];
​
invokerPams.put("class", "net.codingw.oms.vo.OrderItemVo");
paramValues[0] = invokerPams;
//由于我们已经转化为java.util.Map，并且Map中，需要有一个key为class的，表示服务端需要转化的类型，这个从协议转换器中获取
Object result = genericService.$invoke(this.getInvokeMethod(), paramTypes, paramValues);

这样，网关就具备了高扩展性和稳定性，可以非常灵活地支撑业务的扩展，为不同的电商平台配置不同的参数转换，从而在内部只需要开发一套接口就可以非常灵活地支撑业务的扩展，基本做到网关代码零修改。

总结

这节课，我通过一个真实的场景，详细介绍了网关设计的需求背景，然后针对网关设计的痛点给出了设计方案。通过对这个方案中关键代码的解读，你应该能够更加深刻地理解Dubbo泛化调用背后的逻辑，真正做到理论与实际相结合。

值得注意的是，我们这节课提到的转换协议也是一绝，它使用中括号来定义多层嵌套结构，使得该协议具有普适性。

课后题

检测对知识的掌握程度最好的方式是自己写出来。所以，我建议你将我们这节课所讲的方案落到实处，尝试自己实现一个demo级的网关设计。

如果你想听听我的意见，可以提交一个 GitHub的push请求或issues，并把对应地址贴到留言里。我们下节课见！

12｜案例：如何实现蓝绿发布？

作者: 丁威

你好，我是丁威。

前面，我们讲了服务的注册与发现机制，它是微服务体系的基石，这节课，我想聊聊微服务的另外一个重要课题：服务治理。

随着微服务应用的不断增加，各个微服务之间的依赖关系也变得比较复杂，各个微服务的更新、升级部署给整个服务域的稳定性带来很大挑战。怎么以不停机的方式部署升级微服务呢？

这就是我们这节课的任务，我们来看看如何在生产环境用蓝绿发布来满足不停机升级的要求。

设计背景

在进行技术方案的设计之前，我们先来了解一下生产环境的基本部署情况，如下图所示：

用户在面向用户端（下文通称C端）下单后，C端订单系统需要远程调用订单域中的“创建订单“接口。同时，订单域、运单域相关服务都需要调用基础服务域，进行基础数据的查询服务。

从这里也可以看出，基础服务的稳定运行对整个微服务体系至关重要。那如何确保基础服务域不受版本的影响，始终能够提供稳定可控的服务呢？

设计方案

我们公司为了解决这个问题实现了蓝绿发布。那什么是蓝绿发布呢？

蓝绿发布指的是在蓝、绿两套环境中分别运行项目的两个版本的代码。但是在进行版本发布时只更新其中一个环境，这样方便另一个环境快速回滚。

接下来我们看一下蓝绿发布的基本流程。

如果系统采取蓝绿发布，在下一个版本（base-service v1.2.0）发布之前，会这样部署架构：

当前订单域调用流量进入基础服务域 GREEN环境。团队计划在12:00发布新版本（base-service v1.2.0），这时我们通常会执行下面几个操作。

• 将新版本1.2.0全部发布在BLUE环境上。因为此时BLUE环境没有任何流量，对运行中的系统无任何影响。
• 在请求入口对流量进行切分。通常可以按照百分比分配流量，待系统运行良好后，再逐步将流量全部切换到新版本。
• 如果发现新版本存在严重问题，可以将流量全部切换到原来的环境，实现版本快速回滚。

这个过程可以用下面这张图表示：

这个思路听起来很简单，但是怎么实现呢？

这就不得不提到上节课专门提到的路由选择（Router）了，它是Dubbo服务调用中非常重要的一步。路由选择的核心思想是在客户端进行负载均衡之前，通过一定的过滤规则，只在服务提供者列表中选择符合条件的提供者。

我们再看上面的实例图，从订单域消费者的视角，服务提供者列表大概是下面这个样子：

然后呢，我们按照比例对入口流量进行分流。例如，80%的请求颜色为BLUE，20%的请求颜色为GREEN。那些颜色为BLUE的请求，在真正执行RPC服务调用时，只从服务提供者列表中选择“color=BLUE”的服务提供者。同样，颜色为GREEN的请求只选择“color=GREEN”的服务提供者，这就实现了流量切分。

具体的操作是，在Dubbo中为这个场景引入Tag路由机制。

首先，服务提供者在启动时需要通过“-Dubbo.provider.tag”系统参数来设置服务提供者所属的标签。

例如，在192.168.3.100和192.168.3.101这两台机器上启动base-service程序时，需要添加“-Dubbo.provider.tag=BLUE”系统参数；而在192.168.4.100和192.168.4.101这两台机器上启动base-service程序时，则要添加“-Dubbo.provider.tag=GREEN”系统参数，通过这个操作完成对服务提供者的打标。服务提供者启动后，生成的服务提供者URL连接如下所示：

1

dubbo://192.168.3.100:20880/net.codingw.demo.BaseUser?dubbo.tag=BLUE

下一步，在服务入口对流量进行染色，从而实现流量切分。

蓝绿发布的流量通常是在流量入口处进行染色的。例如，我们可以使用随机加权来实现流量切分算法，用它对流量进行染色，具体示范代码如下：

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

public static String selectColor(String[] colorArr, int[] weightArr) {
        int length = colorArr.length;
        boolean sameWeight = true;
        int totalWeight = 0;
 &nbsp; &nbsp; &nbsp; &nbsp;for (int i = 0; i < length; i++) {
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;int weight = weightArr[i];
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;totalWeight += weight;
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;if (sameWeight && totalWeight != weight * (i + 1)) {
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;sameWeight = false;
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;  }
 &nbsp; &nbsp; &nbsp;  }
 &nbsp; &nbsp; &nbsp; &nbsp;if (totalWeight > 0 && !sameWeight) {
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;int offset = ThreadLocalRandom.current().nextInt(totalWeight);
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;System.out.println("offset:" + offset);
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;for (int i = 0; i < length; i++) {
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;if (offset < weightArr[i]) {
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;return colorArr[i];
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;  }
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;  }
 &nbsp; &nbsp; &nbsp;  }
 &nbsp; &nbsp; &nbsp; &nbsp;return colorArr[ThreadLocalRandom.current().nextInt(length)];
 &nbsp;  }
​
 &nbsp; &nbsp;//测试代码
 &nbsp; &nbsp;public static void main(String[] args) {
 &nbsp; &nbsp; &nbsp; &nbsp;String[] colorArr = new String[]{"GREEN","BLUE"};
 &nbsp; &nbsp; &nbsp; &nbsp;int[] weightArr = new int[] {100,50};
 &nbsp; &nbsp; &nbsp; &nbsp;for(int i = 0; i < 20; i ++) {
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;System.out.println(selectColor(colorArr, weightArr));
 &nbsp; &nbsp; &nbsp;  }
 &nbsp;  }

根据流量切分算法计算得到流量标识后，怎么在消费端跟进流量标识从而进行路由选择呢？我们通常会将染色标记放在ThreadLocal中，然后再编写Filter，获取或者传递路由标签。

但这个只是一个流量的切分算法，那如何动态设置蓝绿的比例或者说权重呢？其实，我们可以为发布系统提供一个设置权重的页面，用户设置完权重后写入到配置中心(ZooKeeper、Apollo)，然后应用程序动态感知到变化，利用最新的权重进行流量切分。

通过流量切分算法计算出一个请求的流量标识后，通常会存储在ThreadLocal中，实现代码如下：

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

public class ThreadLocalContext {
 &nbsp; &nbsp;private static final ThreadLocal<String> tagContext &nbsp;= new ThreadLocal<>();
​
 &nbsp; &nbsp;public static void setTag(String tag) {
 &nbsp; &nbsp; &nbsp; &nbsp;tagContext.set(tag);
 &nbsp;  }
​
 &nbsp; &nbsp;public static String getTag() {
 &nbsp; &nbsp; &nbsp; &nbsp;return tagContext.get();
 &nbsp;  }
​
 &nbsp; &nbsp;public static void resetTag() {
 &nbsp; &nbsp; &nbsp; &nbsp;tagContext.remove();
 &nbsp;  }
}
​
​
//在整个请求的入口
String color = selectColor(colorArr, weightArr);
try {
 &nbsp;ThreadLocalContext.setTag(color);
 &nbsp;
 &nbsp;//执行第一个远程调用
 &nbsp;invokeRpc1();
 &nbsp;
 &nbsp;//执行另外一个远程调用
 &nbsp;invokeRpc2();
 &nbsp;
} finally {
 &nbsp;ThreadLocalContext.reset();
}

将请求的流量标识存储到线程本地变量之后，还需要将流量标识附加到RPC请求调用中，这样才能触发正确的路由选择，具体代码示例如下：

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

import org.apache.commons.lang3.StringUtils;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.rpc.*;
import org.apache.dubbo.rpc.cluster.router.tag.TagRouter;
​
import static org.apache.dubbo.common.constants.CommonConstants.CONSUMER;
import static org.apache.dubbo.rpc.Constants.ACCESS_LOG_KEY;
​
@Activate(group = CONSUMER, value = "tagConsumerFilter")
public class TagConsumerContextFilter implements Filter {
​
    private static final Logger logger = LoggerFactory.getLogger(TagConsumerContextFilter.class);
​
    @Override
 &nbsp; &nbsp;public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
 &nbsp; &nbsp; &nbsp; &nbsp;try {
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;String tag = ThreadLocalContext.getTag();
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;if(StringUtils.isNotEmpty(tag)) {
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;invocation.setAttachment(TagRouter.NAME, tag );
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;  }
 &nbsp; &nbsp; &nbsp;  } catch (Throwable t) {
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;logger.warn("Exception in TagConsumerContextFilter of service(" + invoker + " -> " + invocation + ")", t);
 &nbsp; &nbsp; &nbsp;  }
​
 &nbsp; &nbsp; &nbsp; &nbsp;// 调用链传递
 &nbsp; &nbsp; &nbsp; &nbsp;return invoker.invoke(invocation);
 &nbsp;  }
}

这样在RPC调用的过程中，服务调用者就能根据本地线程变量中存储的流量标记，选择不同机房的服务提供者，从而实现蓝绿发布了。

同时，在实际生产环境中，一个调用链条中往往会存在多个RPC调用，那第一个RPC中的路由标签能自动传递到第二个RPC调用吗？

答案是不可以，我们需要再写一个服务端生效的Filter，示例代码如下：

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

import org.apache.commons.lang3.StringUtils;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.rpc.*;
import org.apache.dubbo.rpc.cluster.router.tag.TagRouter;
​
import static org.apache.dubbo.common.constants.CommonConstants.PROVIDER;
​
@Activate(group = PROVIDER, value = "tagProviderFilter")
public class TagProviderContextFilter implements Filter {
​
    private static final Logger logger = LoggerFactory.getLogger(TagProviderContextFilter.class);
​
    @Override
 &nbsp; &nbsp;public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
 &nbsp; &nbsp; &nbsp; &nbsp;try {
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;String tag = invocation.getAttachment(TagRouter.NAME);
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;if(StringUtils.isNotEmpty(tag)) {
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;ThreadLocalContext.setTag(tag);
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;  }
 &nbsp; &nbsp; &nbsp;  } catch (Throwable t) {
 &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;logger.warn("Exception in TagProviderContextFilter of service(" + invoker + " -> " + invocation + ")", t);
 &nbsp; &nbsp; &nbsp;  }
 &nbsp; &nbsp; &nbsp; &nbsp;// 调用链传递
 &nbsp; &nbsp; &nbsp; &nbsp;return invoker.invoke(invocation);
 &nbsp;  }
}

也就是将调用链中的tag存储到服务端的线程本地上下文环境中，当服务端调用其他服务时，可以继续将tag传递到下一个RPC调用链中。

这样，我们的蓝绿发布就基本完成了。但这里还有一个问题。规模较大的公司的生产环境往往会运行很多微服务，我们无法将蓝绿机制一下引入到所有微服务当中，必然会存在一部分应用使用蓝绿发布，但其他应用没有使用蓝绿的情况。怎么做到兼容呢？

比方说，我们公司目前核心业务域的蓝绿部署情况如下：

这里，订单域接入了蓝绿发布；C端应用需要调用订单域相关接口，因此也接入了蓝绿发布；但运单中心并未接入蓝绿发布。这时候，运单中心能调用订单域的服务吗？

要回答这个问题，我们要先看看Dubbo官方的降级策略。

• 如果消费者侧设置了标签，那么如果集群中没有对应标签的服务提供者，默认可以选择不带任何标签的服务提供者进行服务调用。该行为可以通过设置request.tag.force=true来禁止，这就是说如果request.tag.force为true，一旦没有对应标签的服务提供者，就会跑出“No Provider”异常。
• 如果消费者侧没有设置标签，那就只能向集群中没有设置标签的服务提供者发起请求，如果不存在没有标签的服务提供者，则报“No Provider”异常。

回到上面的问题，运单中心由于未接入蓝绿发布，所以不带任何标签，它无法调用订单域的服务。为了解决这个问题，订单域还需要部署一些不带标签的服务。订单域最终的部署大概如下图所示：

也就是说，订单域为了兼容那些还没接入蓝绿发布的应用需要部署3套环境，一套为不设置标签的服务提供者，一套为蓝颜色的服务提供者，另一套为绿颜色的服务提供者。

蓝绿发布实践就介绍到这里了，在这节课的最后，我们再来学习一下蓝绿发布底层依托的原理。

实现原理

先来看一下Dubbo服务调用的基本时序图：

我建议你按照这张时序图跟踪一下源码，更加详细地了解Dubbo服务调用的核心流程与实现关键点，我在这里总结了几个要点：

• Dubbo的服务调用支持容错，对应的抽象类为AbstractClusterInvoker，它封装了服务调用的基本流程。Dubbo内置了failover、failfast、failsafe、failback、forking等失败容错策略，每一个策略对应AbstractClusterInvoker的一个实现；
• 在调用AbstractClusterInvoker服务的时候，首先需要获取所有的服务提供者列表，这个过程我们称之为服务动态发现（具体实现类为DynamicDirectory）。在获取路由信息之前，需要调用RouterChain的route方法，执行路由选择策略，筛选出服务动态发现的服务提供者列表。我们这一课的重点，标签路由的具体实现类TagRouter就是在这里发挥作用的。

我们也详细拆解一下TagRouter的route方法。因为这个方法的实现代码比较多，我们还是分步讲解。

第一步，执行静态路由过滤机制，代码如下：

1
2
3
4

final TagRouterRule tagRouterRuleCopy = tagRouterRule;
if (tagRouterRuleCopy == null || !tagRouterRuleCopy.isValid() || !tagRouterRuleCopy.isEnabled()) {
    return filterUsingStaticTag(invokers, url, invocation);
}

如果路由规则为空，则根据tag进行过滤。我们顺便也看一下基于tag的静态过滤机制是如何实现的：

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