任务一：分布式系统

顺利完成第二阶段的学习，正式进入第三阶段模块一。
分布式系统概念
分布式系统是一个硬件或软件组件分布在不同的网络计算机上，彼此之间仅仅通过消息传递进行通信和协调的系统。
一个业务拆分成多个子业务，分布在不同的服务器节点，共同构成的系统称为分布式系统，同一个分布式系统中的服务器节点在空间部署上是可以随意分布的，这些服务器可能放在不同的机柜中，也可能在不同的机房中，甚至分布在不同的城市。

1. 分布式系统的发展

阿里巴巴发起的"去 IOE"运动 (IOE 指的是 IBM 小型机、Oracle 数据库、EMC 的高端存储)。阿里巴巴2009 年“去IOE”战略技术总监透露，截止到 2013 年 5 月 17 日阿里巴巴最后一台 IBM 小型机在支付宝下线。
为什么要去IOE
1.升级单机处理能力的性价比越来越低
2.单机处理能力存在瓶颈
3.稳定性和可用性这两个指标很难达到

2. 分布式架构的演变

3. 分布式系统面临的问题

1）通信异常
网络本身的不可靠性，因此每次网络通信都会伴随着网络不可用的风险（光纤、路由、DNS等硬件设备或系统的不可用），都会导致最终分布式系统无法顺利进行一次网络通信，另外，即使分布式系统各节点之间的网络通信能够正常执行，其延时也会大于单机操作，存在巨大的延时差别，也会影响消息的收发过程，因此消息丢失和消息延迟变的非常普遍。
2）网络分区
网络之间出现了网络不连通，但各个子网络的内部网络是正常的，从而导致整个系统的网络环境被切分成了若干个孤立的区域，分布式系统就会出现局部小集群，在极端情况下，这些小集群会独立完成原本需要整个分布式系统才能完成的功能，包括数据的事务处理，这就对分布式一致性提出非常大的挑战。
3）节点故障
节点故障是分布式系统下另一个比较常见的问题，指的是组成分布式系统的服务器节点出现的宕机或"僵死"现象，根据经验来说，每个节点都有可能出现故障，并且经常发生.
4）三态
分布式系统每一次请求与响应存在特有的“三态”概念，即成功、失败和超时。分布式系统中，由于网络是不可靠的，虽然绝大部分情况下，网络通信能够接收到成功或失败的响应，但当网络出现异常的情况下，就会出现超时现象，通常有以下两种情况：

1. 由于网络原因，该请求并没有被成功的发送到接收方，而是在发送过程就发生了丢失现象。
2. 该请求成功的被接收方接收后，并进行了处理，但在响应反馈给发送方过程中，发生了消息丢失现象。

4. 分布式理论：一致性

1）什么是分布式一致性
分布式数据一致性，指的是数据在多份副本中存储时，各副本中的数据是一致的。
2）副本一致性
分布式系统当中，数据往往会有多个副本。如果是一台数据库处理所有的数据请求，那么通过ACID四原则，基本可以保证数据的一致性。而多个副本就需要保证数据会有多份拷贝。这就带来了同步的问题，因为我们几乎没有办法保证可以同时更新所有机器当中的包括备份所有数据。 网络延迟，即使我在同一时间给所有机器发送了更新数据的请求，也不能保证这些请求被响应的时间保持一致存在时间差，就会存在某些机器之间的数据不一致的情况。
一致性分类
1、强一致性
这种一致性级别是最符合用户直觉的，它要求系统写入什么，读出来的也会是什么，用户体验好，但实现起来往往对系统的性能影响大。但是强一致性很难实现。
2、弱一致性
这种一致性级别约束了系统在写入成功后，不承诺立即可以读到写入的值，也不承诺多久之后数据能够达到一致，但会尽可能地保证到某个时间级别（比如秒级别）后，数据能够达到一致状态。

5. 分布式理论：CAP定理

CAP 定理
2000 年7月的时候，加州大学伯克利分校的Eric Brewer 教授提出了 CAP 猜想，2年后，被来自于麻省理工的Seth Gilbert 和 Nancy Lynch 从理论上证明了猜想的可能性，从此，CAP 定理正式在学术上成为了分布式计算领域的公认定理。并深深的影响了分布式计算的发展。
CAP 理论含义是，一个分布式系统不可能同时满足一致性（C:Consistency)，可用性（A: Availability）和分区容错性（P：Partition tolerance）这三个基本需求，最多只能同时满足其中的2个。

选项	描述
C 一致性	分布式系统当中的一致性指的是所有节点的数据一致，或者说是所有副本的数据一致
A 可用性	Reads and writes always succeed. 也就是说系统一直可用，而且服务一直保持正常
P 分区容错性	系统在遇到一些节点或者网络分区故障的时候，仍然能够提供满足一致性和可用性的服务

1.舍弃A(可用性)，保留CP(一致性和分区容错性)

一个系统保证了一致性和分区容错性，舍弃可用性。也就是说在极端情况下，允许出现系统无法访问的情况出现，这个时候往往会牺牲用户体验，让用户保持等待，一直到系统数据一致了之后，再恢复服务。

