docker中启动neilpang/acme.sh

1

docker run --rm -itd -v "$(pwd)/out":/acme.sh --net=host --name=[acme.sh](http://acme.sh/) neilpang/acme.sh daemon

注册

1

docker exec [acme.sh](http://acme.sh/) --register-account -m [zhangqian9111@gmail.com](mailto:zhangqian9111@gmail.com)

生成证书

1

docker exec [acme.sh](http://acme.sh/) --issue -d xx.merric.top --standalone

生成的证书在$(pwd)/out目录下

将证书配置到nginx配置

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

server{
        listen 80;
        server_name xx.merric.top;
        rewrite ^(.*) https://$server_name$1 permanent;
    }  
       
    server{
        listen 443 ssl;   
        server_name xx.merric.top; 
        root html;
        index index.html index.htm;
        ssl_certificate /app/acme.sh/out/xx.merric.top_ecc/fullchain.cer;  
        ssl_certificate_key /app/acme.sh/out/xx.merric.top_ecc/xx.merric.top.key; 
        ssl_session_timeout 5m;
        ssl_ciphers ECDHE-RSA-AES128-GCM-SHA256:ECDHE:ECDH:AES:HIGH:!NULL:!aNULL:!MD5:!ADH:!RC4;
        ssl_protocols TLSv1.1 TLSv1.2 TLSv1.3;
        ssl_prefer_server_ciphers on;
        location / {
            proxy_pass http://localhost:8001;
            proxy_set_header HOST $host;
            proxy_set_header X-Forwarded-Proto $scheme;
            proxy_set_header X-Real-IP $remote_addr;
            proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        }
    }

启动nginx

2021，我计划还没列完，半年就过去了…

先上两张图。

TCP

TCP 与 UDP 的区别相当大。它充分地实现了数据传输时各种控制功能，可以进行丢包时的重发控制，还可以对次序乱掉的分包进行顺序控制。而这些在 UDP
中都没有。

此外，TCP 作为一种面向有连接的协议，只有在确认通信对端存在时才会发送数据，从而可以控制通信流量的浪费。

根据 TCP 的这些机制，在 IP 这种无连接的网络上也能够实现高可靠性的通信（
主要通过检验和、序列号、确认应答、重发控制、连接管理以及窗口控制等机制实现）。

通过序列号与确认应答提高可靠性

• 在 TCP 中，当发送端的数据到达接收主机时，接收端主机会返回一个已收到消息的通知。这个消息叫做确认应答（ACK）。当发送端将数据发出之后会等待对端的确认应答。如果有确认应答，说明数据已经成功到达对端。 反之，则数据丢失的可能性很大 。
• 在一定时间内没有等待到确认应答，发送端就可以认为数据已经丢失，并进行重发。由此，即使产生了丢包，仍然能够保证数据能够到达对端，实现可靠传输。
• 未收到确认应答并不意味着数据一定丢失。也有可能是数据对方已经收到，只是返回的确认应答在途中丢失。这种情况也会导致发送端误以为数据没有到达目的地而重发数据。
• 此外，也有可能因为一些其他原因导致确认应答延迟到达，在源主机重发数据以后才到达的情况也屡见不鲜。此时，源主机只要按照机制重发数据即可。
• 对于目标主机来说，反复收到相同的数据是不可取的。为了对上层应用提供可靠的传输，目标主机必须放弃重复的数据包。为此我们引入了序列号。
• 序列号是按照顺序给发送数据的每一个字节（8位字节）都标上号码的编号。接收端查询接收数据 TCP 首部中的序列号和数据的长度，将自己下一步应该接收的序列号作为确认应答返送回去。通过序列号和确认应答号，TCP 能够识别是否已经接收数据，又能够判断是否需要接收，从而实现可靠传输。

重发超时的确定

• 重发超时是指在重发数据之前，等待确认应答到来的那个特定时间间隔。 如果超过这个时间仍未收到确认应答，发送端将进行数据重发。最理想的是，找到一个最小时间，它能保证“确认应答一定能在这个时间内返回”。
• TCP 要求不论处在何种网络环境下都要提供高性能通信，并且无论网络拥堵情况发生何种变化，都必须保持这一特性。为此，它在每次发包时都会计算往返时间及其偏差。将这个往返时间和偏差时间相加，重发超时的时间就是比这个总和要稍大一点的值。
• 在 BSD 的 Unix 以及 Windows 系统中，超时都以0.5秒为单位进行控制，因此重发超时都是0.5秒的整数倍。不过，最初其重发超时的默认值一般设置为6秒左右。
• 数据被重发之后若还是收不到确认应答，则进行再次发送。此时，等待确认应答的时间将会以2倍、4倍的指数函数延长。
• 此外， 数据也不会被无限、反复地重发。达到一定重发次数之后，如果仍没有任何确认应答返回，就会判断为网络或对端主机发生了异常，强制关闭连接。并且通知应用通信异常强行终止。

以段为单位发送数据

• 在建立 TCP 连接的同时，也可以确定发送数据包的单位，我们也可以称其为“最大消息长度”（MSS）。最理想的情况是，最大消息长度正好是 IP 中不会被分片处理的最大数据长度。
• TCP 在传送大量数据时，是以 MSS 的大小将数据进行分割发送。进行重发时也是以 MSS 为单位。
• MSS 在三次握手的时候，在两端主机之间被计算得出。两端的主机在发出建立连接的请求时，会在 TCP 首部中写入 MSS 选项，告诉对方自己的接口能够适应的 MSS 的大小。然后会在两者之间选择一个较小的值投入使用。

利用窗口控制提高速度

TCP 以1个段为单位，每发送一个段进行一次确认应答的处理。这样的传输方式有一个缺点，就是包的往返时间越长通信性能就越低。

为解决这个问题，TCP
引入了窗口这个概念。确认应答不再是以每个分段，而是以更大的单位进行确认，转发时间将会被大幅地缩短。也就是说，发送端主机，在发送了一个段以后不必要一直等待确认应答，而是继续发送。如下图所示：

窗口大小就是指无需等待确认应答而可以继续发送数据的最大值。上图中窗口大小为4个段。这个机制实现了使用大量的缓冲区，通过对多个段同时进行确认应答的功能。

滑动窗口控制

上图中的窗口内的数据即便没有收到确认应答也可以被发送出去。不过，在整个窗口的确认应答没有到达之前，如果其中部分数据出现丢包，那么发送端仍然要负责重传。为此，发送端主机需要设置缓存保留这些待被重传的数据，直到收到他们的确认应答。

在滑动窗口以外的部分包括未发送的数据以及已经确认对端已收到的数据。当数据发出后若如期收到确认应答就可以不用再进行重发，此时数据就可以从缓存区清除。

收到确认应答的情况下，将窗口滑动到确认应答中的序列号的位置。这样可以顺序地将多个段同时发送提高通信性能。这种机制也别称为滑动窗口控制。

窗口控制中的重发控制

在使用窗口控制中， 出现丢包一般分为两种情况：

• ① 确认应答未能返回的情况。在这种情况下，数据已经到达对端，是不需要再进行重发的，如下图：

• ② 某个报文段丢失的情况。接收主机如果收到一个自己应该接收的序列号以外的数据时，会针对当前为止收到数据返回确认应答。如下图所示，当某一报文段丢失后，发送端会一直收到序号为1001的确认应答，因此，在窗口比较大，又出现报文段丢失的情况下，同一个序列号的确认应答将会被重复不断地返回。而发送端主机如果连续3次收到同一个确认应答，就会将其对应的数据进行重发。这种机制比之前提到的超时管理更加高效，因此也被称为高速重发控制。

拥塞控制

在介绍具体的概念之前，首先我要提到的是TCP中对于拥塞控制的最高思想，主要有三条，那就是既要尽量快的利用信道的最大能力，不要浪费；又要谨慎的控制信道的负载，不要拥堵；同时如果发生了拥堵，必须能够自我调节，不要崩溃。根据这个最高思想，TCP设计了一系列拥塞避免的算法。

慢启动

在一个网络环境下，信道是共享的，也就是说所有发送的数据都会在这个信道上穿梭。这个道理和高速公路是一样的，你只是高速公路上车流中的一辆车。所以说，当你准备开始发送数据的时候，你不能仅仅做到不管三七二十一，就先按自己的节奏发出自己的数据。如果这个时候信道已经很拥挤，你发出的数据一定也会堵在路上，从而造成进一步的重传，这些重传更加加重了网络的负担，本来拥堵不堪的道路只会越来越差。

所以TCP的设计者们设计了一个保守的策略，这种策略带有浓浓的工科生特点，就是先谨慎的试试，再大胆的前进，实践检验效果。具体的说就是TCP的设计者们采用了一个新的窗口，称之为拥塞窗口，记为cwnd，这个cwnd初始设置为一个one
segment的大小，在这里我们简单的理解为TCP报文一次允许发送的最大的大小以方便后面叙述。当发出这个一个报文之后，如果发送端能够收到对应的ACK。那么下一次就会发送2个segment的报文，接着如果还是收到这2个segment的ACK，那么说明这个信道还是可以承受当前大小的包的，那么接着进行试探，发送4个segment的包，收到4ACK，就发送8个segment，以此指数级增长的类推。

简单的说，慢启动就是一句话，大胆尝试，小心求证。而且在慢启动算法中，报文的增长速度是很快的，所以说这个算法的过程并不符合这个算法的名称，之所以称之为慢启动，是他小心的去试探网络情况的哲学上的”慢慢来”。

拥塞避免

上面的慢启动很明显可以在很短的时间内让信道的利用率达到一个比较理想的值，由于前面一直强调过，信道的资源总是有限的，所以到某一个阶段，这些快速增长的报文必定会充满信道使得信道变得开始拥挤。TCP的设计者们为了不要使得慢启动算法变成一个只有理论意义，而在很短的时间内又会造成网络崩溃从而导致本末倒置。所以说在慢启动的基础上，设计者们做了一些现实的改进与妥协。除了cwnd这个概念，设计者们还定义了一个叫做ssthresh的概念,slow
start thresh,
中文一般翻译为慢启动阈值，比如65535个字节。如果cwnd的大小达到了这个阈值，那么就不采用指数增长这样的比较暴力的方法，而是采用每次收到一个ACK，cwnd就扩大1个单位的方法，这样比较保守，目的之一就是慢慢而可靠的试出来信道所能承受的最大的负载，符合前面所说的哲学的第一条。

但是信道资源总是有限的，而cwnd在前面描述的过程中无论是怎样的方式，他一直递增的，所以触碰到天花板是必然，按照前面的“三条不要”的最高思想，在遇到信道已经塞满的情况下，需要做的就是快速的减少对信道的负担并且能够快速的恢复。

TCP怎么知道信道拥堵了呢？反过来问，信道拥堵会给发送端带来什么呢？第一个明显的效应就是发送出去的消息收不到ACK，也就是前面说的超时了，如果连续几个消息都没有收到ACK而不停的发生重传，那么就可以判定为信道已经拥堵了。这个时候，拥塞避免算法会进入恢复阶段，其方法很简单：

• ssthresh设置为当前发生拥塞cwnd的一般，如果cwnd在30个字节的时候（当然不可能只有这么小）发生了拥堵，那么新的ssthresh就设置为15。
• TCP重新进入慢启动阶段，也就是将cwnd设置为1，指数增长知道达到新的ssthresh的15，然后再重新按照拥塞避免第一个阶段，线性增长。
  这个过程可以用图表示：

快速重传

网络是一个很复杂的环境，如果有这么一种情况，网络发生了拥堵但是又那么的拥堵，这种情况的表现是什么呢？按照前面介绍过的滑动窗口，TCP不是one by
one的发送数据包的，如果发送的数据包是1,2,3，1和3已经到，但是2没有到，由于拥堵在网络中丢失了，那么接收端会不断告诉发送端下一个需要的报文是2号报文，即使你后面的报文都到了，在2号报文没有收到的情况下，会一直发送对1号报文的ACK，表示需要的是2号报文。如果连续收到三个连续的ACK,就认为网络发生了拥堵。用语言描述有点绕，用图来表示就比较清晰。

这种情况说明了两种情况：网络确实发生了拥堵，但是又没有完全拥堵。因为如果完全拥堵了，那么发送端也不会受到三个ACK数据报文，所以这种情况没有必要从头再来，因为最高思想的第二点让我们最大的利用信道的能力。按照这个，设计者们又改进了上面的算法，提出了一个快速重传的方案，其思想如下：

• ssthresh设置为cwnd的一半
• cwnd设置为ssthresh的值
• 不需要重新进入慢启动阶段而是进入拥塞避免阶段
  用图来表示这个过程如下：

快速恢复

快速重传算法已经尽力快的恢复对于网络的传送，但是设计们本着”面包里面抠面粉”的原则，在上面的快速重传算法中尝试想想有没有进步空间，在全面分析之后，提出了快速恢复的算法，其具体做法如下：

• 在收到3个重复的ACK之后，ssthresh设置为cwnd的一半，然后把cwnd设置为ssthresh加3个单位的大小，接着重传丢失的报文段，如果用前面的例子来举例就是重传2号报文。
• 如果这个时候再次收到重传的ACK，那么拥塞窗口增加1。
• 如果收到的是新的数据包的ACK，把cwnd设置为第一步的ssthresh的值。为什么这么做，因为如果收到的新的ACK，说明网络已经恢复了，可以进入拥塞避免的线性增长阶段了。
  第一个例子里为什么加3呢，因为这个时候连续的收到3个ACK包，那么可以认为网络还有3个单位大小的余额，同时也可以这么想，说明有3个“老”的数据包已经从网络上离开了。

粘包和拆包

首先因为tcp是面向字节流的协议，所以他不保证包的独立性，只能在上层应用层进行拆包处理。

发送粘包的原因：

1. 要发送的数据小于TCP发送缓冲区的大小，TCP将多次写入缓冲区的数据一次发送出去；
2. 接收数据端的应用层没有及时读取接收缓冲区中的数据；
3. 数据发送过快，数据包堆积导致缓冲区积压多个数据后才一次性发送出去(如果客户端每发送一条数据就睡眠一段时间就不会发生粘包)；

拆包：

1. 消息定长，例如每个报文的大小为固定长度200字节，如果不够，空位补空格；
2. 在包尾增加回车换行符进行分割，例如FTP协议；
3. 将消息分为消息头和消息体，消息头中包含表示消息总长度（或者消息体长度）的字段，通常设计思路为消息头的第一个字段使用int32来表示消息的总长度；
4. 更复杂的应用层协议。

UDP

UDP协议，即用户数据报协议（User Datagram
Protocol），是一个简单的面向数据报的传输层协议。UDP协议只在IP数据报服务商增加了很少一点的功能，就是复用和分用，以及差错检测的功能。

UDP 常用于一下几个方面：1.包总量较少的通信（DNS、SNMP等）；2.视频、音频等多媒体通信（即时通信）；3.限定于 LAN
等特定网络中的应用通信；4.广播通信（广播、多播）。

UDP特点

1. 无连接的

发送数据之前不需要建立连接，减少了开销和发送数据之前的时延。

2. 尽最大努力交付

不保证可靠的交付，主机不需要维持复杂的链接状态表。

3. 面向报文的

发送方的UDP对应用程序交下来的报文，在添加首部后就向下交付给IP层。既不拆分，也不合并，而是保留这些报文的边界，因此，应用程序需要选择合适的报文大小。

4. 没有拥塞控制。

5. 支持一对一、多对一和多对多的交互通信。

6. 首部开销小，只有8个字节。

UDP首部结构

UDP首部由4各字段组成，各占两个字节：

1. （1）源端口

在需要对方回信时使用，不需要时全为0。

2. 目的端口

发送UDP数据报的目的地。

3. 长度

UDP数据报的长度，最短为8个字节，只包含首部。

4. 检验和

用于检验UDP数据报在传输过程中有没有出差错，有则丢弃。

两者的差别与联系

• TCP 是面向连接的、可靠的流协议。流就是指不间断的数据结构，当应用程序采用 TCP 发送消息时，虽然可以保证发送的顺序，但还是犹如没有任何间隔的数据流发送给接收端。TCP 为提供可靠性传输，实行“顺序控制”或“重发控制”机制。此外还具备“流控制（流量控制）”、“拥塞控制”、提高网络利用率等众多功能。
• UDP 是不具有可靠性的数据报协议。细微的处理它会交给上层的应用去完成。在 UDP 的情况下，虽然可以确保发送消息的大小，却不能保证消息一定会到达。因此，应用有时会根据自己的需要进行重发处理。
• TCP 和 UDP 的优缺点无法简单地、绝对地去做比较：TCP 用于在传输层有必要实现可靠传输的情况；而在一方面，UDP 主要用于那些对高速传输和实时性有较高要求的通信或广播通信。TCP 和 UDP 应该根据应用的目的按需使用。

计算机网络体系结构

在计算机网络的基本概念中，分层次的体系结构是最基本的。计算机网络体系结构的抽象概念较多，在学习时要多思考。这些概念对后面的学习很有帮助。

网络协议是什么？

在计算机网络要做到有条不紊地交换数据，就必须遵守一些事先约定好的规则，比如交换数据的格式、是否需要发送一个应答信息。这些规则被称为网络协议。

为什么要对网络协议分层？

• 简化问题难度和复杂度。由于各层之间独立，我们可以分割大问题为小问题。
• 灵活性好。当其中一层的技术变化时，只要层间接口关系保持不变，其他层不受影响。
• 易于实现和维护。
• 促进标准化工作。分开后，每层功能可以相对简单地被描述。

网络协议分层的缺点： 功能可能出现在多个层里，产生了额外开销。

为了使不同体系结构的计算机网络都能互联，国际标准化组织 ISO 于1977年提出了一个试图使各种计算机在世界范围内互联成网的标准框架，即著名的开放系统互联基本参考模型 OSI/RM，简称为OSI。

OSI 的七层协议体系结构的概念清楚，理论也较完整，但它既复杂又不实用，TCP/IP 体系结构则不同，但它现在却得到了非常广泛的应用。TCP/IP 是一个四层体系结构，它包含应用层，运输层，网际层和网络接口层（用网际层这个名字是强调这一层是为了解决不同网络的互连问题），不过从实质上讲，TCP/IP 只有最上面的三层，因为最下面的网络接口层并没有什么具体内容，因此在学习计算机网络的原理时往往采用折中的办法，即综合 OSI 和 TCP/IP 的优点，采用一种只有五层协议的体系结构，这样既简洁又能将概念阐述清楚，有时为了方便，也可把最底下两层称为网络接口层。

四层协议，五层协议和七层协议的关系如下：

• TCP/IP是一个四层的体系结构，主要包括：应用层、运输层、网际层和网络接口层。
• 五层协议的体系结构主要包括：应用层、运输层、网络层，数据链路层和物理层。
• OSI七层协议模型主要包括是：应用层（Application）、表示层（Presentation）、会话层（Session）、运输层（Transport）、网络层（Network）、数据链路层（Data Link）、物理层（Physical）。

注：五层协议的体系结构只是为了介绍网络原理而设计的，实际应用还是 TCP/IP 四层体系结构。

TCP/IP 协议族

应用层

应用层( application-layer ）的任务是通过应用进程间的交互来完成特定网络应用。应用层协议定义的是应用进程（进程：主机中正在运行的程序）间的通信和交互的规则。

对于不同的网络应用需要不同的应用层协议。在互联网中应用层协议很多，如域名系统 DNS，支持万维网应用的 HTTP 协议，支持电子邮件的 SMTP 协议等等。

运输层

运输层(transport layer)的主要任务就是负责向两台主机进程之间的通信提供通用的数据传输服务。应用进程利用该服务传送应用层报文。

运输层主要使用一下两种协议

1. 传输控制协议-TCP：提供面向连接的，可靠的数据传输服务。
2. 用户数据协议-UDP：提供无连接的，尽最大努力的数据传输服务（不保证数据传输的可靠性）。

每一个应用层（TCP/IP参考模型的最高层）协议一般都会使用到两个传输层协议之一：

运行在TCP协议上的协议：

• HTTP（Hypertext Transfer Protocol，超文本传输协议），主要用于普通浏览。
• HTTPS（HTTP over SSL，安全超文本传输协议）,HTTP协议的安全版本。
• FTP（File Transfer Protocol，文件传输协议），用于文件传输。
• POP3（Post Office Protocol, version 3，邮局协议），收邮件用。
• SMTP（Simple Mail Transfer Protocol，简单邮件传输协议），用来发送电子邮件。
• TELNET（Teletype over the Network，网络电传），通过一个终端（terminal）登陆到网络。
• SSH（Secure Shell，用于替代安全性差的TELNET），用于加密安全登陆用。

运行在UDP协议上的协议：

• BOOTP（Boot Protocol，启动协议），应用于无盘设备。
• NTP（Network Time Protocol，网络时间协议），用于网络同步。
• DHCP（Dynamic Host Configuration Protocol，动态主机配置协议），动态配置IP地址。

运行在TCP和UDP协议上：

• DNS（Domain Name Service，域名服务），用于完成地址查找，邮件转发等工作。

网络层

网络层的任务就是选择合适的网间路由和交换结点，确保计算机通信的数据及时传送。在发送数据时，网络层把运输层产生的报文段或用户数据报封装成分组和包进行传送。在 TCP/IP 体系结构中，由于网络层使用 IP 协议，因此分组也叫 IP 数据报 ，简称数据报。

互联网是由大量的异构（heterogeneous）网络通过路由器（router）相互连接起来的。互联网使用的网络层协议是无连接的网际协议（Intert Prococol）和许多路由选择协议，因此互联网的网络层也叫做网际层或 IP 层。

数据链路层

数据链路层(data link layer)通常简称为链路层。两台主机之间的数据传输，总是在一段一段的链路上传送的，这就需要使用专门的链路层的协议。

在两个相邻节点之间传送数据时，数据链路层将网络层交下来的 IP 数据报组装成帧，在两个相邻节点间的链路上传送帧。每一帧包括数据和必要的控制信息（如同步信息，地址信息，差错控制等）。

在接收数据时，控制信息使接收端能够知道一个帧从哪个比特开始和到哪个比特结束。

一般的web应用的通信传输流是这样的：

发送端在层与层之间传输数据时，每经过一层时会被打上一个该层所属的首部信息。反之，接收端在层与层之间传输数据时，每经过一层时会把对应的首部信息去除。

物理层

在物理层上所传送的数据单位是比特。 物理层(physical layer)的作用是实现相邻计算机节点之间比特流的透明传送，尽可能屏蔽掉具体传输介质和物理设备的差异。使其上面的数据链路层不必考虑网络的具体传输介质是什么。“透明传送比特流”表示经实际电路传送后的比特流没有发生变化，对传送的比特流来说，这个电路好像是看不见的。

TCP/IP 协议族

在互联网使用的各种协议中最重要和最著名的就是 TCP/IP 两个协议。现在人们经常提到的 TCP/IP 并不一定是单指 TCP 和 IP 这两个具体的协议，而往往是表示互联网所使用的整个 TCP/IP 协议族。

互联网协议套件（英语：Internet Protocol Suite，缩写IPS）是一个网络通讯模型，以及一整个网络传输协议家族，为网际网络的基础通讯架构。它常被通称为TCP/IP协议族（英语：TCP/IP Protocol Suite，或TCP/IP Protocols），简称TCP/IP。因为该协定家族的两个核心协定：TCP（传输控制协议）和IP（网际协议），为该家族中最早通过的标准。

TCP的三次握手四次挥手

TCP是一种面向连接的、可靠的、基于字节流的传输层通信协议，在发送数据前，通信双方必须在彼此间建立一条连接。所谓的“连接”，其实是客户端和服务端保存的一份关于对方的信息，如ip地址、端口号等。

TCP可以看成是一种字节流，它会处理IP层或以下的层的丢包、重复以及错误问题。在连接的建立过程中，双方需要交换一些连接的参数。这些参数可以放在TCP头部。

一个TCP连接由一个4元组构成，分别是两个IP地址和两个端口号。一个TCP连接通常分为三个阶段：连接、数据传输、退出（关闭）。通过三次握手建立一个链接，通过四次挥手来关闭一个连接。

当一个连接被建立或被终止时，交换的报文段只包含TCP头部，而没有数据。

TCP报文的头部结构

在了解TCP连接之前先来了解一下TCP报文的头部结构。

上图中有几个字段需要重点介绍下：

（1）序号：seq序号，占32位，用来标识从TCP源端向目的端发送的字节流，发起方发送数据时对此进行标记。

（2）确认序号：ack序号，占32位，只有ACK标志位为1时，确认序号字段才有效，ack=seq+1。

（3）标志位：共6个，即URG、ACK、PSH、RST、SYN、FIN等，具体含义如下：

• ACK：确认序号有效。
• FIN：释放一个连接。
• PSH：接收方应该尽快将这个报文交给应用层。
• RST：重置连接。
• SYN：发起一个新连接。
• URG：紧急指针（urgent pointer）有效。

需要注意的是：

• 不要将确认序号ack与标志位中的ACK搞混了。
• 确认方ack=发起方seq+1，两端配对。

三次握手

三次握手的本质是确认通信双方收发数据的能力

首先，我让信使运输一份信件给对方，对方收到了，那么他就知道了我的发件能力和他的收件能力是可以的。

于是他给我回信，我若收到了，我便知我的发件能力和他的收件能力是可以的，并且他的发件能力和我的收件能力是可以。

然而此时他还不知道他的发件能力和我的收件能力到底可不可以，于是我最后回馈一次，他若收到了，他便清楚了他的发件能力和我的收件能力是可以的。

• 第一次握手：客户端要向服务端发起连接请求，首先客户端随机生成一个起始序列号ISN(比如是100)，那客户端向服务端发送的报文段包含SYN标志位(也就是SYN=1)，序列号seq=100。
• 第二次握手：服务端收到客户端发过来的报文后，发现SYN=1，知道这是一个连接请求，于是将客户端的起始序列号100存起来，并且随机生成一个服务端的起始序列号(比如是300)。然后给客户端回复一段报文，回复报文包含SYN和ACK标志(也就是SYN=1,ACK=1)、序列号seq=300、确认号ack=101(客户端发过来的序列号+1)。
• 第三次握手：客户端收到服务端的回复后发现ACK=1并且ack=101,于是知道服务端已经收到了序列号为100的那段报文；同时发现SYN=1，知道了服务端同意了这次连接，于是就将服务端的序列号300给存下来。然后客户端再回复一段报文给服务端，报文包含ACK标志位(ACK=1)、ack=301(服务端序列号+1)、seq=101(第一次握手时发送报文是占据一个序列号的，所以这次seq就从101开始，需要注意的是不携带数据的ACK报文是不占据序列号的，所以后面第一次正式发送数据时seq还是101)。当服务端收到报文后发现ACK=1并且ack=301，就知道客户端收到序列号为300的报文了，就这样客户端和服务端通过TCP建立了连接。

四次挥手

四次挥手的目的是关闭一个连接

比如客户端初始化的序列号ISA=100，服务端初始化的序列号ISA=300。TCP连接成功后客户端总共发送了1000个字节的数据，服务端在客户端发FIN报文前总共回复了2000个字节的数据。

• 第一次挥手：当客户端的数据都传输完成后，客户端向服务端发出连接释放报文(当然数据没发完时也可以发送连接释放报文并停止发送数据)，释放连接报文包含FIN标志位(FIN=1)、序列号seq=1101(100+1+1000，其中的1是建立连接时占的一个序列号)。需要注意的是客户端发出FIN报文段后只是不能发数据了，但是还可以正常收数据；另外FIN报文段即使不携带数据也要占据一个序列号。
• 第二次挥手：服务端收到客户端发的FIN报文后给客户端回复确认报文，确认报文包含ACK标志位(ACK=1)、确认号ack=1102(客户端FIN报文序列号1101+1)、序列号seq=2300(300+2000)。此时服务端处于关闭等待状态，而不是立马给客户端发FIN报文，这个状态还要持续一段时间，因为服务端可能还有数据没发完。
• 第三次挥手：服务端将最后数据(比如50个字节)发送完毕后就向客户端发出连接释放报文，报文包含FIN和ACK标志位(FIN=1,ACK=1)、确认号和第二次挥手一样ack=1102、序列号seq=2350(2300+50)。
• 第四次挥手：客户端收到服务端发的FIN报文后，向服务端发出确认报文，确认报文包含ACK标志位(ACK=1)、确认号ack=2351、序列号seq=1102。注意客户端发出确认报文后不是立马释放TCP连接，而是要经过2MSL(最长报文段寿命的2倍时长)后才释放TCP连接。而服务端一旦收到客户端发出的确认报文就会立马释放TCP连接，所以服务端结束TCP连接的时间要比客户端早一些。

为什么TCP连接的时候是3次？2次不可以吗？

因为需要考虑连接时丢包的问题，如果只握手2次，第二次握手时如果服务端发给客户端的确认报文段丢失，此时服务端已经准备好了收发数(可以理解服务端已经连接成功)据，而客户端一直没收到服务端的确认报文，所以客户端就不知道服务端是否已经准备好了(可以理解为客户端未连接成功)，这种情况下客户端不会给服务端发数据，也会忽略服务端发过来的数据。

如果是三次握手，即便发生丢包也不会有问题，比如如果第三次握手客户端发的确认ack报文丢失，服务端在一段时间内没有收到确认ack报文的话就会重新进行第二次握手，也就是服务端会重发SYN报文段，客户端收到重发的报文段后会再次给服务端发送确认ack报文。

为什么TCP连接的时候是3次，关闭的时候却是4次？

因为只有在客户端和服务端都没有数据要发送的时候才能断开TCP。而客户端发出FIN报文时只能保证客户端没有数据发了，服务端还有没有数据发客户端是不知道的。而服务端收到客户端的FIN报文后只能先回复客户端一个确认报文来告诉客户端我服务端已经收到你的FIN报文了，但我服务端还有一些数据没发完，等这些数据发完了服务端才能给客户端发FIN报文(所以不能一次性将确认报文和FIN报文发给客户端，就是这里多出来了一次)。

为什么客户端发出第四次挥手的确认报文后要等2MSL的时间才能释放TCP连接？

这里同样是要考虑丢包的问题，如果第四次挥手的报文丢失，服务端没收到确认ack报文就会重发第三次挥手的报文，这样报文一去一回最长时间就是2MSL，所以需要等这么长时间来确认服务端确实已经收到了。

如果已经建立了连接，但是客户端突然出现故障了怎么办？

TCP设有一个保活计时器，客户端如果出现故障，服务器不能一直等下去，白白浪费资源。服务器每收到一次客户端的请求后都会重新复位这个计时器，时间通常是设置为2小时，若两小时还没有收到客户端的任何数据，服务器就会发送一个探测报文段，以后每隔75秒钟发送一次。若一连发送10个探测报文仍然没反应，服务器就认为客户端出了故障，接着就关闭连接。

什么是HTTP，HTTP 与 HTTPS 的区别

HTTP 是一个在计算机世界里专门在两点之间传输文字、图片、音频、视频等超文本数据的约定和规范

常用HTTP状态码

HTTP状态码表示客户端HTTP请求的返回结果、标识服务器处理是否正常、表明请求出现的错误等。

状态码的类别：

常用HTTP状态码：

GET和POST区别

说道GET和POST，就不得不提HTTP协议，因为浏览器和服务器的交互是通过HTTP协议执行的，而GET和POST也是HTTP协议中的两种方法。

HTTP全称为Hyper Text Transfer Protocol，中文翻译为超文本传输协议，目的是保证浏览器与服务器之间的通信。HTTP的工作方式是客户端与服务器之间的请求-应答协议。

HTTP协议中定义了浏览器和服务器进行交互的不同方法，基本方法有4种，分别是GET，POST，PUT，DELETE。这四种方法可以理解为，对服务器资源的查，改，增，删。

• GET：从服务器上获取数据，也就是所谓的查，仅仅是获取服务器资源，不进行修改。
• POST：向服务器提交数据，这就涉及到了数据的更新，也就是更改服务器的数据。
• PUT：英文含义是放置，也就是向服务器新添加数据，就是所谓的增。
• DELETE：从字面意思也能看出，这种方式就是删除服务器数据的过程。

GET和POST区别

1. Get是不安全的，因为在传输过程，数据被放在请求的URL中；Post的所有操作对用户来说都是不可见的。 但是这种做法也不时绝对的，大部分人的做法也是按照上面的说法来的，但是也可以在get请求加上 request body，给 post请求带上 URL 参数。

2. Get请求提交的url中的数据最多只能是2048字节，这个限制是浏览器或者服务器给添加的，http协议并没有对url长度进行限制，目的是为了保证服务器和浏览器能够正常运行，防止有人恶意发送请求。Post请求则没有大小限制。

3. Get限制Form表单的数据集的值必须为ASCII字符；而Post支持整个ISO10646字符集。

4. Get执行效率却比Post方法好。Get是form提交的默认方法。

5. GET产生一个TCP数据包；POST产生两个TCP数据包。

   对于GET方式的请求，浏览器会把http header和data一并发送出去，服务器响应200（返回数据）；

   而对于POST，浏览器先发送header，服务器响应100 continue，浏览器再发送data，服务器响应200 ok（返回数据）。

什么是对称加密与非对称加密

对称密钥加密是指加密和解密使用同一个密钥的方式，这种方式存在的最大问题就是密钥发送问题，即如何安全地将密钥发给对方；

而非对称加密是指使用一对非对称密钥，即公钥和私钥，公钥可以随意发布，但私钥只有自己知道。发送密文的一方使用对方的公钥进行加密处理，对方接收到加密信息后，使用自己的私钥进行解密。
由于非对称加密的方式不需要发送用来解密的私钥，所以可以保证安全性；但是和对称加密比起来，非常的慢

什么是HTTP2

HTTP2 可以提高了网页的性能。

在 HTTP1 中浏览器限制了同一个域名下的请求数量（Chrome 下一般是六个），当在请求很多资源的时候，由于队头阻塞当浏览器达到最大请求数量时，剩余的资源需等待当前的六个请求完成后才能发起请求。

HTTP2 中引入了多路复用的技术，这个技术可以只通过一个 TCP 连接就可以传输所有的请求数据。多路复用可以绕过浏览器限制同一个域名下的请求数量的问题，进而提高了网页的性能。

Session、Cookie和Token的主要区别

HTTP协议本身是无状态的。什么是无状态呢，即服务器无法判断用户身份。

什么是cookie

cookie是由Web服务器保存在用户浏览器上的小文件（key-value格式），包含用户相关的信息。客户端向服务器发起请求，如果服务器需要记录该用户状态，就使用response向客户端浏览器颁发一个Cookie。客户端浏览器会把Cookie保存起来。当浏览器再请求该网站时，浏览器把请求的网址连同该Cookie一同提交给服务器。服务器检查该Cookie，以此来辨认用户身份。

什么是session

session是依赖Cookie实现的。session是服务器端对象

session 是浏览器和服务器会话过程中，服务器分配的一块储存空间。服务器默认为浏览器在cookie中设置 sessionid，浏览器在向服务器请求过程中传输 cookie 包含 sessionid ，服务器根据 sessionid 获取出会话中存储的信息，然后确定会话的身份信息。

cookie与session区别

• 存储位置与安全性：cookie数据存放在客户端上，安全性较差，session数据放在服务器上，安全性相对更高；
• 存储空间：单个cookie保存的数据不能超过4K，很多浏览器都限制一个站点最多保存20个cookie，session无此限制
• 占用服务器资源：session一定时间内保存在服务器上，当访问增多，占用服务器性能，考虑到服务器性能方面，应当使用cookie。

什么是Token

Token的引入：Token是在客户端频繁向服务端请求数据，服务端频繁的去数据库查询用户名和密码并进行对比，判断用户名和密码正确与否，并作出相应提示，在这样的背景下，Token便应运而生。

Token的定义：Token是服务端生成的一串字符串，以作客户端进行请求的一个令牌，当第一次登录后，服务器生成一个Token便将此Token返回给客户端，以后客户端只需带上这个Token前来请求数据即可，无需再次带上用户名和密码。

使用Token的目的：Token的目的是为了减轻服务器的压力，减少频繁的查询数据库，使服务器更加健壮。

Token 是在服务端产生的。如果前端使用用户名/密码向服务端请求认证，服务端认证成功，那么在服务端会返回 Token 给前端。前端可以在每次请求的时候带上 Token 证明自己的合法地位

session与token区别

• session机制存在服务器压力增大，CSRF跨站伪造请求攻击，扩展性不强等问题；
• session存储在服务器端，token存储在客户端
• token提供认证和授权功能，作为身份认证，token安全性比session好；
• session这种会话存储方式方式只适用于客户端代码和服务端代码运行在同一台服务器上，token适用于项目级的前后端分离（前后端代码运行在不同的服务器下）

Servlet是线程安全的吗

Servlet不是线程安全的，多线程并发的读写会导致数据不同步的问题。

解决的办法是尽量不要定义name属性，而是要把name变量分别定义在doGet()和doPost()方法内。虽然使用synchronized(name){}语句块可以解决问题，但是会造成线程的等待，不是很科学的办法。

注意：多线程的并发的读写Servlet类属性会导致数据不同步。但是如果只是并发地读取属性而不写入，则不存在数据不同步的问题。因此Servlet里的只读属性最好定义为final类型的。

Servlet接口中有哪些方法及Servlet生命周期探秘

在Java Web程序中，Servlet主要负责接收用户请求HttpServletRequest，在*doGet()**，**doPost()中做相应的处理，并将回应HttpServletResponse*反馈给用户。Servlet可以设置初始化参数，供Servlet内部使用。

Servlet接口定义了5个方法，其中前三个方法与Servlet生命周期相关：

• void init(ServletConfig config) throws ServletException
• void service(ServletRequest req, ServletResponse resp) throws ServletException, java.io.IOException
• void destory()
• java.lang.String getServletInfo()
• ServletConfig getServletConfig()

生命周期：

Web容器加载Servlet并将其实例化后，Servlet生命周期开始，容器运行其init()方法进行Servlet的初始化；

请求到达时调用Servlet的service()方法，service()方法会根据需要调用与请求对应的doGet或doPost等方法；

当服务器关闭或项目被卸载时服务器会将Servlet实例销毁，此时会调用Servlet的destroy()方法。

init方法和destory方法只会执行一次，service方法客户端每次请求Servlet都会执行。Servlet中有时会用到一些需要初始化与销毁的资源，因此可以把初始化资源的代码放入init方法中，销毁资源的代码放入destroy方法中，这样就不需要每次处理客户端的请求都要初始化与销毁资源。

如果客户端禁止 cookie 能实现 session 还能用吗？

Cookie 与 Session，一般认为是两个独立的东西，Session采用的是在服务器端保持状态的方案，而Cookie采用的是在客户端保持状态的方案。

但为什么禁用Cookie就不能得到Session呢？因为Session是用Session ID来确定当前对话所对应的服务器Session，而Session ID是通过Cookie来传递的，禁用Cookie相当于失去了Session ID，也就得不到Session了。

假定用户关闭Cookie的情况下使用Session，其实现途径有以下几种：

1. 手动通过URL传值、隐藏表单传递Session ID。
2. 用文件、数据库等形式保存Session ID，在跨页过程中手动调用。

1、养成自己的编码规范清单很重要，清单不需要太多，最重要的几条列在前面要严格遵守

2、我的编码规范清单:
a.要严格识别出写的代码是强依赖、弱依赖，弱依赖不能引起核心流程受阻。
反例:一个查询国家列表的下游接口(弱依赖)挂了导致整个收银台挂掉。

3、对一些原则要有敬畏之心，比如: 开闭原则，单一原则。不遵守原则，大概率会发生一些你意料不到的case(你觉得考虑全面了，但实际被打脸)，但遵守原则，会避免很多意料不到的 case。

BlockingQueue

在实际编程中，会经常使用到 JDK 中 Collection 集合框架中的各种容器类如实现 List,Map,Queue 接口的容器类，但是这些容器类基本上不是线程安全的，除了使用 Collections 可以将其转换为线程安全的容器，Doug Lea 大师为我们都准备了对应的线程安全的容器，如实现 List 接口的 CopyOnWriteArrayList，实现 Map 接口的 ConcurrentHashMap，实现 Queue 接口的 ConcurrentLinkedQueue。

最常用的”生产者-消费者“问题中，队列通常被视作线程间操作的数据容器，这样，可以对各个模块的业务功能进行解耦，生产者将“生产”出来的数据放置在数据容器中，而消费者仅仅只需要在“数据容器”中进行获取数据即可，这样生产者线程和消费者线程就能够进行解耦，只专注于自己的业务功能即可。阻塞队列（BlockingQueue）被广泛使用在“生产者-消费者”问题中，其原因是 BlockingQueue 提供了可阻塞的插入和移除的方法。当队列容器已满，生产者线程会被阻塞，直到队列未满；当队列容器为空时，消费者线程会被阻塞，直至队列非空时为止。

核心方法

放入数据：
　　offer(anObject):表示如果可能的话,将anObject加到BlockingQueue里,即如果BlockingQueue可以容纳,则返回true,否则返回false.（本方法不阻塞当前执行方法的线程）
　　offer(E o, long timeout, TimeUnit unit),可以设定等待的时间，如果在指定的时间内，还不能往队列中加入BlockingQueue，则返回失败。
　　put(anObject):把anObject加到BlockingQueue里,如果BlockQueue没有空间,则调用此方法的线程被阻断直到BlockingQueue里面有空间再继续.
获取数据：
　　poll(time):取走BlockingQueue里排在首位的对象,若不能立即取出,则可以等time参数规定的时间,取不到时返回null;
　　**poll(long timeout, TimeUnit unit)**：从BlockingQueue取出一个队首的对象，如果在指定时间内，队列一旦有数据可取，则立即返回队列中的数据。否则知道时间超时还没有数据可取，返回失败。
　　take()**:取走BlockingQueue里排在首位的对象,若BlockingQueue为空,阻断进入等待状态直到BlockingQueue有新的数据被加入;
　　**drainTo():一次性从BlockingQueue获取所有可用的数据对象（还可以指定获取数据的个数）， 通过该方法，可以提升获取数据效率；不需要多次分批加锁或释放锁。

常用的BlockingQueue

1.ArrayBlockingQueue

ArrayBlockingQueue是由数组实现的有界阻塞队列。该队列命令元素 FIFO（先进先出）。因此，对头元素时队列中存在时间最长的数据元素，而对尾数据则是当前队列最新的数据元素。ArrayBlockingQueue 可作为“有界数据缓冲区”，生产者插入数据到队列容器中，并由消费者提取。ArrayBlockingQueue 一旦创建，容量不能改变。

当队列容量满时，尝试将元素放入队列将导致操作阻塞;尝试从一个空队列中取一个元素也会同样阻塞。

ArrayBlockingQueue 默认情况下不能保证线程访问队列的公平性，所谓公平性是指严格按照线程等待的绝对时间顺序，即最先等待的线程能够最先访问到 ArrayBlockingQueue。而非公平性则是指访问 ArrayBlockingQueue 的顺序不是遵守严格的时间顺序，有可能存在，一旦 ArrayBlockingQueue 可以被访问时，长时间阻塞的线程依然无法访问到 ArrayBlockingQueue。如果保证公平性，通常会降低吞吐量。如果需要获得公平性的 ArrayBlockingQueue，可采用如下代码：

1

private static ArrayBlockingQueue<Integer> blockingQueue = new ArrayBlockingQueue<Integer>(10,true);

关于 ArrayBlockingQueue 的实现原理，可以看这篇文章。

2.LinkedBlockingQueue

LinkedBlockingQueue 是用链表实现的有界阻塞队列，同样满足 FIFO 的特性，与 ArrayBlockingQueue 相比起来具有更高的吞吐量，为了防止 LinkedBlockingQueue 容量迅速增，损耗大量内存。通常在创建 LinkedBlockingQueue 对象时，会指定其大小，如果未指定，容量等于 Integer.MAX_VALUE

3.PriorityBlockingQueue

PriorityBlockingQueue 是一个支持优先级的无界阻塞队列。默认情况下元素采用自然顺序进行排序，也可以通过自定义类实现 compareTo()方法来指定元素排序规则，或者初始化时通过构造器参数 Comparator 来指定排序规则。

4.SynchronousQueue

SynchronousQueue 每个插入操作必须等待另一个线程进行相应的删除操作，因此，SynchronousQueue 实际上没有存储任何数据元素，因为只有线程在删除数据时，其他线程才能插入数据，同样的，如果当前有线程在插入数据时，线程才能删除数据。SynchronousQueue 也可以通过构造器参数来为其指定公平性。

5.LinkedTransferQueue

LinkedTransferQueue 是一个由链表数据结构构成的无界阻塞队列，由于该队列实现了 TransferQueue 接口，与其他阻塞队列相比主要有以下不同的方法：

transfer(E e) 如果当前有线程（消费者）正在调用 take()方法或者可延时的 poll()方法进行消费数据时，生产者线程可以调用 transfer 方法将数据传递给消费者线程。如果当前没有消费者线程的话，生产者线程就会将数据插入到队尾，直到有消费者能够进行消费才能退出；

tryTransfer(E e) tryTransfer 方法如果当前有消费者线程（调用 take 方法或者具有超时特性的 poll 方法）正在消费数据的话，该方法可以将数据立即传送给消费者线程，如果当前没有消费者线程消费数据的话，就立即返回false。因此，与 transfer 方法相比，transfer 方法是必须等到有消费者线程消费数据时，生产者线程才能够返回。而 tryTransfer 方法能够立即返回结果退出。

tryTransfer(E e,long timeout,imeUnit unit)
与 transfer 基本功能一样，只是增加了超时特性，如果数据才规定的超时时间内没有消费者进行消费的话，就返回false。

6.LinkedBlockingDeque

LinkedBlockingDeque 是基于链表数据结构的有界阻塞双端队列，如果在创建对象时为指定大小时，其默认大小为 Integer.MAX_VALUE。与 LinkedBlockingQueue 相比，主要的不同点在于，LinkedBlockingDeque 具有双端队列的特性。

7.DelayQueue

DelayQueue 是一个存放实现 Delayed 接口的数据的无界阻塞队列，只有当数据对象的延时时间达到时才能插入到队列进行存储。如果当前所有的数据都还没有达到创建时所指定的延时期，则队列没有队头，并且线程通过 poll 等方法获取数据元素则返回 null。所谓数据延时期满时，则是通过 Delayed 接口的getDelay(TimeUnit.NANOSECONDS)来进行判定，如果该方法返回的是小于等于 0 则说明该数据元素的延时期已满。

CopyOnWriteArrayList

有很多业务往往是读多写少的，比如系统配置的信息，除了在初始进行系统配置的时候需要写入数据，其他大部分时刻其他模块之后对系统信息只需要进行读取，又比如白名单，黑名单等配置，只需要读取名单配置然后检测当前用户是否在该配置范围以内。类似的还有很多业务场景，它们都是属于读多写少的场景。如果在这种情况用到上述的方法，使用 Vector,Collections 转换的这些方式是不合理的，因为尽管多个读线程从同一个数据容器中读取数据，但是读线程对数据容器的数据并不会发生发生修改。很自然而然的我们会联想到 ReenTrantReadWriteLock，通过读写分离的思想，使得读读之间不会阻塞，无疑如果一个 list 能够做到被多个读线程读取的话，性能会大大提升不少。但是，如果仅仅是将 list 通过读写锁（ReentrantReadWriteLock）进行再一次封装的话，由于读写锁的特性，当写锁被写线程获取后，读写线程都会被阻塞。如果仅仅使用读写锁对 list 进行封装的话，这里仍然存在读线程在读数据的时候被阻塞的情况，如果想 list 的读效率更高的话，这里就是我们的突破口，如果我们保证读线程无论什么时候都不被阻塞，效率岂不是会更高？

Doug Lea 大师就为我们提供 CopyOnWriteArrayList 容器可以保证线程安全，保证读读之间在任何时候都不会被阻塞，CopyOnWriteArrayList 也被广泛应用于很多业务场景之中，CopyOnWriteArrayList 值得被我们好好认识一番。

cow的思想

回到上面所说的，如果简单的使用读写锁的话，在写锁被获取之后，读写线程被阻塞，只有当写锁被释放后读线程才有机会获取到锁从而读到最新的数据，站在读线程的角度来看，即读线程任何时候都是获取到最新的数据，满足数据实时性。既然我们说到要进行优化，必然有 trade-off,我们就可以牺牲数据实时性满足数据的最终一致性即可。而 CopyOnWriteArrayList 就是通过 Copy-On-Write(COW)，即写时复制的思想来通过延时更新的策略来实现数据的最终一致性，并且能够保证读线程间不阻塞。

COW 通俗的理解是当我们往一个容器添加元素的时候，不直接往当前容器添加，而是先将当前容器进行 Copy，复制出一个新的容器，然后新的容器里添加元素，添加完元素之后，再将原容器的引用指向新的容器。对 CopyOnWrite 容器进行并发的读的时候，不需要加锁，因为当前容器不会添加任何元素。所以 CopyOnWrite 容器也是一种读写分离的思想，延时更新的策略是通过在写的时候针对的是不同的数据容器来实现的，放弃数据实时性达到数据的最终一致性。

CopyOnWriteArrayList 的实现原理

现在我们来通过看源码的方式来理解 CopyOnWriteArrayList，实际上 CopyOnWriteArrayList 内部维护的就是一个数组

1
2

/** The array, accessed only via getArray/setArray. */
private transient volatile Object[] array;

并且该数组引用是被 volatile 修饰，注意这里仅仅是修饰的是数组引用，其中另有玄机，稍后揭晓。关于 volatile 很重要的一条性质是它能够够保证可见性。对 list 来说，我们自然而然最关心的就是读写的时候，分别为 get 和 add 方法的实现。

get方法

get 方法的源码为：

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

public E get(int index) {
    return get(getArray(), index);
}
/**
 * Gets the array.  Non-private so as to also be accessible
 * from CopyOnWriteArraySet class.
 */
final Object[] getArray() {
    return array;
}
private E get(Object[] a, int index) {
    return (E) a[index];
}

可以看出来 get 方法实现非常简单，几乎就是一个“单线程”程序，没有对多线程添加任何的线程安全控制，也没有加锁也没有 CAS 操作等等，原因是，所有的读线程只是会读取数据容器中的数据，并不会进行修改。

add方法

再来看下如何进行添加数据的？add 方法的源码为：

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

public boolean add(E e) {
    final ReentrantLock lock = this.lock;
	//1. 使用Lock,保证写线程在同一时刻只有一个
    lock.lock();
    try {
		//2. 获取旧数组引用
        Object[] elements = getArray();
        int len = elements.length;
		//3. 创建新的数组，并将旧数组的数据复制到新数组中
        Object[] newElements = Arrays.copyOf(elements, len + 1);
		//4. 往新数组中添加新的数据
		newElements[len] = e;
		//5. 将旧数组引用指向新的数组
        setArray(newElements);
        return true;
    } finally {
        lock.unlock();
    }
}

add 方法的逻辑也比较容易理解，请看上面的注释。需要注意这么几点：

1. 采用 ReentrantLock，保证同一时刻只有一个写线程正在进行数组的复制，否则的话内存中会有多份被复制的数据；
2. 前面说过数组引用是 volatile 修饰的，因此将旧的数组引用指向新的数组，根据 volatile 的 happens-before 规则，写线程对数组引用的修改对读线程是可见的。
3. 由于在写数据的时候，是在新的数组中插入数据的，从而保证读写实在两个不同的数据容器中进行操作。

总结

我们知道 COW 和读写锁都是通过读写分离的思想实现的，但两者还是有些不同，可以进行比较：

COW vs 读写锁

相同点：1. 两者都是通过读写分离的思想实现；2.读线程间是互不阻塞的

不同点：对读线程而言，为了实现数据实时性，在写锁被获取后，读线程会等待或者当读锁被获取后，写线程会等待，从而解决“脏读”等问题。也就是说如果使用读写锁依然会出现读线程阻塞等待的情况。而 COW 则完全放开了牺牲数据实时性而保证数据最终一致性，即读线程对数据的更新是延时感知的，因此读线程不会存在等待的情况。

这里还有这样一个问题： 为什么需要复制呢？ 如果将 array 数组设定为 volitile 的， 对 volatile 变量写 happens-before 读，读线程不是能够感知到 volatile 变量的变化。

原因是，这里 volatile 的修饰的仅仅只是数组引用，数组中的元素的修改是不能保证可见性的。因此 COW 采用的是新旧两个数据容器，通过第 5 行代码将数组引用指向新的数组。

这也是为什么 concurrentHashMap 只具有弱一致性的原因

COW 的缺点

CopyOnWrite 容器有很多优点，但是同时也存在两个问题，即内存占用问题和数据一致性问题。所以在开发的时候需要注意一下。

1. 内存占用问题：因为 CopyOnWrite 的写时复制机制，所以在进行写操作的时候，内存里会同时驻扎两个对 象的内存，旧的对象和新写入的对象（注意:在复制的时候只是复制容器里的引用，只是在写的时候会创建新对 象添加到新容器里，而旧容器的对象还在使用，所以有两份对象内存）。如果这些对象占用的内存比较大，比 如说 200M 左右，那么再写入 100M 数据进去，内存就会占用 300M，那么这个时候很有可能造成频繁的 minor GC 和 major GC。
2. 数据一致性问题：CopyOnWrite 容器只能保证数据的最终一致性，不能保证数据的实时一致性。所以如果你希望写入的的数据，马上能读到，请不要使用 CopyOnWrite 容器。

ConcurrentLinkedQueue

核心方法

操作Node的几个CAS方法

在队列进行出队入队的时候免不了对节点需要进行操作，在多线程就很容易出现线程安全的问题。可以看出在处理器指令集能够支持CMPXCHG指令后，在 java 源码中涉及到并发处理都会使用 CAS 操作(关于 CAS 操作可以看这篇文章的第 3.1 节)，那么在 ConcurrentLinkedQueue 对 Node 的 CAS 操作有这样几个：

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

//更改Node中的数据域item
boolean casItem(E cmp, E val) {
    return UNSAFE.compareAndSwapObject(this, itemOffset, cmp, val);
}
//更改Node中的指针域next
void lazySetNext(Node<E> val) {
    UNSAFE.putOrderedObject(this, nextOffset, val);
}
//更改Node中的指针域next
boolean casNext(Node<E> cmp, Node<E> val) {
    return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
}

可以看出这些方法实际上是通过调用 UNSAFE 实例的方法，UNSAFE 为sun.misc.Unsafe类，该类是 hotspot 底层方法，目前为止了解即可，知道 CAS 的操作归根结底是由该类提供就好。

offer方法

这是一段看着头疼的代码，暂时看懂了可能睡一觉就忘了。。

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 boolean offer(E e) {
    //e为null则抛出空指针异常
    checkNotNull(e);
    final Node<E> newNode = new Node<E>(e);
    //从尾节点插入
    for (Node<E> t = tail, p = t;;) {

        Node<E> q = p.next;

        //如果q=null说明p是尾节点则插入
        if (q == null) {

            //cas插入（1）
            if (p.casNext(null, newNode)) {
                //cas成功说明新增节点已经被放入链表，然后设置当前尾节点（包含head，1，3，5.。。个节点为尾节点）
                if (p != t) // hop two nodes at a time
                    casTail(t, newNode);  // Failure is OK.
                return true;
            }
            // Lost CAS race to another thread; re-read next
        }
        else if (p == q)//(2)
            //多线程操作时候，由于poll时候会把老的head变为自引用，然后head的next变为新head，所以这里需要
            //重新找新的head，因为新的head后面的节点才是激活的节点
            //或者是新的集合，p节点==p节点的next节点，正准备第一次添加节点，所以返回head节点
            p = (t != (t = tail)) ? t : head;
        else
            // 寻找尾节点(3)
            p = (p != t && t != (t = tail)) ? t : q;
    }
}

从源代码角度来看整个入队过程主要做二件事情。第一是定位出尾节点，第二是使用CAS算法能将入队节点设置成尾节点的next节点，如不成功则重试。
第一步定位尾节点。tail节点并不总是尾节点，所以每次入队都必须先通过tail节点来找到尾节点，尾节点可能就是tail节点，也可能是tail节点的next节点。代码中循环体中的第一个if就是判断tail是否有next节点，有则表示next节点可能是尾节点。获取tail节点的next节点需要注意的是p节点等于p的next节点的情况，只有一种可能就是p节点和p的next节点都等于空，表示这个队列刚初始化，正准备添加第一次节点，所以需要返回head节点；或者多线程的情况

第二步设置入队节点为尾节点。 p.casNext(null, n)方法用于将入队节点设置为当前队列尾节点的next节点，p如果是null表示p是当前队列的尾节点，如果不为null表示有其他线程更新了尾节点，则需要重新获取当前队列的尾节点。

poll方法

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

public E poll() {
    restartFromHead:
    //死循环
    for (;;) {
        //死循环
        for (Node<E> h = head, p = h, q;;) {
            //保存当前节点值
            E item = p.item;
            //当前节点有值则cas变为null（1）
            if (item != null && p.casItem(item, null)) {
                //cas成功标志当前节点以及从链表中移除
                if (p != h) // 类似tail间隔2设置一次头节点（2）
                    updateHead(h, ((q = p.next) != null) ? q : p);
                return item;
            }
            //当前队列为空则返回null（3）
            else if ((q = p.next) == null) {
                updateHead(h, p);
                return null;
            }
            //自引用了，则重新找新的队列头节点（4）
            else if (p == q)
                continue restartFromHead;
            else//(5)
                p = q;
        }
    }
}
    final void updateHead(Node<E> h, Node<E> p) {
        if (h != p && casHead(h, p))
            h.lazySetNext(h);
    }

首先获取头节点的元素，然后判断头节点元素是否为空，如果为空，表示另外一个线程已经进行了一次出队操作将该节点的元素取走，如果不为空，则使用CAS的方式将头节点的引用设置成null，如果CAS成功，则直接返回头节点的元素，如果不成功，表示另外一个线程已经进行了一次出队操作更新了head节点，导致元素发生了变化，需要重新获取头节点。

ConcurrentHashmap

关键属性

1

volatile Node<K,V>[] table

：装载Node的数组，作为ConcurrentHashMap的数据容器，采用懒加载的方式，直到第一次插入数据的时候才会进行初始化操作，数组的大小总是为2的幂次方(因为继承自HashMap)。

1

transient volatile Node<K,V>[] nextTable

：扩容时使用，平时为null，只有在扩容的时候才为非null。逻辑机制和ArrayList底层的数组扩容一致。

1

transient volatile long baseCount

：元素数量基础计数器，该值也是一个阶段性的值(产出的时候可能容器正在被修改)。通过CAS的方式进行更改。

1

transient volatile int sizeCtl

：散列表初始化和扩容的大小都是由该变量来控制。

• 当为负数时，它正在被初始化或者扩容。

  • -1表示正在初始化
  • -N表示N-1个线程正在扩容

• 当为整数时，

  • 此时如果当前table数组为null的话表示table正在初始化过程中，sizeCtl表示为需要新建的数组的长度，默认为0

• 若已经初始化了,表示当前数据容器（table数组）可用容量也可以理解成临界值（插入节点数超过了该临界值就需要扩容）,具体指为数组的长度n 乘以 加载因子loadFactor。 当值为0时，即数组长度为默认初始值。

  1	static final sun.misc.Unsafe U

：在ConcurrentHashMapde的实现中可以看到大量的U.compareAndSwapXXXX的方法去修改ConcurrentHashMap的一些属性。这些方法实际上是利用了CAS算法保证了线程安全性，这是一种乐观策略，假设每一次操作都不会产生冲突(变量实际值!=期望值)，当且仅当冲突发生的时候再去尝试。

• 在大量的同步组件和并发容器的实现中使用CAS是通过sun.misc.Unsafe类实现的。该类提供了一些可以直接操控内存和线程的底层操作，可以理解为java中的“指针”。该成员变量的获取是在静态代码块中：

  1 2 3 4 5 6 7 8

  static { try { U = sun.misc.Unsafe.getUnsafe(); ······ } catch (Exception e) { throw new Error(e); } }

CAS操作依赖于现代处理器指令集，通过底层CMPXCHG指令实现。CAS(V,O,N)核心思想为：若当前变量实际值V与期望的旧值O相同，则表明该变量没被其他线程进行修改，因此可以安全的将新值N赋值给变量；若当前变量实际值V与期望的旧值O不相同，则表明该变量已经被其他线程做了处理，此时将新值N赋给变量操作就是不安全的，再进行重试。

核心方法

构造方法

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

// 1. 构造一个空的map，即table数组还未初始化，初始化放在第一次插入数据时，默认大小为16
ConcurrentHashMap()
// 2. 给定map的大小
ConcurrentHashMap(int initialCapacity)
// 3. 给定一个map
ConcurrentHashMap(Map<? extends K, ? extends V> m)
// 4. 给定map的大小以及加载因子
ConcurrentHashMap(int initialCapacity, float loadFactor)
// 5. 给定map大小，加载因子以及并发度（预计同时操作数据的线程）
ConcurrentHashMap(int initialCapacity,float loadFactor, int concurrencyLevel)
public ConcurrentHashMap(int initialCapacity) {
	//1. 小于0直接抛异常
    if (initialCapacity < 0)
        throw new IllegalArgumentException();
	//2. 判断是否超过了允许的最大值，超过了话则取最大值，否则再对该值进一步处理
    int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?
               MAXIMUM_CAPACITY :
               tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));
	//3. 赋值给sizeCtl
    this.sizeCtl = cap;
}

这段代码的逻辑请看注释，很容易理解，如果小于 0 就直接抛出异常，如果指定值大于了所允许的最大值的话就取最大值，否则，在对指定值做进一步处理。最后将 cap 赋值给 sizeCtl,关于 sizeCtl 的说明请看上面的说明，当调用构造器方法之后，sizeCtl 的大小应该就代表了 ConcurrentHashMap 的大小，即 table 数组长度。tableSizeFor 做了哪些事情了？源码为：

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

/**
 * Returns a power of two table size for the given desired capacity.
 * See Hackers Delight, sec 3.2
 */
private static final int tableSizeFor(int c) {
    int n = c - 1;
    n |= n >>> 1;
    n |= n >>> 2;
    n |= n >>> 4;
    n |= n >>> 8;
    n |= n >>> 16;
    return (n < 0) ? 1 : (n >= MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : n + 1;
}

这个和前一篇文章中写的的方法一样，可以复习一下

初始化 initTable方法

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

private final Node<K,V>[] initTable() {
    Node<K,V>[] tab; int sc;
    while ((tab = table) == null || tab.length == 0) {
        if ((sc = sizeCtl) < 0)
			// 1. 保证只有一个线程正在进行初始化操作
            Thread.yield(); // lost initialization race; just spin
        else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
            try {
                if ((tab = table) == null || tab.length == 0) {
					// 2. 得出数组的大小
                    int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                    @SuppressWarnings("unchecked")
					// 3. 这里才真正的初始化数组
                    Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                    table = tab = nt;
					// 4. 计算数组中可用的大小：实际大小n*0.75（加载因子）
                    sc = n - (n >>> 2);
                }
            } finally {
                sizeCtl = sc;
            }
            break;
        }
    }
    return tab;
}

代码的逻辑请见注释，有可能存在一个情况是多个线程同时走到这个方法中，为了保证能够正确初始化，在第 1 步中会先通过 if 进行判断，若当前已经有一个线程正在初始化即 sizeCtl 值变为-1，这个时候其他线程在 If 判断为 true 从而调用 Thread.yield()让出 CPU 时间片。正在进行初始化的线程会调用 U.compareAndSwapInt 方法将 sizeCtl 改为-1 即正在初始化的状态。另外还需要注意的事情是，在第四步中会进一步计算数组中可用的大小即为数组实际大小 n 乘以加载因子 0.75.可以看看这里乘以 0.75 是怎么算的，0.75 为四分之三，这里n - (n >>> 2)是不是刚好是 n-(1/4)n=(3/4)n，挺有意思的吧:)。如果选择是无参的构造器的话，这里在 new Node 数组的时候会使用默认大小为DEFAULT_CAPACITY（16），然后乘以加载因子 0.75 为 12，也就是说数组的可用大小为 12。

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
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

final V putVal(K key, V value, boolean onlyIfAbsent) {
    if (key == null || value == null) throw new NullPointerException();
	//1. 计算key的hash值
    int hash = spread(key.hashCode());
    int binCount = 0;
    for (Node<K,V>[] tab = table;;) {
        Node<K,V> f; int n, i, fh;
		//2. 如果当前table还没有初始化先调用initTable方法将tab进行初始化
        if (tab == null || (n = tab.length) == 0)
            tab = initTable();
		//3. tab中索引为i的位置的元素为null，则直接使用CAS将值插入即可
        else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
            if (casTabAt(tab, i, null,
                         new Node<K,V>(hash, key, value, null)))
                break;                   // no lock when adding to empty bin
        }
		//4. 当前正在扩容
        else if ((fh = f.hash) == MOVED)
            tab = helpTransfer(tab, f);
        else {
            V oldVal = null;
            synchronized (f) {
                if (tabAt(tab, i) == f) {
					//5. 当前为链表，在链表中插入新的键值对
                    if (fh >= 0) {
                        binCount = 1;
                        for (Node<K,V> e = f;; ++binCount) {
                            K ek;
                            if (e.hash == hash &&
                                ((ek = e.key) == key ||
                                 (ek != null && key.equals(ek)))) {
                                oldVal = e.val;
                                if (!onlyIfAbsent)
                                    e.val = value;
                                break;
                            }
                            Node<K,V> pred = e;
                            if ((e = e.next) == null) {
                                pred.next = new Node<K,V>(hash, key,
                                                          value, null);
                                break;
                            }
                        }
                    }
					// 6.当前为红黑树，将新的键值对插入到红黑树中
                    else if (f instanceof TreeBin) {
                        Node<K,V> p;
                        binCount = 2;
                        if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
                                                       value)) != null) {
                            oldVal = p.val;
                            if (!onlyIfAbsent)
                                p.val = value;
                        }
                    }
                }
            }
			// 7.插入完键值对后再根据实际大小看是否需要转换成红黑树
            if (binCount != 0) {
                if (binCount >= TREEIFY_THRESHOLD)
                    treeifyBin(tab, i);
                if (oldVal != null)
                    return oldVal;
                break;
            }
        }
    }
	//8.对当前容量大小进行检查，如果超过了临界值（实际大小*加载因子）就需要扩容，而且对map的count值+1
    addCount(1L, binCount);
    return null;
}

有了代码里的注释我觉得已经很清楚了，可以看到开头有 如果key或者value是null的话会抛出空指针异常，这也是为什么ConcurrentHashMap中key和value不能为null的原因，之所以这样，首先value不能为null是因为ConcurrentHashMap调用map.get(key)的时候，如果返回了null，那么这个null，都有两重含义:

1. 这个key从来没有在map中映射过。
2. 这个key的value在设置的时候，就是null。

在非线程安全的map集合(HashMap)中可以使用map.contains(key)方法来判断，而ConcurrentHashMap却不可以。因为多线程情况下get和contains方法之间可能容器已经有了改变。这里有源码作者大佬的回答

transfer 方法

当 ConcurrentHashMap 容量不足的时候，需要对 table 进行扩容。这个方法的基本思想跟 HashMap 是很像的，但是由于它是支持并发扩容的，所以要复杂的多。原因是它支持多线程进行扩容操作，而并没有加锁。我想这样做的目的不仅仅是为了满足 concurrent 的要求，而是希望利用并发处理去减少扩容带来的时间影响。transfer 方法源码为：

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
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
    int n = tab.length, stride;
    // 将 length / 8 然后除以 CPU核心数。如果得到的结果小于 16，那么就使用 16。
    // 这里的目的是让每个 CPU 处理的桶一样多，避免出现转移任务不均匀的现象，如果桶较少的话，默认一个 CPU（一个线程）处理 16 个桶
    if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
        stride = MIN_TRANSFER_STRIDE; // subdivide range 细分范围 stridea：TODO
    // 新的 table 尚未初始化
    if (nextTab == null) {            // initiating
        try {
            // 扩容  2 倍
            Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
            // 更新
            nextTab = nt;
        } catch (Throwable ex) {      // try to cope with OOME
            // 扩容失败， sizeCtl 使用 int 最大值。
            sizeCtl = Integer.MAX_VALUE;
            return;// 结束
        }
        // 更新成员变量
        nextTable = nextTab;
        // 更新转移下标，就是 老的 tab 的 length
        transferIndex = n;
    }
    // 新 tab 的 length
    int nextn = nextTab.length;
    // 创建一个 fwd 节点，用于占位。当别的线程发现这个槽位中是 fwd 类型的节点，则跳过这个节点。
    ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
    // 首次推进为 true，如果等于 true，说明需要再次推进一个下标（i--），反之，如果是 false，那么就不能推进下标，需要将当前的下标处理完毕才能继续推进
    boolean advance = true;
    // 完成状态，如果是 true，就结束此方法。
    boolean finishing = false; // to ensure sweep before committing nextTab
    // 死循环,i 表示下标，bound 表示当前线程可以处理的当前桶区间最小下标
    for (int i = 0, bound = 0;;) {
        Node<K,V> f; int fh;
        // 如果当前线程可以向后推进；这个循环就是控制 i 递减。同时，每个线程都会进入这里取得自己需要转移的桶的区间
        while (advance) {
            int nextIndex, nextBound;
            // 对 i 减一，判断是否大于等于 bound （正常情况下，如果大于 bound 不成立，说明该线程上次领取的任务已经完成了。那么，需要在下面继续领取任务）
            // 如果对 i 减一大于等于 bound（还需要继续做任务），或者完成了，修改推进状态为 false，不能推进了。任务成功后修改推进状态为 true。
            // 通常，第一次进入循环，i-- 这个判断会无法通过，从而走下面的 nextIndex 赋值操作（获取最新的转移下标）。其余情况都是：如果可以推进，将 i 减一，然后修改成不可推进。如果 i 对应的桶处理成功了，改成可以推进。
            if (--i >= bound || finishing)
                advance = false;// 这里设置 false，是为了防止在没有成功处理一个桶的情况下却进行了推进
            // 这里的目的是：1. 当一个线程进入时，会选取最新的转移下标。2. 当一个线程处理完自己的区间时，如果还有剩余区间的没有别的线程处理。再次获取区间。
            else if ((nextIndex = transferIndex) <= 0) {
                // 如果小于等于0，说明没有区间了 ，i 改成 -1，推进状态变成 false，不再推进，表示，扩容结束了，当前线程可以退出了
                // 这个 -1 会在下面的 if 块里判断，从而进入完成状态判断
                i = -1;
                advance = false;// 这里设置 false，是为了防止在没有成功处理一个桶的情况下却进行了推进
            }// CAS 修改 transferIndex，即 length - 区间值，留下剩余的区间值供后面的线程使用
            else if (U.compareAndSwapInt
                     (this, TRANSFERINDEX, nextIndex,
                      nextBound = (nextIndex > stride ?
                                   nextIndex - stride : 0))) {
                bound = nextBound;// 这个值就是当前线程可以处理的最小当前区间最小下标
                i = nextIndex - 1; // 初次对i 赋值，这个就是当前线程可以处理的当前区间的最大下标
                advance = false; // 这里设置 false，是为了防止在没有成功处理一个桶的情况下却进行了推进，这样对导致漏掉某个桶。下面的 if (tabAt(tab, i) == f) 判断会出现这样的情况。
            }
        }// 如果 i 小于0 （不在 tab 下标内，按照上面的判断，领取最后一段区间的线程扩容结束）
        //  如果 i >= tab.length(不知道为什么这么判断)
        //  如果 i + tab.length >= nextTable.length  （不知道为什么这么判断）
        if (i < 0 || i >= n || i + n >= nextn) {
            int sc;
            if (finishing) { // 如果完成了扩容
                nextTable = null;// 删除成员变量
                table = nextTab;// 更新 table
                sizeCtl = (n << 1) - (n >>> 1); // 更新阈值
                return;// 结束方法。
            }// 如果没完成
            if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {// 尝试将 sc -1. 表示这个线程结束帮助扩容了，将 sc 的低 16 位减一。
                if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)// 如果 sc - 2 不等于标识符左移 16 位。如果他们相等了，说明没有线程在帮助他们扩容了。也就是说，扩容结束了。
                    return;// 不相等，说明没结束，当前线程结束方法。
                finishing = advance = true;// 如果相等，扩容结束了，更新 finising 变量
                i = n; // 再次循环检查一下整张表
            }
        }
        else if ((f = tabAt(tab, i)) == null) // 获取老 tab i 下标位置的变量，如果是 null，就使用 fwd 占位。
            advance = casTabAt(tab, i, null, fwd);// 如果成功写入 fwd 占位，再次推进一个下标
        else if ((fh = f.hash) == MOVED)// 如果不是 null 且 hash 值是 MOVED。
            advance = true; // already processed // 说明别的线程已经处理过了，再次推进一个下标
        else {// 到这里，说明这个位置有实际值了，且不是占位符。对这个节点上锁。为什么上锁，防止 putVal 的时候向链表插入数据
            synchronized (f) {
                // 判断 i 下标处的桶节点是否和 f 相同
                if (tabAt(tab, i) == f) {
                    Node<K,V> ln, hn;// low, height 高位桶，低位桶
                    // 如果 f 的 hash 值大于 0 。TreeBin 的 hash 是 -2
                    if (fh >= 0) {
                        // 对老长度进行与运算（第一个操作数的的第n位于第二个操作数的第n位如果都是1，那么结果的第n为也为1，否则为0）
                        // 由于 Map 的长度都是 2 的次方（000001000 这类的数字），那么取于 length 只有 2 种结果，一种是 0，一种是1
                        //  如果是结果是0 ，Doug Lea 将其放在低位，反之放在高位，目的是将链表重新 hash，放到对应的位置上，让新的取于算法能够击中他。
                        int runBit = fh & n;
                        Node<K,V> lastRun = f; // 尾节点，且和头节点的 hash 值取于不相等
                        // 遍历这个桶
                        for (Node<K,V> p = f.next; p != null; p = p.next) {
                            // 取于桶中每个节点的 hash 值
                            int b = p.hash & n;
                            // 如果节点的 hash 值和首节点的 hash 值取于结果不同
                            if (b != runBit) {
                                runBit = b; // 更新 runBit，用于下面判断 lastRun 该赋值给 ln 还是 hn。
                                lastRun = p; // 这个 lastRun 保证后面的节点与自己的取于值相同，避免后面没有必要的循环
                            }
                        }
                        if (runBit == 0) {// 如果最后更新的 runBit 是 0 ，设置低位节点
                            ln = lastRun;
                            hn = null;
                        }
                        else {
                            hn = lastRun; // 如果最后更新的 runBit 是 1， 设置高位节点
                            ln = null;
                        }// 再次循环，生成两个链表，lastRun 作为停止条件，这样就是避免无谓的循环（lastRun 后面都是相同的取于结果）
                        for (Node<K,V> p = f; p != lastRun; p = p.next) {
                            int ph = p.hash; K pk = p.key; V pv = p.val;
                            // 如果与运算结果是 0，那么就还在低位
                            if ((ph & n) == 0) // 如果是0 ，那么创建低位节点
                                ln = new Node<K,V>(ph, pk, pv, ln);
                            else // 1 则创建高位
                                hn = new Node<K,V>(ph, pk, pv, hn);
                        }
                        // 其实这里类似 hashMap 
                        // 设置低位链表放在新链表的 i
                        setTabAt(nextTab, i, ln);
                        // 设置高位链表，在原有长度上加 n
                        setTabAt(nextTab, i + n, hn);
                        // 将旧的链表设置成占位符
                        setTabAt(tab, i, fwd);
                        // 继续向后推进
                        advance = true;
                    }// 如果是红黑树
                    else if (f instanceof TreeBin) {
                        TreeBin<K,V> t = (TreeBin<K,V>)f;
                        TreeNode<K,V> lo = null, loTail = null;
                        TreeNode<K,V> hi = null, hiTail = null;
                        int lc = 0, hc = 0;
                        // 遍历
                        for (Node<K,V> e = t.first; e != null; e = e.next) {
                            int h = e.hash;
                            TreeNode<K,V> p = new TreeNode<K,V>
                                (h, e.key, e.val, null, null);
                            // 和链表相同的判断，与运算 == 0 的放在低位
                            if ((h & n) == 0) {
                                if ((p.prev = loTail) == null)
                                    lo = p;
                                else
                                    loTail.next = p;
                                loTail = p;
                                ++lc;
                            } // 不是 0 的放在高位
                            else {
                                if ((p.prev = hiTail) == null)
                                    hi = p;
                                else
                                    hiTail.next = p;
                                hiTail = p;
                                ++hc;
                            }
                        }
                        // 如果树的节点数小于等于 6，那么转成链表，反之，创建一个新的树
                        ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :
                            (hc != 0) ? new TreeBin<K,V>(lo) : t;
                        hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
                            (lc != 0) ? new TreeBin<K,V>(hi) : t;
                        // 低位树
                        setTabAt(nextTab, i, ln);
                        // 高位数
                        setTabAt(nextTab, i + n, hn);
                        // 旧的设置成占位符
                        setTabAt(tab, i, fwd);
                        // 继续向后推进
                        advance = true;
                    }
                }
            }
        }
    }
}