2.舍弃C(一致性)，保留AP(可用性和分区容错性)

这种是大部分的分布式系统的设计，保证高可用和分区容错，但是会牺牲一致性。

3.舍弃P(分区容错性)，保留CA(一致性和可用性)

如果要舍弃P，那么就是要舍弃分布式系统，CAP也就无从谈起了。可以说P是分布式系统的前提，所以这种情况是不存在
的。

6. 分布式理论：BASE 理论

什么是BASE理论
BASE：全称：Basically Available(基本可用)，Soft state（软状态）,和 Eventually consistent（最终一致性）三个短语的缩写，来自 ebay 的架构师提出。
BASE是对CAP中一致性和可用性权衡的结果，BASE理论的核心思想是：即使无法做到强一致性，但每个应用都可以根据自身业务特点，采用适当的方式来使系统达到最终一致性。
①Basically Available(基本可用)
基本可用是指分布式系统在出现不可预知故障的时候，允许损失部分可用性——但请注意，这绝不等价于系统不可用。以下就是两个"基本可用"的例子：

响应时间上的损失：正常情况下，一个在线搜索引擎需要在0.5秒之内返回给用户相应的查询结果，但由于出现故障（比如系统部分机房发生断电或断网故障），查询结果的响应时间增加到了1~2秒。

功能上的损失：正常情况下，在一个电子商务网站（比如淘宝）上购物，消费者几乎能够顺利地完成每一笔订单。但在一些节日大促购物高峰的时候（比如双十一、双十二），由于消费者的购物行为激增，为了保护系统的稳定性（或者保证一致性），部分消费者可能会被引导到一个降级页面。

②Soft state（软状态）
什么是软状态呢？相对于一致性，要求多个节点的数据副本都是一致的，这是一种 “硬状态”。
软状态指的是：允许系统中的数据存在中间状态，并认为该状态不影响系统的整体可用性，即允许系统在多个不同节点的数据副本之间进行数据同步的过程中存在延迟。
③Eventually consistent（最终一致性）
最终一致性强调的是系统中所有的数据副本，在经过一段时间的同步后，最终能够达到一个一致的状态。因此最终一致性的本质是需要系统保证最终数据能够达到一致，而不需要实时保证系统数据的强一致性。

7. 分布式事务

事务的基本特性：(ACID)

Atomicity（原子性）:是说事务是一个不可分割的整体，所有操作要么全做，要么全不做；只要事务中有一个操作出错，回滚到事务开始前的状态的话，那么之前已经执行的所有操作都是无效的，都应该回滚到开始前的状态。

Consistency（一致性）：是说事务执行前后，数据从一个状态到另一个状态必须是一致的，比如A向B转账（A、B的总金额就是一个一致性状态），不可能出现A扣了钱，B却没收到的情况发生。

Isolation（隔离性）：多个并发事务之间相互隔离，不能互相干扰。关于事务的隔离性，可能不是特别好理解，这里的并发事务是指两个事务操作了同一份数据的情况；而对于并发事务操作同一份数据的隔离性问题，则是要求不能出现脏读、幻读的情况，即事务A不能读取事务B还没有提交的数据，或者在事务A读取数据进行更新操作时，不允许事务B率先更新掉这条数据。而为了解决这个问题，常用的手段就是加锁了，对于数据库来说就是通过数据库的相关锁机制来保证。

Durablity（持久性）：事务完成后，对数据库的更改是永久保存的。

8. 分布式理论：一致性协议 2PC

什么是 2PC
2PC （ Two-Phase Commit缩写）即两阶段提交协议，是将整个事务流程分为两个阶段，准备阶段（Prepare phase）、提交阶段（commit phase），2是指两个阶段，P是指准备阶段，C是指提交阶段。在计算机中部分关系数据库如Oracle、MySQL支持两阶段提交协议。
两个阶段过程：

1. 准备阶段（Prepare phase）：事务管理器给每个参与者发送Prepare消息，每个数据库参与者在本地执行事务，并写本地的Undo/Redo日志，此时事务没有提交。 （Undo日志是记录修改前的数据，用于数据库回滚，Redo日志是记录修改后的数据，用于提交事务后写入数据文件）
2. 提交阶段（commit phase）：如果事务管理器收到了参与者的执行失败或者超时消息时，直接给每个参与者发送回滚(Rollback)消息；否则，发送提交(Commit)消息；参与者根据事务管理器的指令执行提交或者回滚操作，并释放事务处理过程中使用的锁资源。注意:必须在最后阶段释放锁资源。