该方法的执行逻辑如下：

1. 通过计算 CPU 核心数和 Map 数组的长度得到每个线程（CPU）要帮助处理多少个桶，并且这里每个线程处理都是平均的。默认每个线程处理 16 个桶。因此，如果长度是 16 的时候，扩容的时候只会有一个线程扩容。

2. 初始化临时变量 nextTable。将其在原有基础上扩容两倍。

3. 死循环开始转移。多线程并发转移就是在这个死循环中，根据一个 finishing 变量来判断，该变量为 true 表示扩容结束，否则继续扩容。

   3.1 进入一个 while 循环，分配数组中一个桶的区间给线程，默认是 16. 从大到小进行分配。当拿到分配值后，进行 i– 递减。这个 i 就是数组下标。（其中有一个 bound 参数，这个参数指的是该线程此次可以处理的区间的最小下标，超过这个下标，就需要重新领取区间或者结束扩容，还有一个 advance 参数，该参数指的是是否继续递减转移下一个桶，如果为 true，表示可以继续向后推进，反之，说明还没有处理好当前桶，不能推进)
   3.2 出 while 循环，进 if 判断，判断扩容是否结束，如果扩容结束，清空临死变量，更新 table 变量，更新库容阈值。如果没完成，但已经无法领取区间（没了），该线程退出该方法，并将 sizeCtl 减一，表示扩容的线程少一个了。如果减完这个数以后，sizeCtl 回归了初始状态，表示没有线程再扩容了，该方法所有的线程扩容结束了。（这里主要是判断扩容任务是否结束，如果结束了就让线程退出该方法，并更新相关变量）。然后检查所有的桶，防止遗漏。
   3.3 如果没有完成任务，且 i 对应的槽位是空，尝试 CAS 插入占位符，让 putVal 方法的线程感知。
   3.4 如果 i 对应的槽位不是空，且有了占位符，那么该线程跳过这个槽位，处理下一个槽位。
   3.5 如果以上都是不是，说明这个槽位有一个实际的值。开始同步处理这个桶。
   3.6 到这里，都还没有对桶内数据进行转移，只是计算了下标和处理区间，然后一些完成状态判断。同时，如果对应下标内没有数据或已经被占位了，就跳过了。

4. 处理每个桶的行为都是同步的。防止 putVal 的时候向链表插入数据。
   4.1 如果这个桶是链表，那么就将这个链表根据 length 取于拆成两份，取于结果是 0 的放在新表的低位，取于结果是 1 放在新表的高位。
   4.2 如果这个桶是红黑数，那么也拆成 2 份，方式和链表的方式一样，然后，判断拆分过的树的节点数量，如果数量小于等于 6，改造成链表。反之，继续使用红黑树结构。
   4.3 到这里，就完成了一个桶从旧表转移到新表的过程。

这段代码是既牛逼又头疼，这注释和逻辑也可能不完全对，后面需要再看再修改

size 方法

ConcurrentHashMap提供了baseCount、counterCells两个辅助变量和一个CounterCell辅助内部类。

1
2
3
4
5
6
7
8
9

@sun.misc.Contended static final class CounterCell {
        volatile long value;
        CounterCell(long x) { value = x; }
}

//ConcurrentHashMap中元素个数,但返回的不一定是当前Map的真实元素个数。基于CAS无锁更新
private transient volatile long baseCount;

private transient volatile CounterCell[] counterCells;

size()方法定义如下：

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

public int size() {
        long n = sumCount();
        return ((n < 0L) ? 0 :
                (n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
                (int)n);
}
final long sumCount() {
        CounterCell[] as = counterCells; CounterCell a;
        long sum = baseCount;
        if (as != null) {
            for (int i = 0; i < as.length; ++i) {
                //遍历，所有counter求和
                if ((a = as[i]) != null)
                    sum += a.value;     
            }
        }
        return sum;
}

通过上述size()逻辑可以知道：**size = baseCount + counterCells[0...n-1].value**，我们通过增加一个元素的逻辑代码来看这两个变量的含义。

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

private final void addCount(long x, int check) {
        CounterCell[] as; long b, s;
        // s = b + x，完成baseCount++操作；
        if ((as = counterCells) != null ||
            !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
            CounterCell a; long v; int m;
            boolean uncontended = true;
            if (as == null || (m = as.length - 1) < 0 ||
                (a = as[ThreadLocalRandom.getProbe() & m]) == null ||
                !(uncontended =
                  U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
                //  多线程CAS发生失败时执行
                fullAddCount(x, uncontended);
                return;
            }
            if (check <= 1)
                return;
            s = sumCount();
        }

        // 检查是否进行扩容
    }

在并发量很高时，如果存在两个线程同时执行CAS修改baseCount值，则失败的线程会继续执行方法体中的逻辑，使用CounterCell记录元素个数的变化；
如果通过CAS设置cellsBusy字段失败的话，则继续尝试通过CAS修改baseCount字段，如果修改baseCount字段成功的话，就退出循环，否则继续循环插入CounterCell对象；
所以在1.8中的size实现比1.7简单多，因为元素个数保存baseCount中，部分元素的变化个数保存在CounterCell数组中，实现如下：
通过累加baseCount和CounterCell数组中的数量，即可得到元素的总个数；

线程的生命周期及基本状态

1. 新建(new)：新创建了一个线程对象。

2. 可运行(runnable)：线程对象创建后，当调用线程对象的 start()方法，该线程处于就绪状态，等待被线程调度选中，获取cpu的使用权。

3. 运行(running)：可运行状态(runnable)的线程获得了cpu时间片（timeslice），执行程序代码。注：就绪状态是进入到运行状态的唯一入口，也就是说，线程要想进入运行状态执行，首先必须处于就绪状态中；

4. 阻塞(block)：处于运行状态中的线程由于某种原因，暂时放弃对 CPU的使用权，停止执行，此时进入阻塞状态，直到其进入到就绪状态，才有机会再次被 CPU 调用以进入到运行状态。

   阻塞的情况分三种：
   (一). 等待阻塞：运行状态中的线程执行 wait()方法，JVM会把该线程放入等待队列(waitting queue)中，使本线程进入到等待阻塞状态；
   (二). 同步阻塞：线程在获取 synchronized 同步锁失败(因为锁被其它线程所占用)，则JVM会把该线程放入锁池(lock pool)中，线程会进入同步阻塞状态；
   (三). 其他阻塞: 通过调用线程的 sleep()或 join()或发出了 I/O 请求时，线程会进入到阻塞状态。当 sleep()状态超时、join()等待线程终止或者超时、或者 I/O 处理完毕时，线程重新转入就绪状态。

5. 死亡(dead)：线程run()、main()方法执行结束，或者因异常退出了run()方法，则该线程结束生命周期。死亡的线程不可再次复生。

创建线程的几种方式

• 继承 Thread 类；
• 实现 Runnable 接口；
• 实现 Callable 接口；
• 使用 Executors 工具类创建线程池

继承 Thread 类

1. 定义一个Thread类的子类，重写run方法，将相关逻辑实现，run()方法就是线程要执行的业务逻辑方法
2. 创建自定义的线程子类对象
3. 调用子类实例的star()方法来启动线程

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

public class MyThread extends Thread {
    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName() + " run()方法正在执行...");
    }
}
public class TheadTest {
    public static void main(String[] args) {
        MyThread myThread = new MyThread(); 	
        myThread.start();
        System.out.println(Thread.currentThread().getName() + " main()方法执行结束");
    }
}