优点：原理简单，实现方便
缺点：
同步阻塞，单点问题，数据不一致，过于保守
同步阻塞：
二阶段提交协议存在最明显也是最大的一个问题就是同步阻塞，在二阶段提交的执行过程中，所有参与该事务操作的逻辑都处于阻塞状态，也就是说，各个参与者在等待其他参与者响应的过程中，无法进行其他操作。这种同步阻塞极大的限制了分布式系统的性能。
单点问题：
协调者在整个二阶段提交过程中很重要，如果协调者在提交阶段出现问题，那么整个流程将无法运转，更重要的是：其他参与者将会处于一直锁定事务资源的状态中，而无法继续完成事务操作。
数据不一致：
假设当协调者向所有的参与者发送 commit 请求之后，发生了局部网络异常或者是协调者在尚未发送完所有commit 请求之前自身发生了崩溃，导致最终只有部分参与者收到了 commit 请求。这将导致严重的数据不一致问题。
过于保守：
如果在二阶段提交的提交询问阶段中，参与者出现故障而导致协调者始终无法获取到所有参与者的响应信息的话，这时协调者只能依靠其自身的超时机制来判断是否需要中断事务，显然，这种策略过于保守。换句话说，二阶段提交协议没有设计较为完善的容错机制，任意一个节点失败都会导致整个事务的失败。

9. 分布式理论：一致性协议 3PC

3PC，全称 “three phase commit”，是 2PC 的改进版，将 2PC 的 “提交事务请求” 过程一分为二，共形成了由CanCommit、PreCommit和doCommit三个阶段组成的事务处理协议。

2PC对比3PC
1.首先对于协调者和参与者都设置了超时机制（在2PC中，只有协调者拥有超时机制，即如果在一定时间内没有收到参与者的消息则默认失败）,主要是避免了参与者在长时间无法与协调者节点通讯（协调者挂掉了）的情况下，无法释放资源的问题，因为参与者自身拥有超时机制会在超时后，自动进行本地commit从而进行释放资源。而这种机制也侧面降低了整个事务的阻塞时间和范围。
2.通过CanCommit、PreCommit、DoCommit三个阶段的设计，相较于2PC而言，多设置了一个缓冲阶段保证了在最后提交阶段之前各参与节点的状态是一致的 。
3.PreCommit是一个缓冲，保证了在最后提交阶段之前各参与节点的状态是一致的。
问题：3PC协议并没有完全解决数据不一致问题。

10. 分布式理论：一致性算法 Paxos

Paxos算法是Lamport提出的一种基于消息传递的分布式一致性算法，使其获得2013年图灵奖。
Paxos由Lamport于1998年在《The Part-Time Parliament》论文中首次公开，最初的描述使用希腊的一个小岛Paxos作为比喻，描述了Paxos小岛中通过决议的流程，并以此命名这个算法，但是这个描述理解起来比较有挑战性。后来在2001年，Lamport觉得同行不能理解他的幽默感，于是重新发表了朴实的算法描述版本《Paxos Made Simple》。
自Paxos问世以来就持续垄断了分布式一致性算法，Paxos这个名词几乎等同于分布式一致性。Google的很多大型分布式系统都采用了Paxos算法来解决分布式一致性问题，如Chubby、Megastore以及Spanner等。开源的ZooKeeper，以及MySQL 5.7推出的用来取代传统的主从复制的MySQL Group Replication等纷纷采用Paxos算法解决分布式一致性问题。然而，Paxos的最大特点就是难，不仅难以理解，更难以实现。

分布式系统才用多副本进行存储数据 , 如果对多个副本执行序列不控制, 那多个副本执行更新操作,由于网络延迟 超时等故障到值各个副本的数据不一致。
我们希望每个副本的执行序列是 [ op1 op2 op3 .... opn ] 不变的, 相同的。
Paxos 一次来确定不可变变量 opi的取值 , 每次确定完Opi之后,各个副本执行opi操作,一次类推。
结论： Paxos算法需要解决的问题就是如何在一个可能发生上述异常的分布式系统中，快速且正确地在集群内部对某个数据的值达成一致。
注：这里某个数据的值并不只是狭义上的某个数，它可以是一条日志，也可以是一条命令（command）。。。根据应用场景不同，某个数据的值有不同的含义。

11. 分布式理论：一致性算法 Raft

什么是Raft 算法？
首先说什么是 Raft 算法：Raft 是一种为了管理复制日志的一致性算法。
Raft提供了和Paxos算法相同的功能和性能，但是它的算法结构和Paxos不同。Raft算法更加容易理解并且更容易构建实际的系统。
Raft将一致性算法分解成了3模块

1. 领导人选举
2. 日志复制
3. 安全性
   Raft算法分为两个阶段，首先是选举过程，然后在选举出来的领导人带领进行正常操作，比如日志复制等。

任务二：分布式系统设计策略

分布式系统本质是通过低廉的硬件攒在一起以获得更好的吞吐量、性能以及可用性等。
在分布式环境下，有几个问题是普遍关心的，我们称之为设计策略：

1. 如何检测当前节点还活着？
2. 如何保障高可用？
3. 容错处理
4. 负载均衡

1. 心跳检测