运行结果

1
2

main main()方法执行结束
Thread-0 run()方法正在执行...

实现 Runnable 接口

1. 定义Runnable接口实现类MyRunnable，并重写run()方法
2. 创建MyRunnable实例myRunnable，以myRunnable作为target创建Thead对象，该Thread对象才是真正的线程对象
3. 调用线程对象的start()方法

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

public class MyRunnable implements Runnable {
    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName() + " run()方法执行中...");
    }
}
public class RunnableTest {
    public static void main(String[] args) {
        MyRunnable myRunnable = new MyRunnable();
        Thread thread = new Thread(myRunnable);
        thread.start();
        System.out.println(Thread.currentThread().getName() + " main()方法执行完成");
    }
}

执行结果

1
2

main main()方法执行完成
Thread-0 run()方法执行中...

实现 Callable 接口

1. 创建实现Callable接口的类myCallable
2. 以myCallable为参数创建FutureTask对象
3. 将FutureTask作为参数创建Thread对象
4. 调用线程对象的start()方法

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 class MyCallable implements Callable<Integer> {
    @Override
    public Integer call() {
        System.out.println(Thread.currentThread().getName() + " call()方法执行中...");
        return 1;
    }
}
public class CallableTest {
    public static void main(String[] args) {
        FutureTask<Integer> futureTask = new FutureTask<Integer>(new MyCallable());
        Thread thread = new Thread(futureTask);
        thread.start();

        try {
            Thread.sleep(1000);
            System.out.println("返回结果 " + futureTask.get());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        System.out.println(Thread.currentThread().getName() + " main()方法执行完成");
    }
}

执行结果

1
2
3

Thread-0 call()方法执行中...
返回结果 1
main main()方法执行完成

使用 Executors 工具类创建线程池

Executors提供了一系列工厂方法用于创先线程池，返回的线程池都实现了ExecutorService接口。

主要有newFixedThreadPool，newCachedThreadPool，newSingleThreadExecutor，newScheduledThreadPool，后续详细介绍这四种线程池

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

public class MyRunnable implements Runnable {
    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName() + " run()方法执行中...");
    }
}

public class SingleThreadExecutorTest {
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        MyRunnable runnableTest = new MyRunnable();
        for (int i = 0; i < 5; i++) {
            executorService.execute(runnableTest);
        }
        System.out.println("线程任务开始执行");
        executorService.shutdown();
    }
}

执行结果

1
2
3
4
5
6

线程任务开始执行
pool-1-thread-1 is running...
pool-1-thread-1 is running...
pool-1-thread-1 is running...
pool-1-thread-1 is running...
pool-1-thread-1 is running...

runnable 和 callable 的区别

相同点

• 都是接口
• 都可以编写多线程程序
• 都采用Thread.start()启动线程

主要区别

• Runnable 接口 run 方法无返回值；Callable 接口 call 方法有返回值，是个泛型，和Future、FutureTask配合可以用来获取异步执行的结果
• Runnable 接口 run 方法只能抛出运行时异常，且无法捕获处理；Callable 接口 call 方法允许抛出异常，可以获取异常信息

注：Callalbe接口支持返回执行结果，需要调用FutureTask.get()得到，此方法会阻塞主进程的继续往下执行，如果不调用不会阻塞。

线程的 run()和 start()的区别

每个线程都是通过某个特定Thread对象所对应的方法run()来完成其操作的，run()方法称为线程体。通过调用Thread类的start()方法来启动一个线程。

start() 方法用于启动线程，run() 方法用于执行线程的运行时代码。run() 可以重复调用，而 start() 只能调用一次。

start()方法来启动一个线程，真正实现了多线程运行。调用start()方法无需等待run方法体代码执行完毕，可以直接继续执行其他的代码； 此时线程是处于就绪状态，并没有运行。 然后通过此Thread类调用方法run()来完成其运行状态， run()方法运行结束， 此线程终止。然后CPU再调度其它线程。

run()方法是在本线程里的，只是线程里的一个函数，而不是多线程的。 如果直接调用run()，其实就相当于是调用了一个普通函数而已，直接待用run()方法必须等待run()方法执行完毕才能执行下面的代码，所以执行路径还是只有一条，根本就没有线程的特征，所以在多线程执行时要使用start()方法而不是run()方法。

为什么我们调用 start() 方法时会执行 run() 方法，为什么我们不能直接调用 run() 方法？

new 一个 Thread，线程进入了新建状态。调用 start() 方法，会启动一个线程并使线程进入了就绪状态，当分配到时间片后就可以开始运行了。 start() 会执行线程的相应准备工作，然后自动执行 run() 方法的内容，这是真正的多线程工作。

而直接执行 run() 方法，会把 run 方法当成一个 main 线程下的普通方法去执行，并不会在某个线程中执行它，所以这并不是多线程工作。

总结： 调用 start 方法方可启动线程并使线程进入就绪状态，而 run 方法只是 thread 的一个普通方法调用，还是在主线程里执行。

FutureTask