在分布式环境中，我们提及过存在非常多的节点（Node），其实质是这些节点分担任务的运行、计算或者程序逻辑处理。那么就有一个非常重要的问题，如何检测一个节点出现了故障乃至无法工作了？
通常解决这一问题是采用心跳检测的手段，如同通过仪器对病人进行一些检测诊断一样。
心跳顾名思义，就是以固定的频率向其他节点汇报当前节点状态的方式。收到心跳，一般可以认为一个节点和现在的网络拓扑是良好的。当然，心跳汇报时，一般也会携带一些附加的状态、元数据信息，以便管理。
收到心跳可以确认节点正常，但是收不到心跳也不能认为该节点就已经宣告“死亡”。此时，可以通过一些方法帮助Server做决定： 周期检测心跳机制、累计失效检测机制。
周期检测心跳机制
Server端每间隔 t 秒向Node集群发起监测请求，设定超时时间，如果超过超时时间，则判断“死亡”。
累计失效检测机制
在周期检测心跳机制的基础上，统计一定周期内节点的返回情况（包括超时及正确返回），以此计算节点的“死亡”概率。另外，对于宣告“濒临死亡”的节点可以发起有限次数的重试，以作进一步判断。
通过周期检测心跳机制、累计失效检测机制可以帮助判断节点是否“死亡”，如果判断“死亡”，可以把该节点踢出集群。

2. 高可用设计

高可用(High Availability)是系统架构设计中必须考虑的因素之一,通常是指,经过设计来减少系统不能提供服务的时间。
系统高可用性的常用设计模式包括三种：主备（Master-SLave）、互备（Active-Active）和集群（Cluster）模式。
1.主备模式
主备模式就是Active-Standby模式，当主机宕机时，备机接管主机的一切工作，待主机恢复正常后，按使用者的设定以自动（热备）或手动（冷备）方式将服务切换到主机上运行。在数据库部分，习惯称之为MS模式。MS模式即Master/Slave模式，这在数据库高可用性方案中比较常用，如MySQL、Redis等就采用MS模式实现主从复制。
2.互备模式
互备模式指两台主机同时运行各自的服务工作且相互监测情况。在数据库高可用部分，常见的互备是MM模式。MM模式即Multi-Master模式，指一个系统存在多个master，每个master都具有read-write能力，会根据时间戳或业务逻辑合并版本。
我们使用过的、构建过的MySQL服务绝大多数都是Single-Master，整个拓扑中只有一个Master承担写请求。比如，基于Master-Slave架构的主从复制，但是也存在由于种种原因，我们可能需要MySQL服务具有Multi-Master的特性，希望整个拓扑中可以有不止一个Master承担写请求。
3.集群模式
集群模式是指有多个节点在运行，同时可以通过主控节点分担服务请求。如Zookeeper。集群模式需要解决主控节点本身的高可用问题，一般采用主备模式。

3. 容错性

容错顾名思义就是IT系统对于错误包容的能力。
容错的处理是保障分布式环境下相应系统的高可用或者健壮性，一个典型的案例就是对于缓存穿透问题的解决方案。
我们在项目中使用缓存通常都是先检查缓存中是否存在，如果存在直接返回缓存内容，如果不存在就直接查询数据库然后再缓存查询结果返回。这个时候如果我们查询的某一个数据在缓存中一直不存在，就会造成每一次请求都查询DB，这样缓存就失去了意义，在流量大时，或者有人恶意攻击如频繁发起为id为“-1”的条件进行查询，可能DB就挂掉了。
那这种问题有什么好办法解决呢？
一个比较巧妙的方法是，可以将这个不存在的key预先设定一个值。比如，key=“null”。在返回这个null值的时候，我们的应用就可以认为这是不存在的key，那我们的应用就可以决定是否继续等待访问，还是放弃掉这次操作。如果继续等待访问，过一个时间轮询点后，再次请求这个key，如果取到的值不再是null，则可以认为这时候key有值了，从而避免了透传到数据库，把大量的类似请求挡在了缓存之中。

4. 负载均衡

负载均衡：其关键在于使用多台集群服务器共同分担计算任务，把网络请求及计算分配到集群可用的不同服务器节点上，从而达到高可用性及较好的用户操作体验。
负载均衡器有硬件解决方案，也有软件解决方案。硬件解决方案有著名的F5，软件有LVS、HAProxy、Nginx等。

任务三：分布式架构网络通信

在分布式服务框架中，一个最基础的问题就是远程服务是怎么通讯的，在Java领域中有很多可实现远程通讯的技术，例如：RMI、Hessian、SOAP、ESB和JMS等，它们背后到底是基于什么原理实现的呢。

1. 基本原理

要实现网络机器间的通讯，首先得来看看计算机系统网络通信的基本原理，在底层层面去看，网络通信需要做的就是将流从一台计算机传输到另外一台计算机，基于传输协议和网络IO来实现，其中传输协议比较出名的有tcp、udp等等，tcp、udp都是在基于Socket概念上为某类应用场景而扩展出的传输协议，网络IO，主要有bio、nio、aio三种方式，所有的分布式应用通讯都基于这个原理而实现，只是为了应用的易用，各种语言通常都会提供一些更为贴近应用易用的应用层协议。

2. 什么是RPC