FutureTask 表示一个异步运算的任务。FutureTask 里面可以传入一个 Callable 的具体实现类，可以对这个异步运算的任务的结果进行等待获取、判断是否已经完成、取消任务等操作。只有当运算完成的时候结果才能取回，如果运算尚未完成 get 方法将会阻塞。一个 FutureTask 对象可以对调用了 Callable 和 Runnable 的对象进行包装，由于 FutureTask 也是Runnable 接口的实现类，所以 FutureTask 也可以放入线程池中。

为什么用线程池

1. 创建/销毁线程伴随着系统开销，过于频繁的创建/销毁线程，会很大程度上影响处理效率，应用能够更加充分合理地协调利用CPU、内存、网络、I/O等系统资源；
2. 线程并发数量过多，抢占系统资源从而导致阻塞，利用线程池管理并复用线程，控制最大并发数；
3. 使用线程池可以对线程进行一些简单的管理，实现任务线程队列缓存策略和拒绝机制，实现某些与时间相关的功能，例如定时执行、周期执行等，另外还可以隔离线程环境；

ThreadPoolExecutor的重要参数

corePoolSize：核心线程数

1
2
3

* 核心线程会一直存活，即使没有任务需要执行
* 当线程数小于核心线程数时，即使有线程空闲，线程池也会优先创建新线程处理
* 设置allowCoreThreadTimeout=true（默认false）时，核心线程会超时关闭

一般来说：

• CPU密集型：corePoolSize = CPU核数 + 1
• IO密集型：corePoolSize = CPU核数 * 2

maximumPoolSize：最大线程数

1
2

* 当线程数>=corePoolSize，且任务队列已满时。线程池会创建新线程来处理任务
* 当线程数=maxPoolSize，且任务队列已满时，线程池会拒绝处理任务而抛出异常

keepAliveTime：线程空闲时间

1
2

* 当线程空闲时间达到keepAliveTime时，线程会退出，直到线程数量=corePoolSize
* 如果allowCoreThreadTimeout=true，则会直到线程数量=0

unit：线程空闲时间的单位

1
2
3
4
5
6
7

* NANOSECONDS ： 1微毫秒 = 1微秒 / 1000
* MICROSECONDS ： 1微秒 = 1毫秒 / 1000
* MILLISECONDS ： 1毫秒 = 1秒 /1000
* SECONDS ： 秒
* MINUTES ： 分
* HOURS ： 小时
* DAYS ： 天

workQueue：用于缓存任务的阻塞队列

常用的workQueue类型：

• SynchronousQueue：这个队列接收到任务的时候，会直接提交给线程处理，而不保留它，如果所有线程都在工作怎么办？那就新建一个线程来处理这个任务！所以为了保证不出现<线程数达到了maximumPoolSize而不能新建线程>的错误，使用这个类型队列的时候，maximumPoolSize一般指定成Integer.MAX_VALUE，即无限大
• LinkedBlockingQueue：这个队列接收到任务的时候，如果当前线程数小于核心线程数，则新建线程(核心线程)处理任务；如果当前线程数等于核心线程数，则进入队列等待。由于这个队列没有最大值限制，即所有超过核心线程数的任务都将被添加到队列中，这也就导致了maximumPoolSize的设定失效，因为总线程数永远不会超过corePoolSize
• ArrayBlockingQueue：可以限定队列的长度，接收到任务的时候，如果没有达到corePoolSize的值，则新建线程(核心线程)执行任务，如果达到了，则入队等候，如果队列已满，则新建线程(非核心线程)执行任务，又如果总线程数到了maximumPoolSize，并且队列也满了，则发生错误
• DelayQueue：队列内元素必须实现Delayed接口，这就意味着你传进去的任务必须先实现Delayed接口。这个队列接收到任务时，首先先入队，只有达到了指定的延时时间，才会执行任务

handler：表示当workQueue已满，且池中的线程数达到maximumPoolSize时，线程池拒绝添加新任务时采取的策略。

1
2
3
4
5

* AbortPolicy 丢弃任务，抛运行时异常 （默认）
* CallerRunsPolicy 执行任务
* DiscardPolicy 忽视，什么都不会发生
* DiscardOldestPolicy 从队列中踢出最先进入队列（最后一个执行）的任务
* 实现RejectedExecutionHandler接口，可自定义处理器

threadFactory：指定创建线程的工厂

在阿里巴巴java开发手册以及一些sonar扫描规则中，需要手动创建线程池，和线程工厂，来对线程池进行命名。例如

1
2
3
4

private static final ThreadFactory NAMED_THREAD_FACTORY = new ThreadFactoryBuilder()
            .setNameFormat("thread-name-%d").build();
    private static final ExecutorService THREAD_POOL = new ThreadPoolExecutor(1, 1,
            0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), NAMED_THREAD_FACTORY);

默认常用线程池

虽然是默认常用，但是如果加了sonar规则的话这几个还不能直接用。。还是要用线程池的构造函数自己去构造。

newFixedThreadPool

创建固定线程数的线程池因为最大线程数和核心线程数相等，并且是无界队列，可控制线程最大并发数（同时执行的线程数），超出的线程会在队列中等待；

1
2
3
4
5

public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }

newSingleThreadExecutor

单任务队列的线程池,最大线程数和核心线程数都是1，无界队列，所有的任务都按照顺序进行执行；

1
2
3
4
5
6

public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }

ScheduledThreadPool

支持定时周期性执行任务的线程池；

1
2
3

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
       return new ScheduledThreadPoolExecutor(corePoolSize);
   }

CachedThreadPool

线程数无限制 有空闲线程则复用空闲线程，若无空闲线程则新建线程 一定程序减少频繁创建/销毁线程，减少系统开销.

1
2
3
4
5

public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

源码分析

任务的提交

submit方法源码

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

public Future<?> submit(Runnable task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<Void> ftask = newTaskFor(task, null);
    execute(ftask);
    return ftask;
}
 
public <T> Future<T> submit(Callable<T> task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<T> ftask = newTaskFor(task);
    execute(ftask);
    return ftask;
}
public <T> Future<T> submit(Runnable task, T result) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<T> ftask = newTaskFor(task, result);
        execute(ftask);
        return ftask;
}

这个方法一共有三个重载方法，分别是重写Callable和Runnable方法的重载。submit的实现方法位于抽象类AbstractExecutorService中，而此时execute方法还未实现（而是在AbstractExecutorService的继承类ThreadPoolExecutor中实现），可以看出无论哪个submit方法都最终调用了execute方法。

execute方法源码

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

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
 
    int c = ctl.get();
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
     
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        if (! isRunning(recheck) && remove(command))
            reject(command);
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    else if (!addWorker(command, false))
        reject(command);
}

简要分析一下execute源码，执行一个Runnable对象时，首先通过workerCountOf(c)获取线程池中线程的数量，如果池中的数量小于corePoolSize就调用addWorker添加一个线程来执行这个任务。否则通过workQueue.offer(command)方法入列。如果入列成功还需要在一次判断池中的线程数，因为我们创建线程池时可能要求核心线程数量为0，所以我们必须使用addWorker(null, false)来创建一个临时线程去阻塞队列中获取任务来执行。

第二次addWorker是在工作线程为0时调用的，我们现在假设此时是创建核心线程，即false改为true；那么addWorker方法中wc >= (core ? corePoolSize : maximumPoolSize)这个地方会去判断当前工作线程是否大于核心线程，在高并发的情况下，会存在其他线程将工作线程的数量创建的大于核心线程数，导致返回false，并且不会创建新线程，虽然有工作线程的存在，但是会导致原本可以及时处理的任务，要去排队执行。

isRunning(c) 的作用是判断线程池是否处于运行状态，如果入列后发现线程池已经关闭，则出列。不需要在入列前判断线程池的状态，因为判断一个线程池工作处于RUNNING状态到执行入列操作这段时间，线程池可能被其它线程关闭了，所以提前判断毫无意义。

其中addWorker方法就是创建一个线程来执行Runnable对象

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

private boolean addWorker(Runnable firstTask, boolean core) {
        retry:
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);

            // Check if queue empty only if necessary.
            //判断线程池的是否可以接收新任务
            if (rs >= SHUTDOWN &&
                ! (rs == SHUTDOWN &&
                   firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;

            for (;;) {
            	//获取工作线程的数量
                int wc = workerCountOf(c);
                //判断工作线程的数量是否大于等于线程池的上限或者核心或者最大线程数
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                //使用cas增加工作线程数
                if (compareAndIncrementWorkerCount(c))
                    break retry;
                c = ctl.get();  // Re-read ctl
                //如果添加失败，并且线程池状态发生了改变，重来一遍
                if (runStateOf(c) != rs)
                    continue retry;
                // else CAS failed due to workerCount change; retry inner loop
            }
        }
		//上面的逻辑是考虑是否能够添加线程，如果可以使用cas来增加工作线程数量
		//下面正式启动线程
        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
        	//新建worker
            w = new Worker(firstTask);
            // 获取当前线程
            final Thread t = w.thread;
            if (t != null) {
            	//获取重入锁
                final ReentrantLock mainLock = this.mainLock;
                //锁住
                mainLock.lock();
                try {
                    // Recheck while holding lock.
                    // Back out on ThreadFactory failure or if
                    // shut down before lock acquired.
                    int rs = runStateOf(ctl.get());
					// rs < SHUTDOWN  -- 状态即为：RUNNING
					//rs == SHUTDOWN && firstTask == null 
                    if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) {
                        //如果线程已经启动，抛出异常
                        if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();
                        //workers是一个HashSet，必须在锁住的情况下，操作
                        workers.add(w);
                        int s = workers.size();
                        //设置largestPoolSize ，标记workerAdded 
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        workerAdded = true;
                    }
                } finally {
                    mainLock.unlock();
                }
                //如果添加成功，启动线程
                if (workerAdded) {
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
        	//启动线程失败,回滚
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }

第一个参数firstTask不为null，则创建的线程就会先执行firstTask对象，然后去阻塞队列中取任务，否直接到阻塞队列中获取任务来执行。第二个参数，core参数为真，则用corePoolSize作为池中线程数量的最大值；为假，则以maximumPoolSize作为池中线程数量的最大值。

任务的执行

runWorker源码

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

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
        while (task != null || (task = getTask()) != null) {
            w.lock();
            // If pool is stopping, ensure thread is interrupted;
            // if not, ensure thread is not interrupted.  This
            // requires a recheck in second case to deal with
            // shutdownNow race while clearing interrupt
            if ((runStateAtLeast(ctl.get(), STOP) ||
                 (Thread.interrupted() &&
                  runStateAtLeast(ctl.get(), STOP))) &&
                !wt.isInterrupted())
                wt.interrupt();
            try {
                beforeExecute(wt, task);
                Throwable thrown = null;
                try {
                    task.run();
                } catch (RuntimeException x) {
                    thrown = x; throw x;
                } catch (Error x) {
                    thrown = x; throw x;
                } catch (Throwable x) {
                    thrown = x; throw new Error(x);
                } finally {
                    afterExecute(task, thrown);
                }
            } finally {
                task = null;
                w.completedTasks++;
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        processWorkerExit(w, completedAbruptly);
    }
}

Thread的run方法实际上调用了Worker类的runWorker方法，而Worker类继承了AQS类，并实现了lock、unlock、trylock方法。但是这些方法不是真正意义上的锁，所以在代码中加锁操作和解锁操作没有成对出现。runWorker方法中获取到任务就“加锁”，完成任务后就“解锁”。也就是说在“加锁”到“解锁”的这段时间内，线程处于忙碌状态，而其它时间段，处于空闲状态。线程池就可以通过trylock方法来确定这个线程是否空闲。

getTask方法的主要作用是从阻塞队列中获取任务。

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

   beforeExecute(wt, task)和afterExecute(task, thrown)是个钩子函数，如果我们需要在任务执行之前和任务执行以后进行一些操作，那么我们可以自定义一个继承ThreadPoolExecutor类，并覆盖这两个方法。
private Runnable getTask() {
    boolean timedOut = false; // Did the last poll() time out?
 
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);
 
        // Check if queue empty only if necessary.
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }
 
        int wc = workerCountOf(c);
 
        // Are workers subject to culling?
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
 
        if ((wc > maximumPoolSize || (timed && timedOut))
            && (wc > 1 || workQueue.isEmpty())) {
            if (compareAndDecrementWorkerCount(c))
                return null;
            continue;
        }
 
        try {
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}

线程池的关闭

shutdown源码

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

public void shutdown() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();
        advanceRunState(SHUTDOWN);
        interruptIdleWorkers();
        onShutdown(); // hook for ScheduledThreadPoolExecutor
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
}

advanceRunState(SHUTDOWN)的作用是通过CAS操作将线程池的状态更改为SHUTDOWN状态。
interruptIdleWorkers是对空闲的线程进行中断，它实际上调用了重载带参数的函数interruptIdleWorkers(false)

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

private void interruptIdleWorkers(boolean onlyOne) {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        for (Worker w : workers) {
            Thread t = w.thread;
            if (!t.isInterrupted() && w.tryLock()) {
                try {
                    t.interrupt();
                } catch (SecurityException ignore) {
                } finally {
                    w.unlock();
                }
            }
            if (onlyOne)
                break;
        }
    } finally {
        mainLock.unlock();
    }
}

通过workers容器，遍历池中的线程，对每个线程进行tryLock()操作，如果成功说明线程空闲，则设置其中断标志位。而线程是否响应中断则由任务的编写者决定。

上次说了code review 在开发中的重要性。这次说一说怎么才能让 code review 有效发挥它的作用。

使用方便的工具。要在团队中成功推行 code review，有好的工具是很重要的。谁都不愿意在工作中加入繁琐的过程，所以如果要让人接受一个新的步骤，最好让它有尽可能好的体验。在我用过的开源 code review 工具中，Phabricator 是最完善的。Review Board 应该也是不错的选择，只是我很久没实际用了，不知道最新状态。GitHub 在支持 side-by-side diff 后，它的 pull request 也提供了很好的 code review 体验，唯一不足的是无法从机制上强制所有 commit 都经过 pull request 的过程。

把简单的检查自动化。有很多检查是可以自动化的，比如一些风格规范（缩进、空行、行尾空格、命名等），这类问题应该尽可能写脚本检查。一方面可以让负责 review 的工程师把更多注意力放在更高层面的问题上；另一方面，从接受 review 的人心理上说，由程序提出这些细节问题比让另一个人来挑刺要更容易接受些。当然，这并不是说人工 review 的时候应该忽略细节问题。

控制每次 review 的代码量。每次 review 包含 200 行左右代码是比较理想的，最多不要超过 400 行。因为如果代码太多，review 的人容易因为注意力分散而忽略一些问题，另外也可能让时间拖得过长。因此，开发的时候需要把大的改动分解成多个小的步骤，每完成一个步骤就提交一次 review。

使用异步的工作流。这和上一点是相关的，当你需要把一个改动分为相互依赖的多步时，不应该因为等待 review 而 block 住自己的工作。在当前分支等待 review 时，可以从这个点开一个新的分支继续开发，之后再把 review 完的分支 merge 进来。

作者应该提供清晰的 commit note。Code review 的重要作用之一是同事间的交流，每个 commit 的 commit note 很重要也很影响 review 的效率，应该包含这个改动的目的，以及实现方式的概述。如果使用的 review 系统支持对 review 本身的描述（如 GitHub 的 pull request），那么应该写清楚作者希望 reviewer 重点关注的问题。

在合并到主干之前进行 review。有的人主张小团队应该做事后的 review，因为这样效率更高、更「轻量」。事实上，这样并不会减少工作量，并且如果工作流安排合理，事前 review 并不会导致效率降低。从心理上说，代码并合并到主干之后往往就意味着「完事了」，在时间比较紧张的时候很难坚持 review 所有代码。所以应该把 code review 作为代码合并到主干前必过的一道关口。

所有人的代码都要经过 review。Code review 并不是资深工程师对初级工程师做的事情，而应该全员平等参与，每个人都会有所收获。关于这方面在上一篇「为什么每个团队都需要 Code Review」里也有提及。

关注设计方面的问题和客观的规范，避免在主观意见上争执。Code review 的讨论应该专注于设计层面的问题以及在团队中有明确共识的规范（代码风格，测试覆盖等），而要避免在一些主观意见的分歧上浪费时间。这也意味着应该在一些本身不重要，但很影响一致性的细节上尽早达成有共识的规范，如缩进方式，括号的位置等等，避免在 review 中去争论这些问题。

从正面看待在 review 中发现的问题。当发现一个错误时，并不应该看作是一个人犯了错被另一个人发现了，而是两个人配合改进了代码、避免了错误。

Review 所有的代码，包括很简单的改动。每一个 commit，无论多小都应该经过 review。一方面，很难界定什么算简单的改动，如果一行的 commit 不用 review，那两行是不是差不多同样简单，三行呢？另一方面，我已经不记得自己曾有多少次觉得「这个改动太简单，不用跑测试就可以提交了」，然后很快因为测试通不过而很尴尬地被别人 rollback。所以无论多小的改动都应该有测试、都应该通过 review。

循序渐进。如果现在完全没有做 code review，那么到整个团队都能严格有效地进行 code review 是一个循序渐进的过程。如果你所处的角色无法很快改变团队的工作方式，那么可以从把自己的代码发给别人 review 开始，实际可见的价值是最有说服力的。

不少开发团队和创业公司都在纠结是否要执行 code review，既希望改进代码质量，又担心带来的负担会拖慢项目进度。实时上，在软件开发中质量和效率往往并不是二选其一的关系。能产出高质量代码的团队通常效率也非常高。

在我作为工程师的职业经历中，自动化测试和 code review 可说是能同时提高代码质量和开发效率的两个最有效的手段。所谓 code review，和学术界的 peer review 类似。Peer review 是由同事或同行对一位作者的作品进行查阅并提出建议和问题，只有当所有提出的问题都得到满意的答案后，作品才能发表。对于 code review 来说，作品就是代码，发表就是把代码 commit 到官方代码库。

在 Code Complete 这本书中讲述了两个很有说服力的案例。在一项对同一个团队开发的很多个程序进行对比的研究中，没有经过 review 的程序平均每 100 行有 4.5 个错误，而经过 review 的程序平均每 100 行只有 0.82 个错误，也就是说 80% 的错误在 review 中被修正了。AT & T 的一个 200 多人的部门在开始执行 code review 后，开发效率提高了 14%，而错误减少了 90% 左右。

除了减少缺陷，避免在诊断错误上浪费时间，review 的过程还可以通过相互的督促保证代码有好的可读性、文档、风格，并同时检查测试覆盖率等开发过程中的规范，从而提高团队的协作效率。对于所有复杂的事情来说，总是越早发现问题，解决问题的成本越低。

对于经验不足或者刚开始一份新工作的人来说，通过 code review 可以得到更资深的人帮助，更快熟悉现有的规范和架构，在新的环境和团队中快速提升。

对于资深的工程师来说，让其他同事 review 代码，有利于在团队中传播经验、知识和好的实践。身边的同事水平提高会让自己的工作也更高效。并且谁都有需要休假的时候，无论是公司还是个人都不希望有太多工作因此而停滞，如果有平时就熟悉自己工作的同事，这个问题就很好解决。

像很多其他事情一样，code review 最难的就是迈出第一步。一旦开始，花在 review 过程的每一分钟都会很快被成倍地赚回来。如果你不在一个可以一下改变团队流程的位置上，那么至少可以和认同这件事的少数同事先开始实践，当价值开始体现的时候，相信其他人会乐于效仿。

什么是git

Git是免费、开源的 分布式版本控制 系统，可以有效、高速地处理从很小到非常大的项目版本管理。

版本控制

版本控制是指对软件开发过程中各种程序代码、配置文件及说明文档等文件变更的管理，是软件配置管理的核心思想之一。那些年，我们的毕业论文,其实就是版本变更的真实写照…脑洞一下，版本控制就是这些论文变更的管理

集中化的版本控制系统

那么，集中化的版本控制系统又是什么呢，说白了，就是有一个集中管理的中央服务器，保存着所有文件的修改历史版本，而协同开发者通过客户端连接到这台服务器，从服务器上同步更新或上传自己的修改。

分布式的版本控制系统

分布式版本控制系统，就是远程仓库同步所有版本信息到本地的每个用户。这里分三点阐述：

• 用户在本地就可以查看所有的历史版本信息，但是偶尔要从远程更新一下，因为可能别的用户有文件修改提交到远程。
• 用户即使离线也可以本地提交，push推送到远程服务器才需要联网。
• 每个用户都保存了历史版本，所以只要有一个用户设备没问题，就可以恢复数据了

Git的相关理论基础

Git的四大工作区域

先复习Git的几个工作区域：

• Workspace ：你电脑本地看到的文件和目录，在Git的版本控制下，构成了工作区。
• Index/Stage ：暂存区，一般存放在 .git目录下，即.git/index,它又叫待提交更新区，用于临时存放你未提交的改动。比如，你执行git add，这些改动就添加到这个区域啦。
• Repository ：本地仓库，你执行git clone 地址，就是把远程仓库克隆到本地仓库。它是一个存放在本地的版本库，其中 HEAD指向最新放入仓库的版本 。当你执行git commit，文件改动就到本地仓库来了~
• Remote ：远程仓库，就是类似github，码云等网站所提供的仓库，可以理解为远程数据交换的仓库

Git的工作流程

上一小节介绍完Git的四大工作区域，把git的操作命令和几个工作区域结合起来，个人觉得更容易理解一些，看图：

git 的正向工作流程一般就这样：

• 从远程仓库拉取文件代码回来；
• 在工作目录，增删改查文件；
• 把改动的文件放入暂存区；
• 将暂存区的文件提交本地仓库；
• 将本地仓库的文件推送到远程仓库；

Git文件的四种状态

根据一个文件是否已加入版本控制，可以把文件状态分为：Tracked(已跟踪)和Untracked(未跟踪)，而tracked(已跟踪)又包括三种工作状态：Unmodified，Modified，Staged

• Untracked : 文件还没有加入到git库，还没参与版本控制，即未跟踪状态。这时候的文件，通过git add 状态，可以变为Staged状态
• Unmodified ：文件已经加入git库, 但是呢，还没修改, 就是说版本库中的文件快照内容与文件夹中还完全一致。Unmodified的文件如果被修改, 就会变为Modified. 如果使用git remove移出版本库, 则成为Untracked文件。
• Modified ：文件被修改了，就进入modified状态啦，文件这个状态通过stage命令可以进入staged状态
• staged ：暂存状态. 执行git commit则将修改同步到库中, 这时库中的文件和本地文件又变为一致, 文件为Unmodified状态.

git的常用命令

日常开发中，Git的基本常用命令

• git clone
• git checkout -b dev
• git add
• git commit
• git log
• git diff
• git status
• git pull/git fetch
• git push

这个图只是模拟一下git基本命令使用的大概流程

git clone

当我们要进行开发，第一步就是克隆远程版本库到本地

git clone url  克隆远程版本库  
  

git checkout -b dev

克隆完之后呢，开发新需求的话，我们需要新建一个开发分支，比如新建开发分支dev

创建分支：

git checkout -b dev   创建开发分支dev，并切换到该分支下  

git add

git add的使用格式：

git add .	添加当前目录的所有文件到暂存区  
git add [dir]	添加指定目录到暂存区，包括子目录  
git add [file1]	添加指定文件到暂存区  

有了开发分支dev之后，我们就可以开始开发啦，假设我们开发完HelloWorld.java，可以把它加到暂存区，命令如下

git add Hello.java  把HelloWorld.java文件添加到暂存区去  

git commit

git commit的使用格式：

git commit -m [message] 提交暂存区到仓库区,message为说明信息  
git commit [file1] -m [message] 提交暂存区的指定文件到本地仓库  
git commit --amend -m [message] 使用一次新的commit，替代上一次提交  
  

把HelloWorld.java文件加到暂存区后，我们接着可以提交到本地仓库

git commit -m 'helloworld开发'  
  

git status

git status,表示查看工作区状态，使用命令格式：

git status  查看当前工作区暂存区变动  
git status -s  查看当前工作区暂存区变动，概要信息  
git status  --show-stash 查询工作区中是否有stash（暂存的文件）  

当你忘记是否已把代码文件添加到暂存区或者是否提交到本地仓库，都可以用git status看看

git log

git log，这个命令用得应该比较多，表示查看提交历史/提交日志，要回滚代码就经常用它看看提交历史。

git log  查看提交历史  
git log --oneline 以精简模式显示查看提交历史  
git log -p <file> 查看指定文件的提交历史  
git blame <file> 一列表方式查看指定文件的提交历史  

git diff

git diff 显示暂存区和工作区的差异  
git diff filepath   filepath路径文件中，工作区与暂存区的比较差异  
git diff HEAD filepath 工作区与HEAD ( 当前工作分支)的比较差异  
git diff branchName filepath 当前分支的文件与branchName分支的文件的比较差异  
git diff commitId filepath 与某一次提交的比较差异  

如果你想对比一下你改了哪些内容，可以用git diff对比一下文件修改差异

git pull/git fetch

git pull  拉取远程仓库所有分支更新并合并到本地分支。  
git pull origin master 将远程master分支合并到当前本地分支  
git pull origin master:master 将远程master分支合并到当前本地master分支，冒号后面表示本地分支  
git fetch --all  拉取所有远端的最新代码  
git fetch origin master 拉取远程最新master分支代码  

我们一般都会用git pull拉取最新代码看看的，解决一下冲突，再推送代码到远程仓库的。

有些伙伴可能对使用git pull还是git fetch有点疑惑，其实 git pull = git fetch+ git
merge。pull的话，拉取远程分支并与本地分支合并，fetch只是拉远程分支，怎么合并，可以自己再做选择。

git push

git push 可以推送本地分支、标签到远程仓库，也可以删除远程分支。

git push origin master 将本地分支的更新全部推送到远程仓库master分支。  
git push origin -d <branchname>   删除远程branchname分支  
git push --tags 推送所有标签  

如果我们在dev开发完，或者就想把文件推送到远程仓库，给别的伙伴看看，就可以使用git push origin dev

git branch

git branch用处多多呢，比如新建分支、查看分支、删除分支等等

新建分支：

git checkout -b dev2  新建一个分支，并且切换到新的分支dev2  
git branch dev2 新建一个分支，但是仍停留在原来分支  

查看分支：

git branch    查看本地所有的分支  
git branch -r  查看所有远程的分支  
git branch -a  查看所有远程分支和本地分支  

删除分支：

git branch -D <branchname>  删除本地branchname分支  

git checkout

切换分支：

git checkout master 切换到master分支  

git merge

我们在开发分支dev开发、测试完成在发布之前，我们一般需要把开发分支dev代码合并到master，所以git merge也是程序员必备的一个命令。

git merge master  在当前分支上合并master分支过来  
git merge --no-ff origin/dev  在当前分支上合并远程分支dev  
git merge --abort 终止本次merge，并回到merge前的状态  

比如，你开发完需求后，发版需要把代码合到主干master分支

Git进阶之撤销与回退

Git的撤销与回退，在日常工作中使用的比较频繁。比如我们想将某个修改后的文件撤销到上一个版本，或者想撤销某次多余的提交，都要用到git的撤销和回退操作。

代码在Git的每个工作区域都是用哪些命令撤销或者回退的呢，如下图所示：

有关于Git的撤销与回退，一般就以下几个核心命令

• git checkout
• git reset
• git revert

git checkout

如果文件还在 工作区 ，还没添加到暂存区，可以使用git checkout撤销

git checkout [file]  丢弃某个文件file  
git checkout .  丢弃所有文件  

git reset

git reset的理解

git reset的作用是修改HEAD的位置，即将HEAD指向的位置改变为之前存在的某个版本.

为了更好地理解git reset，我们来回顾一下,Git的版本管理及HEAD的理解

Git的所有提交，会连成一条时间轴线，这就是分支。如果当前分支是master，HEAD指针一般指向当前分支，如下：

假设执行git reset，回退到版本二之后，版本三不见了,如下：

git reset的使用

Git Reset的几种使用模式

git reset HEAD --file回退暂存区里的某个文件，回退到当前版本工作区状态  
git reset –-soft 目标版本号 可以把版本库上的提交回退到暂存区，修改记录保留  
git reset –-mixed 目标版本号 可以把版本库上的提交回退到工作区，修改记录保留  
git reset –-hard  可以把版本库上的提交彻底回退，修改的记录全部revert。  

先看一个栗子demo吧，代码 git add到暂存区，并未commit提交 ,可以酱紫回退，如下：

git reset HEAD file 取消暂存  
git checkout file 撤销修改  

再看另外一个栗子吧，代码已经git commit了，但是还没有push：

git log  获取到想要回退的commit_id  
git reset --hard commit_id  想回到过去，回到过去的commit_id  

如果代码已经push到远程仓库了呢，也可以使用reset回滚

git log  
git reset --hard commit_id  
git push origin HEAD --force  

git revert

与git reset不同的是，revert复制了那个想要回退到的历史版本，将它加在当前分支的最前端。

revert之前：

revert 之后：

当然，如果代码已经推送到远程的话，还可以考虑revert回滚

git log  得到你需要回退一次提交的commit id  
git revert -n <commit_id>  撤销指定的版本，撤销也会作为一次提交进行保存  

git rebase

rebase又称为衍合，是合并的另外一种选择。

假设有两个分支master和test

     D---E test        
    /  
A---B---C---F--- master  
  

执行 git merge test得到的结果

      D--------E      
     /          \  
A---B---C---F----G---   test, master  
  

执行git rebase test，得到的结果

A---B---D---E---C‘---F‘---   test, master  
  

rebase好处是： 获得更优雅的提交树，可以线性的看到每一次提交，并且没有增加提交节点。所以很多时候，看到有些伙伴都是这个命令拉代码：git
pull –rebase，就是因为想更优雅，哈哈

git stash

stash命令可用于临时保存和恢复修改

git stash  把当前的工作隐藏起来 等以后恢复现场后继续工作  
git stash list 显示保存的工作进度列表  
git stash pop stash@{num} 恢复工作进度到工作区  
git stash show ：显示做了哪些改动  
git stash drop stash@{num} ：删除一条保存的工作进度  
git stash clear 删除所有缓存的stash。