RPC全称为remote procedure call，即远程过程调用。
借助RPC可以做到像本地调用一样调用远程服务，是一种进程间的通信方式比如两台服务器A和B，A服务器上部署一个应用，B服务器上部署一个应用，A服务器上的应用想调用B服务器上的应用提供的方法，由于两个应用不在一个内存空间，不能直接调用，所以需要通过网络来表达调用的语义和传达调用的数据。
需要注意的是RPC并不是一个具体的技术，而是指整个网络远程调用过程。
RPC架构
一个完整的RPC架构里面包含了四个核心的组件，分别是Client，Client Stub，Server以及Server Stub，这个Stub可以理解为存根。
客户端(Client)，服务的调用方。
客户端存根(Client Stub)，存放服务端的地址消息，再将客户端的请求参数打包成网络消息，然后通过网络远程发送给服务方。
服务端(Server)，真正的服务提供者。
服务端存根(Server Stub)，接收客户端发送过来的消息，将消息解包，并调用本地的方法。

3. 什么是RMI

Java RMI 指的是远程方法调用 (Remote Method Invocation),是java原生支持的远程调用 ,采用JRMP（Java Remote Messageing protocol）作为通信协议，可以认为是纯java版本的分布式远程调用解决方案， RMI主要用于不同虚拟机之间的通信，这些虚拟机可以在不同的主机上、也可以在同一个主机上，这里的通信可以理解为一个虚拟机上的对象调用另一个虚拟机上对象的方法。
1.创建远程接口

import java.rmi.Remote;
import java.rmi.RemoteException;
/**
* 远程服务对象接口必须继承Remote接口；同时方法必须抛出RemoteExceptino异常
*/
public interface Hello extends Remote {
public String sayHello(User user) throws RemoteException;
}

import java.io.Serializable;
/**
* 引用对象应该是可序列化对象，这样才能在远程调用的时候：1. 序列化对象 2. 拷贝 3. 在网络中传输
* 4. 服务端反序列化 5. 获取参数进行方法调用； 这种方式其实是将远程对象引用传递的方式转化为值传递的方式
*/
public class User implements Serializable {
private String name;
private int age;
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public int getAge() {
return age;
}
public void setAge(int age) {
this.age = age;
}
}

2.实现远程服务对象

import java.rmi.RemoteException;
import java.rmi.server.UnicastRemoteObject;
/**
* 远程服务对象实现类写在服务端；必须继承UnicastRemoteObject或其子类
**/
public class HelloImpl extends UnicastRemoteObject implements Hello {
/**
* 因为UnicastRemoteObject的构造方法抛出了RemoteException异常，因此这里默认的构造方法必须写，必须
声明抛出RemoteException异常
*
* @throws RemoteException
*/
private static final long serialVersionUID = 3638546195897885959L;
protected HelloImpl() throws RemoteException {
super();
// TODO Auto-generated constructor stub
}
@Override
public String sayHello(User user) throws RemoteException {
System.out.println("this is server, hello:" + user.getName());
return "success";
}
}

3.服务端程序

import java.net.MalformedURLException;
import java.rmi.Naming;
import java.rmi.RemoteException;
import java.rmi.registry.LocateRegistry;
/**
* 服务端程序
**/
public class Server {
public static void main(String[] args) {
try {
Hello hello = new HelloImpl(); // 创建一个远程对象，同时也会创建stub对象、skeleton对象
//本地主机上的远程对象注册表Registry的实例，并指定端口为8888，这一步必不可少（Java默认端口是1099），必不可缺的一步，缺少注册表创建，则无法绑定对象到远程注册表上
LocateRegistry.createRegistry(8080); //启动注册服务
try {
//绑定的URL标准格式为：rmi://host:port/name(其中协议名可以省略，下面两种写法都是正确的）
Naming.bind("//127.0.0.1:8080/zm", hello); //将stub引用绑定到服务地址上
} catch (MalformedURLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
System.out.println("service bind already!!");
} catch (RemoteException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}

4.客户端程序

import java.net.MalformedURLException;
import java.rmi.Naming;
import java.rmi.NotBoundException;
import java.rmi.RemoteException;
/**
* 客户端程序
* @author zm
*
*/
public class Client {
public static void main(String[] args) {
try {
//在RMI服务注册表中查找名称为RHello的对象，并调用其上的方法
Hello hello = (Hello) Naming.lookup("//127.0.0.1:8080/zm");//获取远程对象
User user = new User();
user.setName("james");
System.out.println(hello.sayHello(user));
} catch (MalformedURLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (RemoteException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (NotBoundException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}

4. BIO、NIO、AIO

同步（synchronize）、异步（asychronize）是指应用程序和内核的交互而言的。
1.同步：指用户进程触发IO操作等待或者轮训的方式查看IO操作是否就绪。
2.异步：当一个异步进程调用发出之后，调用者不会立刻得到结果。而是在调用发出之后，被调用者通过状态、通知来通知调用者，或者通过回调函数来处理这个调用。
3.阻塞和非阻塞：针对于进程访问数据的时候,根据IO操作的就绪状态来采取不同的方式。
4.简单点说就是一种读写操作方法的实现方式。阻塞方式下读取和写入将一直等待, 而非阻塞方式下,读取和写入方法会理解返回一个状态值。
BIO
同步阻塞IO。B代表blocking
服务器实现模式为一个连接一个线程，即客户端有连接请求时服务器端就需要启动一个线程进行处理，如果这个连接不做任何事情会造成不必要的线程开销，当然可以通过线程池机制改善。
适用场景：Java1.4之前唯一的选择，简单易用但资源开销太高。
服务端代码

package com.lagou.bio;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
public class IOServer {
public static void main(String[] args) throws Exception {
//首先创建了一个serverSocket
ServerSocket serverSocket = new ServerSocket();
serverSocket.bind(new InetSocketAddress("127.0.0.1",8081));
while (true){
Socket socket = serverSocket.accept(); //同步阻塞
new Thread(()->{
try {
byte[] bytes = new byte[1024];
int len = socket.getInputStream().read(bytes); //同步阻塞
System.out.println(new String(bytes,0,len));
socket.getOutputStream().write(bytes,0,len);
socket.getOutputStream().flush();
} catch (IOException e) {
e.printStackTrace();
}
}).start();
}
}
}

客户端代码

package com.lagou.bio;
import java.io.IOException;
import java.net.Socket;
public class IOClient {
public static void main(String[] args) throws IOException {
Socket socket = new Socket("127.0.0.1",8081);
socket.getOutputStream().write("hello".getBytes());
socket.getOutputStream().flush();
System.out.println("server send back data =====");
byte[] bytes = new byte[1024];
int len = socket.getInputStream().read(bytes);
System.out.println(new String(bytes,0,len));
socket.close();
}
}

NIO

1.NIO介绍
同步非阻塞IO （non-blocking IO / new io）是指JDK 1.4 及以上版本。
服务器实现模式为一个请求一个通道，即客户端发送的连接请求都会注册到多路复用器上，多路复用器轮询到连接有IO请求时才启动一个线程进行处理。
通道（Channels）
NIO 新引入的最重要的抽象是通道的概念。Channel 数据连接的通道。 数据可以从Channel读到Buffer中，也可以从Buffer 写到Channel中 。
缓冲区（Buffers）
通道channel可以向缓冲区Buffer中写数据，也可以像buffer中存数据。
选择器（Selector）
使用选择器，借助单一线程，就可对数量庞大的活动 I/O 通道实时监控和维护。

2.特点
当一个连接创建后，不会需要对应一个线程，这个连接会被注册到多路复用器，所以一个连接只需要一个线程即可，所有的连接需要一个线程就可以操作，该线程的多路复用器会轮训，发现连接有请求时，才开启一个线程处理。
服务端代码

package com.lagou.nio.server;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Scanner;
public class NIOServer extends Thread{
//1.声明多路复用器
private Selector selector;
//2.定义读写缓冲区
private ByteBuffer readBuffer = ByteBuffer.allocate(1024);
private ByteBuffer writeBuffer = ByteBuffer.allocate(1024);
//3.定义构造方法初始化端口
public NIOServer(int port) {
init(port);
}
//4.main方法启动线程
public static void main(String[] args) {
new Thread(new NIOServer(8888)).start();
}
//5.初始化
private void init(int port) {
try {
System.out.println("服务器正在启动......");
//1)开启多路复用器
this.selector = Selector.open();
//2) 开启服务通道
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
//3)设置为非阻塞
serverSocketChannel.configureBlocking(false);
//4)绑定端口
serverSocketChannel.bind(new InetSocketAddress(port));
//5)注册,标记服务通标状态
serverSocketChannel.register(this.selector, SelectionKey.OP_ACCEPT);
System.out.println("服务器启动完毕");
} catch (IOException e) {
e.printStackTrace();
}
}
public void run(){
while (true){
try {
//1.当有至少一个通道被选中,执行此方法
this.selector.select();
//2.获取选中的通道编号集合
Iterator<SelectionKey> keys = this.selector.selectedKeys().iterator();
//3.遍历keys
while (keys.hasNext()) {
SelectionKey key = keys.next();
//4.当前key需要从动刀集合中移出,如果不移出,下次循环会执行对应的逻辑,造成业务错乱
keys.remove();
//5.判断通道是否有效
if (key.isValid()) {
try {
//6.判断是否可读
if (key.isAcceptable()) {
accept(key);
}
} catch (CancelledKeyException e) {
//出现异常断开连接
key.cancel();
}
try {
//7.判断是否可读
if (key.isReadable()) {
read(key);
}
} catch (CancelledKeyException e) {
//出现异常断开连接
key.cancel();
}
try {
//8.判断是否可写
if (key.isWritable()) {
write(key);
}
} catch (CancelledKeyException e) {
//出现异常断开连接
key.cancel();
}
}
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
private void accept(SelectionKey key) {
try {
//1.当前通道在init方法中注册到了selector中的ServerSocketChannel
ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();
//2.阻塞方法, 客户端发起后请求返回.
SocketChannel channel = serverSocketChannel.accept();
///3.serverSocketChannel设置为非阻塞
channel.configureBlocking(false);
//4.设置对应客户端的通道标记,设置次通道为可读时使用
channel.register(this.selector, SelectionKey.OP_READ);
} catch (IOException e) {
e.printStackTrace();
}
}
//使用通道读取数据
private void read(SelectionKey key) {
try{
//清空缓存
this.readBuffer.clear();
//获取当前通道对象
SocketChannel channel = (SocketChannel) key.channel();
//将通道的数据(客户发送的data)读到缓存中.
int readLen = channel.read(readBuffer);
//如果通道中没有数据
if(readLen == -1 ){
//关闭通道
key.channel().close();
//关闭连接
key.cancel();
return;
}
//Buffer中有游标,游标不会重置,需要我们调用flip重置. 否则读取不一致
this.readBuffer.flip();
//创建有效字节长度数组
byte[] bytes = new byte[readBuffer.remaining()];
//读取buffer中数据保存在字节数组
readBuffer.get(bytes);
System.out.println("收到了从客户端 "+ channel.getRemoteAddress() + " : "+ new
String(bytes,"UTF-8"));
//注册通道,标记为写操作
channel.register(this.selector,SelectionKey.OP_WRITE);
}catch (Exception e){
}
}
//给通道中写操作
private void write(SelectionKey key) {
//清空缓存
this.readBuffer.clear();
//获取当前通道对象
SocketChannel channel = (SocketChannel) key.channel();
//录入数据
Scanner scanner = new Scanner(System.in);
try {
System.out.println("即将发送数据到客户端..");
String line = scanner.nextLine();
//把录入的数据写到Buffer中
writeBuffer.put(line.getBytes("UTF-8"));
//重置缓存游标
writeBuffer.flip();
channel.write(writeBuffer);
channel.register(this.selector,SelectionKey.OP_READ);
} catch (Exception e) {
e.printStackTrace();
}
}
}

客户端代码

package com.lagou.nio.client;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.Buffer;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.Scanner;
public class NIOClient {
public static void main(String[] args) {
//创建远程地址
InetSocketAddress address = new InetSocketAddress("127.0.0.1",8888);
SocketChannel channel = null;
//定义缓存
ByteBuffer buffer = ByteBuffer.allocate(1024);
try {
//开启通道
channel = SocketChannel.open();
//连接远程远程服务器
channel.connect(address);
Scanner sc = new Scanner(System.in);
while (true){
System.out.println("客户端即将给 服务器发送数据..");
String line = sc.nextLine();
if(line.equals("exit")){
break;
}
//控制台输入数据写到缓存
buffer.put(line.getBytes("UTF-8"));
//重置buffer 游标
buffer.flip();
//数据发送到数据
channel.write(buffer);
//清空缓存数据
buffer.clear();
//读取服务器返回的数据
int readLen = channel.read(buffer);
if(readLen == -1){
break;
}
//重置buffer游标
buffer.flip();
byte[] bytes = new byte[buffer.remaining()];
//读取数据到字节数组
buffer.get(bytes);
System.out.println("收到了服务器发送的数据 : "+ new String(bytes,"UTF-8"));
buffer.clear();
}
} catch (IOException e) {
e.printStackTrace();
} finally {
if (null != channel){
try {
channel.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
}

AIO
异步非阻塞IO。A代表asynchronize
当有流可以读时,操作系统会将可以读的流传入read方法的缓冲区,并通知应用程序,对于写操作,OS将write方法的流写入完毕是操作系统会主动通知应用程序。因此read和write都是异步 的，完成后会调用回调函数。
使用场景：连接数目多且连接比较长（重操作）的架构，比如相册服务器。重点调用了OS参与并发操作，编程比较复杂。Java7开始支持

5. Netty

Netty 是由 JBOSS 提供一个异步的、 基于事件驱动的网络编程框架。
Netty 可以帮助你快速、 简单的开发出一 个网络应用， 相当于简化和流程化了 NIO 的开发过程。 作为当前最流行的 NIO 框架， Netty 在互联网领域、 大数据分布式计算领域、 游戏行业、 通信行业等获得了广泛的应用， 知名的 Elasticsearch 、 Dubbo 框架内部都采用了 Netty。
为什么使用Netty
NIO缺点
1.NIO 的类库和 API 繁杂，使用麻烦。你需要熟练掌握 Selector、ServerSocketChannel、SocketChannel、ByteBuffer 等。
2.可靠性不强，开发工作量和难度都非常大
3.NIO 的 Bug。例如 Epoll Bug，它会导致 Selector 空轮询，最终导致 CPU 100%。
Netty优点
1.对各种传输协议提供统一的 API
2.高度可定制的线程模型——单线程、一个或多个线程池
3.更好的吞吐量，更低的等待延迟
4.更少的资源消耗
5.最小化不必要的内存拷贝

<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.6.Final</version>
</dependency>

服务端代码

package com.lagou.Netty;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
public class NettyServer {
public static void main(String[] args) throws InterruptedException {
//1.创建 NioEventLoopGroup的两个实例:bossGroup workerGroup
// 当前这两个实例代表两个线程池，默认线程数为CPU核心数乘2
// bossGroup接收客户端传过来的请求
// workerGroup处理请求
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
//2、创建服务启动辅助类:组装一些必要的组件
ServerBootstrap serverBootstrap = new ServerBootstrap();
//设置组,第一个bossGroup负责连接, workerGroup负责连接之后的io处理
serverBootstrap.group(bossGroup,workerGroup)
//channel方法指定服务器监听的通道类型
.channel(NioServerSocketChannel.class)
//设置channel handler , 每一个客户端连接后,给定一个监听器进行处理
.childHandler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel ch) throws Exception {
//传输通道
ChannelPipeline pipeline = ch.pipeline();
//在通道上添加对通道的处理器 , 该处理器可能还是一个监听器
pipeline.addLast(new StringEncoder());
pipeline.addLast(new StringDecoder());
//监听器队列上添加我们自己的处理方式..
pipeline.addLast(new SimpleChannelInboundHandler<String>() {
@Override
protected void channelRead0(ChannelHandlerContext
channelHandlerContext, String s) throws Exception {
System.out.println(s);
}
});
}
});
//bind监听端口
ChannelFuture f = serverBootstrap.bind(8000).sync();
System.out.println("tcp server start success..");
//会阻塞等待直到服务器的channel关闭
f.channel().closeFuture().sync();
}
}

客户端代码

public class NettyClient {
public static void main(String[] args) throws InterruptedException {
//客户端的启动辅助类
Bootstrap bootstrap = new Bootstrap();
//线程池的实例
NioEventLoopGroup group = new NioEventLoopGroup();
//添加到组中
bootstrap.group(group)
//channel方法指定通道类型
.channel(NioSocketChannel.class)
//通道初始化了
.handler(new ChannelInitializer<Channel>() {
@Override
protected void initChannel(Channel ch) {
ch.pipeline().addLast(new StringEncoder());
}
});
Channel channel = bootstrap.connect("127.0.0.1", 8000).channel();
while (true) {
channel.writeAndFlush(new Date() + ": hello world!");
Thread.sleep(2000);
}
}
}

6. 基于Netty自定义RPC

public interface UserService {
String sayHello(String word);
}

public class UserServiceImpl implements UserService {
@Override
public String sayHello(String word) {
System.out.println("调用成功--参数：" + word);
return "调用成功--参数：" + word;
}
public static void startServer(String hostName, int port) {
try {
NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(eventLoopGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline p = ch.pipeline();
p.addLast(new StringDecoder());
p.addLast(new StringEncoder());
p.addLast(new UserServerHandler());
}
});
bootstrap.bind(hostName, port).sync();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

public class UserServerHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
// 如何符合约定，则调用本地方法，返回数据
if (msg.toString().startsWith("UserService")) {
String result = new UserServiceImpl()
.sayHello(msg.toString().substring(msg.toString().lastIndexOf("#") + 1));
ctx.writeAndFlush(result);
}
}
}

public class ServerBootstrap {
public static void main(String[] args) {
UserServiceImpl.startServer("localhost", 8990);
}
}

public class RpcConsumer {
private static ExecutorService executor =
Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
private static UserClientHandler client;
/**
* 创建一个代理对象
*/
public Object createProxy(final Class<?> serviceClass, final String providerName) {
return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
new Class<?>[]{serviceClass}, (proxy, method, args) -> {
if (client == null) {
initClient();
}
// 设置参数
client.setPara(providerName + args[0]);
return executor.submit(client).get();
});
}
/**
* 初始化客户端
*/
private static void initClient() {
client = new UserClientHandler();
EventLoopGroup group = new NioEventLoopGroup();
Bootstrap b = new Bootstrap();
b.group(group)
.channel(NioSocketChannel.class)
.option(ChannelOption.TCP_NODELAY, true)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline p = ch.pipeline();
p.addLast(new StringDecoder());
p.addLast(new StringEncoder());
p.addLast(client);
}
});
try {
b.connect("localhost", 8990).sync();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

public class UserClientHandler extends ChannelInboundHandlerAdapter implements Callable {
private ChannelHandlerContext context;
private String result;
private String para;
@Override
public void channelActive(ChannelHandlerContext ctx) {
context = ctx;
}
/**
* 收到服务端数据，唤醒等待线程
*/
@Override
public synchronized void channelRead(ChannelHandlerContext ctx, Object msg) {
result = msg.toString();
notify();
}
/**
* 写出数据，开始等待唤醒
*/
@Override
public synchronized Object call() throws InterruptedException {
context.writeAndFlush(para);
wait();
return result;
}
public void setPara(String para) {
this.para = para;
}
}

public class ClientBootstrap {
public static final String providerName = "UserService#sayHello#";
public static void main(String[] args) throws InterruptedException {
// 创建一个代理对象
UserService service = (UserService) consumer.createProxy(UserService.class,
providerName);
for (;;) {
Thread.sleep(1000);
System.out.println(service.sayHello("are you ok ?"));
}
}
}