JUC并发—3.volatile和synchronized原理
大纲
1.volatile关键字的使用例子
2.主内存和CPU的缓存模型
3.CPU高速缓存的数据不一致问题
4.总线锁和缓存锁及MESI缓存一致性协议
5.Java的内存模型JMM
6.JMM如何处理并发中的原子性可见性有序性
7.volatile如何保证可见性
8.volatile为什么无法保证原子性
9.volatile如何保证有序性
10.volatile的原理(Lock前缀指令 + 内存屏障)
11.双重检查单例模式的volatile优化
12.基于volatile优化微服务的优雅关闭机制
13.基于volatile优化微服务存活状态检查机制
14.i++的多线程安全问题演示
15.JMM是多线程并发安全问题的根本原因
16.synchronized可解决多线程并发安全问题
17.synchronized的常见使用方法总结
18.synchronized的底层原理
19.微服务定时拉取注册表信息
20.基于synchronized解决注册表并发问题
21.微服务注册中心的自我保护机制
22.基于synchronized实现服务心跳计数器
23.微服务关闭时的服务下线实现
24.基于synchronized修改触发自我保护阈值
25.基于synchronized开启自我保护机制
1.volatile关键字的使用例子
(1)volatile关键字的使用场景
如果多个线程共用一个共享变量，有的线程写、有的线程读，那么可能会导致有的线程没法及时读到其他线程修改的变量值。volatile关键字可让某线程修改变量后，其他线程立即看到该变量的修改值。
(2)volatile关键字的使用例子
public class VolatileDemo {
//如果没有volatile，那么读取的线程可能一直读到旧值而不打印，CPU缓存模型的数据一致性导致的
static volatile int flag = 0;
public static void main(String[] args) {
new Thread() {
public void run() {
int localFlag = flag;
while(true) {
if (localFlag != flag) {
System.out.println("读取到了修改后的标志位：" + flag);
localFlag = flag;
}
}
};
}.start();
new Thread() {
public void run() {
int localFlag = flag;
while(true) {
System.out.println("标志位被修改为了：" + ++localFlag);
flag = localFlag;
try {
TimeUnit.SECONDS.sleep(2);
} catch (Exception e) {
e.printStackTrace();
}
}
};
}.start();
}
}
(3)volatile关键字的理解路径
一.CPU缓存模型
二.Java内存模型JMM
三.原子性、可见性、有序性
四.volatile的作用
五.volatile的底层原理
六.volatile案例
2.主内存和CPU的缓存模型
CPU如果频繁读写主内存，那么就会导致CPU的计算性能比较差。所以现代的计算机，一般都会在CPU和内存之间加几层高速缓存。这样每个CPU就可以直接操作自己对应的高速缓存，从而不需要直接和主内存进行频繁的通信，保证了CPU的计算效率。
3.CPU高速缓存的数据不一致问题
主内存的数据会被加载到CPU高速缓存里，CPU后续会读写自己的高速缓存。但多线程并发运行时，就可能引发各个CPU高速缓存里的数据不一致问题。
上面volatile的例子，在没有volatile修饰flag的时候：负责执行线程0的CPU一开始会将主内存的flag值读到其高速缓存。之后在执行线程0的指令时，便会在该CPU的高速缓存里读取flag的值。此时该CPU无法感知负责执行线程1的其他CPU对其高速缓存的flag的修改。因为负责执行线程1的CPU可能没有将其修改的缓存值及时刷新回到主内存，或者负责执行线程0的CPU可能没有主动更新主内存的最新值到其高速缓存中。
所以CPU的缓存模型，在多线程并发运行时可能存在数据不一致的问题。也就是各个CPU的高速缓存和主内存没有及时同步，同一个数据在各CPU可能都不一样，从而导致数据的不一致。
4.总线锁和缓存锁及MESI缓存一致性协议
(1)总线锁和缓存锁机制
(2)MESI缓存一致性协议
(3)CPU、高速缓存、内存之间的关系总结
(1)总线锁和缓存锁机制
一.什么是总线
所谓总线，就是CPU与内存和输入输出设备传输信息的公共通道。当CPU和内存进行数据交互时，必须经过总线来传输。
二.总线锁
所谓总线锁就是：如果某个CPU要修改主内存的某数据，那么就往总线发出一个Lock#信号。这个信号能够确保主内存只有该CPU可以访问，其他CPU的请求会被阻塞。这就使得同一时刻只有一个CPU能够访问主内存，从而解决缓存不一致问题。
所以总线锁可以理解为：当一个CPU往总线发出一个Lock#信号时，其他CPU的的请求将会被阻塞，于是该CPU就能独占主内存(共享内存)了。
总线锁把CPU和内存之间的通信锁住了，从而使得锁定期间，其他CPU不能操作其他内存地址的数据。所以总线锁虽然解决了缓存不一致的问题，但却大幅降低了CPU的利用率，于是CPU使用缓存锁来替代总线锁。
三.缓存锁
如果当前CPU访问的数据已经缓存在其他CPU的高速缓存中，那么当前CPU便不会在总线上发出一个Lock#信号，而是采用MESI缓存一致性协议来保证多个CPU的缓存一致性。
(2)MESI缓存一致性协议
该协议要求每个CPU都可以监听到总线上的数据事件并做出相应的处理。当某个CPU向总线发出请求时，其他CPU便能监听到总线收到的请求，从而可以根据当前缓存行的状态和监听的请求类型来更新缓存行状态。这其实也就是所谓的CPU嗅探机制。
这样MESI缓存一致性协议就能保证，在CPU的缓存模型下，就不会出现多线程并发读写变量，各CPU没有办法及时感知到的问题，也就是解决了CPU缓存的一致性问题。
(3)CPU、高速缓存、内存之间的关系总结
一.高速缓存可解决CPU与内存的速度矛盾
为了解决CPU与内存速度之间的矛盾，引入了高速缓存作为内存与CPU之间的缓冲。
这里的高速缓存其实就是三级缓存。每个CPU可能有多个物理核，每个物理核会有多个逻辑核。每个CPU都有自己的三级缓存，每个物理核都有自己的一二级缓存。
二.每个CPU都有自己的高速缓存
每个CPU都有自己的高速缓存，而它们又共享同一个主内存。当多个CPU的运算任务都涉及到同一块主内存区域时，将可能导致各自的高速缓存的数据不一致。为了解决这种缓存数据不一致，引入了如MESI这些缓存一致性协议(嗅探)。
三.CPU的乱序执行优化
为了使CPU内部的运算单元能尽量被充分利用，CPU可能会对输入代码进行乱序执行优化。与CPU的乱序执行优化类似，JVM的即时编译中也有指令重排序优化。
5.Java的内存模型JMM
(1)Java内存模型JMM简介
(2)主内存与工作内存的关系
(3)主内存与工作内存的交互
(4)对于volatile变量的特殊规则
(5)针对long和double型变量的特殊规则
(1)Java内存模型JMM简介
Java内存模型是用来屏蔽各种硬件和操作系统的内存访问差异的，以实现让Java程序在各种平台下都能达到一致的内存访问效果。
Java内存模型JMM的具体内容包括如下：
一.主内存与工作内存的关系
二.主内存与工作内存之间的交互
三.对于volatile变量的特殊规则
四.针对long和double型变量的特殊规则
五.原子性、可见性和有序性
六.Happens-Before原则(先行发生原则)
(2)主内存与工作内存的关系
一.JMM规定所有共享变量都存储在主内存
这里的共享变量只包括实例字段、静态字段和构成数组对象的元素。但不包括局部变量与方法参数，因为这些是线程私有的，不会共享。
二.每个线程都有自己的工作内存
这里的工作内存可以理解为各个CPU上的高速缓存，线程的工作内存中保存了被该线程使用的变量的主内存副本。线程对变量的读写操作必须在工作内存中进行，不能直接读写主内存的数据。线程之间无法直接访问对方工作内存中的变量，线程之间变量值的传递均需要通过主内存来完成。
三.主内存和工作内存分别存堆栈数据
主内存对应于Java堆中对象的实例数据，工作内存对应于虚拟机栈中的部分区域。对象包括对象头、实例数据、对象填充。
(3)主内存与工作内存的交互
JMM定义了8种操作来完成：一个变量如何从主内存拷贝到工作内存，如何从工作内存同步回主内存。这8种操作都是原子的、不可再分的。
一.read：读取主内存的变量值并传输到线程的工作内存中，配合load使用
二.load：把read操作从主内存读取到的变量值写入工作内存的变量副本中
三.use：从工作内存中读取出数据，然后交给执行引擎进行计算
四.assign：将执行引擎计算好的值赋值给工作内存中的变量
五.store：把工作内存中的变量值传输回主内存中，配合write使用
六.write：把store操作从工作内存中得到的变量值写入主内存的变量中
七.lock：作用于主内存中的变量，用来标识变量被某个线程独占
八.unlock：作用于主内存中的变量，用来释放锁定状态
在Java内存模型下，多线程并发运行依然存在CPU高速缓存不一致问题。线程1修改了flag之后write回主内存，线程2也还是没法感知到flag已修改。
(4)对于volatile变量的特殊规则
volatile变量对所有线程是立即可见的，对volatile变量的所有写操作都能立刻反映到其他线程之中，volatile变量在各个线程的工作内存中是不存在一致性问题的。
从物理存储角度看，各线程的工作内存的volatile变量也可存在不一致的情况。但由于各线程每次使用volatile变量前都刷新，执行引擎看不到不一致的情况。因此可以认为不存在不一致的问题。
volatile变量是禁止指令重排序优化的。
(5)针对long和double型变量的特殊规则
long合double的非原子性协定：允许虚拟机将没有被volatile修饰的64位数据的读写操作，划分为两次32位的操作来进行。即允许虚拟机自行选择是否要保证64位数据类型的：read、load、store和write四个操作的原子性。
如果有多个线程共享一个并未声明位volatile的long或double类型的变量，且同时对它们进行读取和修改，那么某些线程可能会读取到一个既不是原值，也不是其他线程修改值的"半个变量"的值。这种情况很罕见，64位的JVM不会出现，但32位的JVM有可能出现。
6.JMM如何处理并发中的原子性可见性有序性
(1)并发过程中可能产生的三类问题
(2)JMM如何处理原子性
(3)JMM如何处理可见性
(4)JMM如何处理有序性
(1)并发过程中可能产生的三类问题
一.原子性问题
对于一行代码"n++"，只要多个线程并发执行，都不保证该操作是原子性的。如果保证了该自增操作的原子性，那么下图线程1的i++为1，线程2的i++为2。
二.可见性问题
线程1修改完了主内存的某个变量值，线程2一直读取的是CPU高速缓存中的该变量值，线程1修改完的该变量值对线程2不可见。
三.有序性问题
有序性指的是程序按照代码的先后顺序执行。而编译器为了优化性能，有时候会改变程序中语句的先后顺序。即编译器和指令器有时为了提高代码执行效率，会将指令进行重排序。比如"a=1;b=2;"，编译器优化后可能就变成了"b=7;a=6"。此时编译器虽然调整了语句的顺序，但不会影响最终结果。
例子一：
假设有两个线程，线程1先执行init()方法，线程2后执行doxxx()方法。由于init()方法的操作1和操作2没有依赖关系，所以编译器可能会对其重排序。经过编译器的重排序后，线程1可能会先执行操作2，然后再执行操作1。这时线程2在while循环中就会发现flag = true，于是执行execute()方法。但此时init()方法的prepare()还没执行完，从而引发代码逻辑异常。
public class ReOrderExample {
boolean flag = false;
//线程1先执行init()方法
public void init() {
//准备资源
prepare();//操作1
flag = true;//操作2
}
//线程2后执行doxxx()方法
public void doxxx() {
while(!flag) {
Thread.sleep(1000);
}
execute();//基于准备好的资源执行操作
}
private void prepare() {
...
}
//prepare()执行完才能保证execute()方法的逻辑执行正确
private void execute() {
...
}
}
例子二：
在下面利用双重检查创建单例的代码中，Singleton.getInstance()方法会先判断inst是否为空。如果为空则锁定Singleton.class并再次检查inst是否为空，如果还为空那么就创建一个Singleton的对象实例。
其中"inst = new Singleton()"会执行三个指令：
指令1：分配对象的内存空间
指令2：初始化对象
指令3：设置inst变量指向指令1分配的内存地址
如果按照正常顺序来执行，那么是不会有问题的。但是由于指令2和指令3不存在依赖关系，所以编译器优化后可能进行重排序。于是执行顺序变为：指令1 -> 指令3 -> 指令2。那么就可能发生：在线程A刚把inst指向对应地址后，线程B获取到执行权。然后线程B便获取到一个没有初始化的对象，从而产生空指针异常。
public class Singleton {
private static Singleton inst;
private Singleton() {
}
public static Singleton getInstance() {
if (inst == null) {
synchronized(Singleton.class) {
if (inst == null) {
//开辟空间，inst指向地址，初始化
inst = new Singleton();
}
}
}
return inst;
}
}
(2)JMM如何处理原子性
由JMM来直接保证的原子性变量操作包括：read、load、assign、use、store和write六个。
基本数据类型的访问、读写都是原子性的，例外就是long和double的非原子性协定。
JMM还提供了lock和unlock操作来满足更大范围的原子性保证。这是通过字节码指令monitorenter和monitorexit来隐式地使用这两个操作的，这两个字节码指令反映到Java代码中就是同步块(synchronized修饰的代码)。
(3)JMM如何处理可见性
可见性就是指当一个线程修改了共享变量的值，其他线程能立即得知该修改。
JMM是通过在变量被修改后将新值同步回主内存，在变量被读取前从主内存刷新变量值的方式来实现可见性的，无论普通变量还是volatile变量都是如此。
普通变量和volatile变量的区别是：volatile保证了新值能立即同步到主内存，以及每次使用前立即从主内存刷新。也就是volatile保证了多线程操作时变量的可见性，而普通变量则不能保证。
除了volatile关键字，synchronized和final两个关键字也能实现可见性。
synchronized的可见性是通过如下这条规则获得的：对一个变量执行unlock操作之前，必须把变量先同步回主内存中。
final的可见性是指被final修饰的字段在构造器中一旦被初始化完成，并且构造器没有把引用this传递出去，那么其他线程就能看见final字段的值。
(4)JMM如何处理有序性
一.volatile和synchronized关键字可保证多线程操作的有序性
volatile的有序性是由于volatile变量禁止了指令重排序优化。
synchronized的有序性则是通过如下这条规则来获得的：一个变量在同一时刻只允许一个线程对其进行lock操作。
二.通过Happens-Before规则来保证多线程操作的有序性
Happens-Before规则指定了哪些操作是不能进行重排序的。
7.volatile如何保证可见性
(1)volatile型变量的特殊规则
(2)volatile如何保证可见性
(1)volatile型变量的特殊规则
一.volatile变量对所有线程都是立即可见的
对volatile变量的所有写操作都能立刻反映到其他线程之中，volatile变量在各个线程的工作内存中是不存在数据不一致性的问题。
从物理存储的角度看，各个线程的工作内存中，volatile变量也可能存在不一致。但由于各个工作线程在每次使用volatile变量之前都要先刷新其值，于是执行引擎便看不到不一致的情况，因此可以认为不存在不一致的问题。
二.volatile变量是禁止指令重排序优化的
指令重排序是指CPU将多条指令，不按程序规定的顺序，分开发送给各个相应的电路单元进行处理。可见，volatile型变量的特殊规则就规定了volatile变量对所有线程立即可见。
(2)volatile如何保证可见性
普通变量和volatile变量的区别是：volatile保证了新值能立即同步到主内存，以及每次使用前立即从主内存刷新。也就是volatile保证了多线程操作时变量的可见性，而普通变量则不能保证。
如果flag变量是加了volatile关键字，那么当线程1通过assign操作将flag = 1写回工作内存时，会立即执行store和write操作将flag = 1同步到主内存。
同时还会让线程2的工作内存中的flag变量的缓存过期，这样当线程2后续从工作内存里读取flag变量的值时，发现缓存已经过期就会重新从主内存中加载flag = 1的值。
所以通过volatile关键字可以实现这样的效果：当一个线程修改了变量值，其他线程可以马上感知这个变量值。
8.volatile为什么无法保证原子性
(1)volatile型变量的特殊规则
(2)volatile不能保证原子性的字节码解释
(1)volatile型变量的特殊规则
一.volatile变量对所有线程都是立即可见的
对volatile变量的所有写操作都能立刻反映到其他线程之中，volatile变量在各个线程的工作内存中是不存在数据不一致性的问题的。
从物理存储的角度看，各个线程的工作内存中，volatile变量也可能存在不一致。但由于各个工作线程在每次使用volatile变量之前都要先刷新其值，于是执行引擎便看不到不一致的情况，因此可以认为不存在不一致的问题。
二.volatile变量是禁止指令重排序优化的
指令重排序是指CPU将多条指令，不按程序规定的顺序，分开发送给各个相应的电路单元进行处理。可见，volatile型变量的特殊规则并没有原子性方面的保证。
(2)volatile不能保证原子性的字节码解释
比如increase()方法只有一行代码，用javap反编译可知由4条字节码指令构成。当get_field指令把n的值取到操作栈顶时，volatile保证了n的值此时是最新的。但线程1执行iconst_1、iadd这些指令时，线程2可能已经把n的值改变了。于是此时线程1的操作栈顶的n值，就变成了过期数据，所以线程1执行put_field指令后就会把较小的n值同步回主内存中。
//n定义为初始值为0的静态变量
public volatile int n = 0;
public void increase() {
n++;
}
//javap反编译后的字节码
get_field
iconst_1
iadd
put_field
严格来说，volatile并不是轻量级的锁或者是轻量级同步机制。因为对于n++这样的基本操作，加了volatile关键字也无法保证原子性。而锁和同步机制，如synchonized或者lock是可以保证原子性的。
9.volatile如何保证有序性
(1)Happens-Before规则介绍
(2)Happens-Before规则详情
(3)volatile如何保证有序性
(1)Happens-Before规则介绍
Happens-Before规则指定了两个操作间的执行顺序。如果一个操作Happens-Before另一个操作，那么第一个操作的执行结果将对第二个操作可见，而且第一个操作的执行顺序排在第二个操作之前。
两个操作之间存在Happens-Before关系，并不意味着实际就是按Happens-Before规则指定的顺序来执行的。如果重排序后的执行结果与按Happens-Before规则执行的结果一致，那么JMM是允许这种重排序的。
As-If-Serial保证了单线程内程序的执行结果不被改变(不管怎么重排序)，Happens-Before保证了多线程下的程序执行结果不被改变(不管怎么重排序)。这两者都是为了不改变程序执行结果的前提，尽可能提高程序执行的并行度。
总结：
虽然编译器、指令器可能会对代码进行重排序，但要遵守一定的规则。Happens-Before规则就是限制了不能随便重排序，如果不符合Happens-Before规则，那么就可以按编译器、指令器要求重排序。
(2)Happens-Before规则详情
Happens-Before就是先行发生的意思。
一.程序次序规则
一个线程中的每个操作，先行发生于该线程中的任意后续操作。
二.锁规则
对一个锁的unlock操作先行发生于后面对该锁的lock操作。
三.volatile变量规则
对一个volatile变量的写操作先行发生于后面对这个volatile变量的读操作。
四.传递规则
如果操作A先行发生于操作B，而操作B又先行发生于操作C，那么就可以得出操作A先行发生于操作C。
五.start()规则
如果线程A执行线程B的start()方法启动线程B，那么线程A执行线程B的start()方法这一操作，先行发生于线程B的任意操作。
六.join()规则
如果线程A执行线程B的join()方法并成功返回，那么线程B的任意操作先行发生于，线程A执行线程B的join()方法的这一操作。
(3)volatile如何保证有序性
一.Happens-Before规则的volatile变量规则
程序中的代码如果满足上面这8条规则，就一定会保证指令的顺序。但是如果没满足上面的8条规则，那么就可能会出现指令重排。
如对一个volatile变量的写操作先行发生于后面对这个volatile变量的读操作。
二.volatile型变量的特殊规则
volatile型变量会禁止指令重排序优化。在有序性问题的例子一中，使用volatile修饰flag能禁止重排序避免逻辑异常。在有序性问题的例子二中，使用volatile修饰instance能禁止重排序避免异常。
10.volatile的原理(Lock前缀指令 + 内存屏障)
(1)Lock前缀指令 + MESI实现可见性
(2)通过内存屏障实现禁止指令重排序
(1)Lock前缀指令 + MESI实现可见性
如果对volatile关键字修饰的变量执行写操作，那么JVM就会向CPU发送一条Lock前缀指令，将这个变量所在的缓存行数据写回到主内存中。
同时根据MESI缓存一致性协议，各个CPU会通过嗅探在总线上传播的数据，来检查该变量的缓存值是否过期。如果发现过期，CPU就会将该变量所在的缓存行设置成无效状态。后续当这个CPU要读取该变量时，就会从主内存中加载最新的数据。
所以Lock前缀指令 + MESI缓存一致性协议实现了volatile型变量的可见性。Lock前缀指令会引起将volatile型变量所在的缓存行数据写回到主内存，MESI缓存一致性协议可让CPU检查出哪些缓存被修改，同时令缓存失效。
(2)通过内存屏障实现禁止指令重排序
一.通过内存屏障来禁止某些指令重排序
加了volatile关键字的变量，可以保证前后的一些代码不会被指令重排。那么这个是如何做到的呢？volatille是如何保证有序性的呢？
为了保证内存可见性，Java编译器会在生成指令序列的适当位置，插入内存屏障指令来禁止特定类型的指令重排序。
二.JMM的4种内存屏障指令
一.LoadLoad屏障
Load1;LoadLoad;Load2
确保Load1数据的装载，先于Load2及所有后续装载指令的装载。也就是Load1对应的代码和Load2对应的代码，是不能指令重排的。
二.StoreStore屏障
Store1;StoreStore;Store2
确保Store1数据刷新到主内存，先于Store2及所有后续存储指令的存储。
三.LoadStore屏障
Load1;LoadStore;Store2
确保Load1数据的装载，先于Store2及所有后续的存储指令刷新到主内存。
四.StoreLoad屏障
Store1;StoreLoad;Load2
确保Store1数据刷新到主内存，先于Load2及所有后续装载指令的装载。
三.JMM制定的volatile重排序规则表
举例来说，第三行最后一个单元格的意思是：当第一个操作为普通变量的读或写时，如果第二个操作为volatile些，则编译器不能重排序这两个操作。
当第一个操作是volatile读时，不管第二个操作是什么，都不能重排序。该规则确保volatile读之后的操作不会被编译器重排序到volatile读之前。
当第二个操作是volatile写时，不管第一个操作是什么，都不能重排序。该规则确保volatile写之前的操作不会被编译器重排序到volatile写之后。
当第一个操作是volatile写，第二个操作是volatile读时，不能重排序。
四.JMM内存屏障的插入策略
volatile的作用就是对于volatile变量的读写操作，都会加入内存屏障。
每个volatile写操作前，加StoreStore屏障，禁止前面的普通写和它重排；
每个volatile写操作后，加StoreLoad屏障，禁止后面的volatile读/写和它重排；
每个volatile读操作后，加LoadLoad屏障，禁止后面的普通读和它重排；
每个volatile读操作后，加LoadStore屏障，禁止后面的普通写和它重排；
public class VolatileStoreStoreExample {
int a = 0;
volatile boolean flag = false;
//线程1执行writer()方法
public void writer() {
a = 1;//普通写，操作1
//加入StoreStore屏障，避免操作1和操作2发生重排序，影响线程2的执行结果
flag = true;//volatile写，操作2
}
//线程2执行reader()方法
public void reader() {
if (flag) {//操作3
int i = a * a;//操作4
...
}
}
}
public class VolatileStoreLoadExample {
volatile int a = 0;
volatile boolean flag = false;
//线程1执行writer()方法
public void writer() {
a = 1;//volatile写，操作1
//加入StoreLoad屏障，避免操作1和操作2发生重排序，影响线程2的执行结果
flag = true;//volatile写，操作2
}
//线程2执行reader()方法
public void reader() {
if (flag) {//操作3
int i = a * a;//操作4
...
}
}
}
11.双重检查单例模式的volatile优化
(1)双重检查单例模式的实现缺陷
(2)双重检查单例模式的volatile优化
(1)双重检查单例模式的实现缺陷
在下面利用双重检查创建单例的代码中，Singleton.getInstance()方法会先判断inst是否为空，如果为空则锁定Singleton.class并再次检查inst是否为空，如果还为空就创建一个Singleton的对象实例。
其中"inst = new Singleton()"会执行三个指令：
指令1：分配对象的内存空间
指令2：初始化对象
指令3：设置inst变量指向指令1分配的内存地址
如果按照正常顺序来执行，那么是不会有问题的。但是由于指令2和指令3不存在依赖关系，所以编译器优化后可能进行重排序。于是执行顺序变为：指令1 -> 指令3 -> 指令2。那么就可能发生：在线程A刚把inst指向对应地址后，线程B获取到执行权。然后线程B便获取到一个没有初始化的对象，从而产生空指针异常。
public class Singleton {
private static Singleton inst;
private Singleton() {
}
public static Singleton getInstance() {
if (inst == null) {
synchronized(Singleton.class) {
if (inst == null) {
//开辟空间，inst指向地址，初始化
inst = new Singleton();
}
}
}
return inst;
}
}
(2)双重检查单例模式的volatile优化
public class Singleton {
private static volatile Singleton inst;
private Singleton() {
}
public static Singleton getInstance() {
if (inst == null) {
synchronized(Singleton.class) {
if (inst == null) {
//开辟空间，inst指向地址，初始化
inst = new Singleton();
}
}
}
return inst;
}
}
12.基于volatile优化微服务的优雅关闭机制
//在微服务上被创建和启动，负责和register-server进行通信
public class RegisterClient {
...
//服务实例是否在运行
private volatile Boolean isRunning;
public RegisterClient() {
this.serviceInstanceId = UUID.randomUUID().toString().replace("-", "");
this.httpSender = new HttpSender();
this.heartbeatWorker = new HeartbeatWorker();
this.isRunning = true;
}
//启动RegisterClient组件
public void start() {
try {
RegisterWorker registerWorker = new RegisterWorker();
registerWorker.start();
registerWorker.join();
heartbeatWorker.start();
} catch (Exception e) {
e.printStackTrace();
}
}
//停止RegisterClient组件
public void shutdown() {
this.isRunning = false;
this.heartbeatWorker.interrupt();
}
//服务注册线程
private class RegisterWorker extends Thread {
...
}
//心跳线程
private class HeartbeatWorker extends Thread {
@Override
public void run() {
//如果注册成功，就进入while true死循环
HeartbeatRequest heartbeatRequest = new HeartbeatRequest();
heartbeatRequest.setServiceName(SERVICE_NAME);
heartbeatRequest.setServiceInstanceId(serviceInstanceId);
HeartbeatResponse heartbeatResponse = null;
while(isRunning) {
try {
heartbeatResponse = httpSender.heartbeat(heartbeatRequest);
System.out.println("心跳的结果为：" + heartbeatResponse.getStatus() + "......");
Thread.sleep(HEARTBEAT_INTERVAL);
} catch (Exception e) {
e.printStackTrace();
}
}
}
}
}
13.基于volatile优化微服务存活状态检查机制
private class Lease {
//最近一次心跳的时间
private volatile Long latestHeartbeatTime = System.currentTimeMillis();
//续约，只要微服务发送一次心跳，就相当于维护了register-client和register-server之间的一个契约
//进行续约的意思就是表明还存活着，分布式系统中大都有契约机制
public void renew() {
this.latestHeartbeatTime = System.currentTimeMillis();
System.out.println("服务实例[" + serviceInstanceId + "]，进行续约：" + latestHeartbeatTime);
}
//判断当前服务实例的契约是否还存活
public Boolean isAlive() {
Long currentTime = System.currentTimeMillis();
if (currentTime - latestHeartbeatTime > NOT_ALIVE_PERIOD) {
System.out.println("服务实例[" + serviceInstanceId + "]，不再存活");
return false;
}
System.out.println("服务实例[" + serviceInstanceId + "]，保持存活");
return true;
}
}
14.i++的多线程安全问题演示
多个线程对一个共享数据并发写，可能会导致数据出错，这就是原子性问题。
public class ThreadUnsafeDemo {
private static int data = 0;
public static void main(String[] args) throws Exception {
Thread thread1 = new Thread() {
@Override
public void run() {
for (int i = 0; i < 10000; i++) {
ThreadUnsafeDemo.data++;
System.out.println("thread1:" + data);
}
}
};
Thread thread2 = new Thread() {
@Override
public void run() {
for (int i = 0; i < 10000; i++) {
ThreadUnsafeDemo.data++;
System.out.println("thread2:" + data);
}
}
};
thread1.start();
thread2.start();
thread1.join();
thread2.join();
//最后的结果未必是20000
System.out.println(data);
}
}
15.JMM是多线程并发安全问题的根本原因
多线程并发写一个共享变量会出现问题的根本原因是Java内存模型JMM。
在Java内存模型下，多个线程并发执行时，每个线程(一般对应一个CPU)都会有自己的工作内存，每个线程读写数据时，线程对应的CPU会从主内存获取数据到本地进行缓存。
volatile是无法保证原子性的，因为volatile的底层机制是：Lock前缀指令 + MESI缓存一致性协议。某线程修改变量时会刷主内存，并使其他线程工作内存的该变量缓存过期。
16.synchronized可解决多线程并发安全问题
public class ThreadUnsafeDemo {
private static int data = 0;
public static void main(String[] args) throws Exception {
Thread thread1 = new Thread() {
@Override
public void run() {
for (int i = 0; i < 10000; i++) {
increment();
}
}
};
Thread thread2 = new Thread() {
@Override
public void run() {
for (int i = 0; i < 10000; i++) {
increment();
}
}
};
thread1.start();
thread2.start();
thread1.join();
thread2.join();
//最后的结果未必是20000
System.out.println(data);
}
private synchronized static void increment() {
ThreadUnsafeDemo.data++;
System.out.println("thread1:" + data);
}
}
17.synchronized的常见使用方法总结
(1)加类锁
(2)加对象锁
(3)类锁和对象锁的区别
(1)加类锁
如果synchronized一个静态方法，就是对这个类的Class对象加锁。如果synchronized(类.class)，也是对这个类的Class对象加锁。同一时间只有一个线程可以访问同一个类的synchronized方法。注意：每个类都会对应一个Class对象。
public class ThreadUnsafeDemo {
private static int data = 0;
public static void main(String[] args) throws Exception {
Thread thread1 = new Thread() {
public void run() {
for (int i = 0; i < 10000; i++) {
increment();
}
}
};
Thread thread2 = new Thread() {
public void run() {
for (int i = 0; i < 10000; i++) {
increment();
}
}
};
thread1.start();
thread2.start();
thread1.join();
thread2.join();
//最后的结果未必是20000
System.out.println(data);
}
private synchronized static void increment() {
ThreadUnsafeDemo.data++;
System.out.println("thread1:" + data);
}
}
//synchronized一个类的静态方法，等价于synchronized(该类.class)
synchronized(MyObject.class) {
}
(2)加对象锁
如果synchronized一个普通的方法，那么就是对当前的对象实例加锁。同一时间只有一个线程可以访问同一个对象实例的synchronized方法。注意：synchronized一个代码片段的常见写法，就是synchronized(this)，意思就是基于当前这个对象实例来进行加锁。
public class ThreadUnsafeDemo {
private static int data = 0;
public static void main(String[] args) throws Exception {
final ThreadUnsafeDemo demo = new ThreadUnsafeDemo();
Thread thread1 = new Thread() {
public void run() {
for (int i = 0; i < 10000; i++) {
demo.increment();
}
}
};
Thread thread2 = new Thread() {
public void run() {
for (int i = 0; i < 10000; i++) {
demo.increment();
}
}
};
thread1.start();
thread2.start();
thread1.join();
thread2.join();
//最后的结果未必是20000
System.out.println(data);
}
private synchronized void increment() {
ThreadUnsafeDemo.data++;
System.out.println("thread1:" + data);
}
}
或者
public class ThreadUnsafeDemo {
private static int data = 0;
public static void main(String[] args) throws Exception {
final ThreadUnsafeDemo demo = new ThreadUnsafeDemo();
Thread thread1 = new Thread() {
public void run() {
for (int i = 0; i < 10000; i++) {
synchronized(demo) {
ThreadUnsafeDemo.data++;
System.out.println("thread1:" + data);
}
}
}
};
Thread thread2 = new Thread() {
public void run() {
for (int i = 0; i < 10000; i++) {
synchronized(demo) {
ThreadUnsafeDemo.data++;
System.out.println("thread1:" + data);
}
}
}
};
thread1.start();
thread2.start();
thread1.join();
thread2.join();
//最后的结果未必是20000
System.out.println(data);
}
}
//对对象实例的方法加锁
private synchronized void increment() {
ThreadUnsafeDemo.data++;
System.out.println("thread1:" + data);
}
//等价于synchronized(this)
private void increment() {
synchronized(this) {
ThreadUnsafeDemo.data++;
System.out.println("thread1:" + data);
}
}
//等价于synchronized(对象实例)
synchronized(myObject) {
}
(3)类锁和对象锁的区别
synchronized锁分两种：一是对某个对象加锁，二是对某个类加锁。对类加锁其实也是在对一个对象加锁，只不过是对类的Class对象加锁。
类是在JVM启动过程中加载的，每个.class文件被装载后会产生一个Class对象，每个.class文件产生的Class对象在JVM进程中是全局唯一的。static修饰的成员变量和方法，它们的生命周期都是属于类级别的，它们随着类的定义被分配和装载到内存，随着类被卸载而回收。
实例对象的生命周期伴随着实例对象的创建而开始，同时伴随着实例对象的回收而结束。
因此，类锁和对象锁的最大区别就是：锁的生命周期不同。
18.synchronized的底层原理
(1)锁信息的存储
(2)锁的四种状态
(3)什么是偏向锁
(4)什么是轻量级锁
(5)什么是重量级锁
(6)锁升级的流程
(7)锁膨胀的流程
(8)synchronized的lock和unlock操作规定保证原子性 + 可见性 + 有序性
(9)内核态和用户态说明
(1)锁信息的存储
一个Java对象的存储结构由三部分组成：对象头、实例数据、对齐填充。其中对象头也由三部分组成：Mark Word、Klass Pointer、Length，而Mark Word会记录该对象的HashCode、分代年龄和锁标记位；
(2)锁的四种状态
为了减少获得锁和释放锁带来的性能损耗，JDK 1.6引入了偏向锁和轻量级锁。锁一共有4种状态：无锁状态、偏向锁状态、轻量级锁状态、重量级锁状态，这几个状态会随着竞争情况逐渐升级。锁可以升级但不能降级，这意味着偏向锁升级为轻量级锁后不能降级回偏向锁。
(3)什么是偏向锁
一.偏向锁的介绍
大多数情况下，锁不仅不存在多线程竞争，而且总会由同一线程多次获得。所以，为了让线程获得锁的代价更低，便引入了偏向锁。
偏向锁可以认为是在没有多线程竞争的情况下，访问同步块的加锁场景。也就是在单线程执行同步块的情况下，就没有必要使用重量级锁了。为了提升性能，没必要基于操作系统级别的Mutex Lock来实现锁的抢占。
偏向锁的作用是：线程在没有线程竞争的情况下去访问同步块代码时，会先尝试通过偏向锁来抢占访问资格，这个抢占过程是基于CAS来完成的。如果抢占锁成功，则直接修改对象头中的Mark Word信息。也就是修改偏向锁标记为1、锁标记为01，以及存储当前获得锁的线程ID。
偏向的意思是：如果线程X获得了偏向锁，当线程X后续再访问这个同步块时，就会判断出对象头中的线程ID和线程X相等，于是就不需要再次抢占锁了。
二.偏向锁的实现流程
三.偏向锁的实现原理
偏向锁的实现原理就是使用CAS来设置对象头中的线程ID。如果成功则获得偏向锁，如果失败则升级到轻量级锁。
(4)什么是轻量级锁
一.轻量级锁的介绍
如果没有线程竞争，使用偏向锁能够在不影响性能的前提下获得锁。如果有多个线程并发访问同步块，那么没抢占到锁的线程只能进行阻塞等待。但在使用重量级锁阻塞等待前，还有更好的平衡方案，也就是使用轻量级锁。
所谓的轻量级锁，就是没有抢占到锁的线程，进行一定次数的自旋重试。如果线程在重试过程中抢占到了锁，那么这个线程就不需要阻塞了。
如果持有锁的线程占用锁的时间比较短，则自旋带来的性能提高会比较明显。如果持有锁的线程占用锁的时间比较长，则自旋就会浪费CPU资源。所以线程通过自旋来重试抢占锁的次数必须要有一个限制。
为了优化自旋，JDK还引入了自适应自旋锁。如果在一个锁对象中，通过自旋获得锁很少成功，则JVM会缩短自旋次数。否则，JVM可能会增加自旋次数。
二.轻量级锁的实现流程
三.轻量级锁的实现原理
如果偏向锁存在竞争或者偏向锁未开启，那么当线程访问同步块代码时就会通过轻量级锁来抢占锁资源。轻量级锁的原理就是，通过CAS来修改锁对象中指向Lock Record的指针。
其中偏向锁存在竞争指的是，还没有线程通过CAS设置对象头的线程ID。此时多个线程会同时执行CAS尝试获取偏向锁，失败的就升级轻量级锁。
而偏向锁未开启指的是，已有线程成功通过CAS设置了对象头的线程ID。此时多个线程同时访问同步块代码，就会直接升级轻量级锁。
(5)什么是重量级锁
轻量级锁能够通过一定次数的重试，让每一个没获得锁的线程有可能抢占到锁。但轻量级锁只有在获得锁的线程持有锁的时间比较短的情况下才能提升性能。如果持有锁的线程占用锁的时间较长，那么不能让没抢到锁的线程一直自旋。
如果没抢到锁的线程通过一定次数的自旋后，发现仍然没有获得锁。那么就只能升级到重量级锁，来进行阻塞等待了。
重量级锁的本质是：没有获得锁的线程会通过park()方法挂起，接着被获得锁的线程通过unpark()方法唤醒后再次抢占锁，直到抢占成功。
重量级锁依赖于底层操作系统的Mutex Lock来实现。使用Mutex Lock时需要从用户态切换到内核态，才能将当前线程挂起，所以性能开销比较大。
从偏向锁到轻量级锁再到重量级锁，整个优化过程其实使用了乐观锁的思想。
(6)锁升级的流程
一.锁升级的流程
当一个线程访问使用了synchronized修饰的代码块时，如果当前还没有线程获得偏向锁，则先通过CAS尝试获得偏向锁。如果当前已有线程获得偏向锁，则尝试升级到轻量级锁去抢占锁。轻量级锁就是通过多次CAS(也就是自旋)来完成的。如果线程通过多次自旋仍然无法获得锁，那么就只能升级到重量级锁进行阻塞等待。
二.偏向锁和轻量级锁的区别
偏向锁只能保证偏向同一个线程，只要有线程获得过偏向锁，那么当其他线程抢占锁时，只能通过轻量级锁来实现，除非触发了重新偏向。如果获得轻量级锁的线程在后续的20次访问中，发现每次访问锁的线程都是同一个，那么就会触发重新偏向。
轻量级锁可以灵活释放。如果线程1抢占了轻量级锁，那么在锁用完并释放后，线程2可以继续通过轻量级锁来抢占锁资源。
偏向锁，就是在一段时间内只会由同一个线程来获得和释放锁，加锁的方式是把线程ID保存到锁对象的Mark Word中。
轻量级锁，就是在一段时间内可能会由多个线程来获得和释放锁。存在锁交替竞争的场景，但在同一时刻不会有多个线程同时获得锁。加锁的方式是首先在每个线程的栈帧中分配一个Lock Record，然后把锁对象中的Mark Word拷贝到Lock Record中，最后把锁对象的Mark Word的指针指向Lock Record。
(7)锁膨胀的流程
一.获取重量级锁之前的锁膨胀
如果线程在运行synchronized修饰的同步块代码时，发现锁状态是轻量级锁并且有其他线程抢占了锁资源，那么该线程就会触发锁膨胀升级到重量级锁。
在获取重量级锁之前会先实现锁膨胀，锁膨胀时首先会创建一个ObjectMonitor对象，然后把ObjectMonitor对象的指针保存到锁对象的Mark Word中。
重量级锁的实现是在ObjectMonitor中完成的，所以锁膨胀的意义就是构建一个ObjectMonitor对象。
二.ObjectMonitor对象的重要字段
_owner：保存当前持有锁的线程
_cxq：没有获得锁的线程队列
_waitset：被wait()方法阻塞的线程队列
_recursions：锁被重入的次数
三.重量级锁的获取流程
重量级锁的竞争都是在ObjectMonitor对象中完成的。首先判断当前线程是否是重入，如果是则重入次数 + 1。
然后通过CAS自旋来判断ObjectMonitor中的_owner字段是否为空。如果为空，则表示重量级锁已经被释放，当前线程可以获得锁。如果不为空，就继续进行自适应自旋重试。
最后如果通过自旋竞争锁失败，那么就把当前线程构建成一个ObjectWaiter结点，插入到ObjectMonitor的_cxq队列的队头，然后再调用park()方法阻塞当前线程。
四.重量级锁的释放流程
首先把ObjectMonitor的_owner字段设置为null，然后从ObjectMonitor的_cxq队列中调用unpark()方法唤醒一个阻塞的线程。被唤醒的线程会重新竞争重量级锁，如果没抢到，则继续阻塞等待。因为synchronized是非公平锁，被唤醒的线程不一定能重新抢占到锁。
(8)synchronized的lock和unlock操作规定保证原子性 + 可见性 + 有序性
synchronized是由monitorenter和monitorexit这两条字节码指令来实现的。可以理解为monitorenter指令对应了重量级锁的获取流程，monitorexit指令对应了重量级锁的释放流程。
这两个字节码指令最终又会在内存间的交互时，使用lock操作和unlock操作。通过lock操作和unlock操作的语义来实现synchronized的原子性。
lcok操作前需要从主内存同步最新值到工作内存，unlock操作前会将工作内存上的值刷新回主内存，这样lock操作和unlock操作就实现了synchronized的可见性。
此外还规定，同一时刻只有一条线程可以进行lock操作，这样就实现了synchronized的有序性。
(9)内核态和用户态说明
在重量级锁中，线程的阻塞和唤醒是通过park()方法和unpark()方法来完成的，park()方法和unpark()方法需要通过系统调用来完成。
由于系统调用是在内核态中运行的，所以进行系统调用时，系统需要从用户态切换到内核态。系统从用户态切换到内核态的这个过程会产生性能损耗。在切换之前需要保存用户态的状态，包括寄存器、程序指令等。然后才能执行内核态的系统调用指令，最后还要恢复用户态。
用户态和内核态表示的是操作系统中的不同执行权限。两者最大的区别在于：运行在用户空间中的进程不能直接访问操作系统内核的指令和程序，运行在内核空间中的进程可以直接访问操作系统内核的指令和程序。进行权限划分是为了避免用户在进程中直接操作危险的系统指令，从而影响进程和系统的稳定。
19.微服务定时拉取注册表信息
(1)HTTP请求组件增加拉取服务注册表信息
(2)每隔30秒定时去服务端拉取注册表信息
(3)添加客户端缓存的注册表到register-client
(1)HTTP请求组件增加拉取服务注册表信息
//负责发送各种HTTP请求的组件
public class HttpSender {
//发送注册请求
public RegisterResponse register(RegisterRequest request) {
//实际可能会基于类似HttpClient这种开源的网络包来进行发送
System.out.println("服务实例[" + request + "]，发送请求进行注册......");
//收到register-server响应后，封装一个Response对象
RegisterResponse response = new RegisterResponse();
response.setStatus(RegisterResponse.SUCCESS);
return response;
}
//发送心跳请求
public HeartbeatResponse heartbeat(HeartbeatRequest request) {
System.out.println("服务实例[" + request + "]，发送请求进行心跳......");
HeartbeatResponse response = new HeartbeatResponse();
response.setStatus(RegisterResponse.SUCCESS);
return response;
}
//模拟拉取服务注册表信息
public Map<String, Map<String, ServiceInstance>> fetchServiceRegistry() {
Map<String, Map<String, ServiceInstance>> registry = new HashMap<String, Map<String, ServiceInstance>>();
ServiceInstance serviceInstance = new ServiceInstance();
serviceInstance.setHostname("finance-service-01");
serviceInstance.setIp("192.168.31.1207");
serviceInstance.setPort(9000);
serviceInstance.setServiceInstanceId("FINANCE-SERVICE-192.168.31.207:9000");
serviceInstance.setServiceName("FINANCE-SERVICE");
Map<String, ServiceInstance> serviceInstances = new HashMap<String, ServiceInstance>();
serviceInstances.put("FINANCE-SERVICE-192.168.31.207:9000", serviceInstance);
registry.put("FINANCE-SERVICE", serviceInstances);
System.out.println("拉取注册表：" + registry);
return registry;
}
}
(2)每隔30秒定时去服务端拉取注册表信息
//服务注册中心的客户端，缓存的一个服务注册表
public class ClientCachedServiceRegistry {
//拉取服务注册表的间隔时间：30s
private static final Long SERVICE_REGISTRY_FETCH_INTERVAL = 30 * 1000L;
//客户端缓存的服务注册表
private Map<String, Map<String, ServiceInstance>> registry = new HashMap<String, Map<String, ServiceInstance>>();
//负责定时拉取注册表到客户端进行缓存的后台线程
private Daemon daemon;
//RegisterClient
private RegisterClient registerClient;
//HTTP通信组件
private HttpSender httpSender;
//构造方法
public ClientCachedServiceRegistry(RegisterClient registerClient, HttpSender httpSender) {
this.daemon = new Daemon();
this.registerClient = registerClient;
this.httpSender = httpSender;
}
//初始化
public void initialize() {
this.daemon.start();
}
//销毁这个组件
public void destroy() {
this.daemon.interrupt();
}
//获取服务注册表
public Map<String, Map<String, ServiceInstance>> getRegistry() {
return registry;
}
//负责定时拉取注册表信息到本地缓存
private class Daemon extends Thread {
@Override
public void run() {
while(registerClient.isRunning()) {
try {
registry = httpSender.fetchServiceRegistry();
Thread.sleep(SERVICE_REGISTRY_FETCH_INTERVAL);
} catch (Exception e) {
e.printStackTrace();
}
}
}
}
}
(3)添加客户端缓存的注册表到register-client
//在服务上被创建和启动，负责跟register-server进行通信
public class RegisterClient {
public static final String SERVICE_NAME = "inventory-service";
public static final String IP = "192.168.31.207";
public static final String HOSTNAME = "inventory01";
public static final int PORT = 9000;
private static final Long HEARTBEAT_INTERVAL = 30 * 1000L;
//服务实例id
private String serviceInstanceId;
//HTTP通信组件
private HttpSender httpSender;
//心跳线程
private HeartbeatWorker heartbeatWorker;
//服务实例是否在运行
private volatile Boolean isRunning;
//客户端缓存的注册表
private ClientCachedServiceRegistry registry;
//构造方法
public RegisterClient() {
this.serviceInstanceId = UUID.randomUUID().toString().replace("-", "");
this.httpSender = new HttpSender();
this.heartbeatWorker = new HeartbeatWorker();
this.isRunning = true;
this.registry = new ClientCachedServiceRegistry(this, httpSender);
}
//启动RegisterClient组件
public void start() {
try {
//这个线程刚启动时，首先需要完成注册
//完成注册后，就会进入一个while true循环，每隔30秒发送一个心跳请求
RegisterWorker registerWorker = new RegisterWorker();
registerWorker.start();
registerWorker.join();
//启动心跳线程，定时发送心跳请求
heartbeatWorker.start();
//初始化客户端缓存的服务注册表组件
this.registry.initialize();
} catch (Exception e) {
e.printStackTrace();
}
}
//停止RegisterClient组件
public void shutdown() {
this.isRunning = false;
this.heartbeatWorker.interrupt();
this.registry.destroy();
}
//返回RegisterClient是否正在运行
public Boolean isRunning() {
return isRunning;
}
//服务注册线程
private class RegisterWorker extends Thread {
@Override
public void run() {
RegisterRequest registerRequest = new RegisterRequest();
registerRequest.setServiceName(SERVICE_NAME);
registerRequest.setIp(IP);
registerRequest.setHostname(HOSTNAME);
registerRequest.setPort(PORT);
registerRequest.setServiceInstanceId(serviceInstanceId);
RegisterResponse registerResponse = httpSender.register(registerRequest);
System.out.println("服务注册的结果是：" + registerResponse.getStatus() + "......");
}
}
//心跳线程
private class HeartbeatWorker extends Thread {
@Override
public void run() {
//如果注册成功，就进入while true循环
HeartbeatRequest heartbeatRequest = new HeartbeatRequest();
heartbeatRequest.setServiceName(SERVICE_NAME);
heartbeatRequest.setServiceInstanceId(serviceInstanceId);
HeartbeatResponse heartbeatResponse = null;
while(isRunning) {
try {
heartbeatResponse = httpSender.heartbeat(heartbeatRequest);
System.out.println("心跳的结果为：" + heartbeatResponse.getStatus() + "......");
Thread.sleep(HEARTBEAT_INTERVAL);
} catch (Exception e) {
e.printStackTrace();
}
}
}
}
}
20.基于synchronized解决注册表并发问题
//服务注册表
public class ServiceRegistry {
//注册表是一个单例
private static ServiceRegistry instance = new ServiceRegistry();
private ServiceRegistry() {
}
//核心的内存数据结构：注册表
//Map<String, ServiceInstance>：key是服务名称，比如库存服务，value是这个服务的所有的服务实例，比如提供库存服务的2台机器
private Map<String, Map<String, ServiceInstance>> registry = new HashMap<String, Map<String, ServiceInstance>>();
//服务注册
public synchronized void register(ServiceInstance serviceInstance) {
Map<String, ServiceInstance> serviceInstanceMap = registry.get(serviceInstance.getServiceName());
if (serviceInstanceMap == null) {
serviceInstanceMap = new HashMap<String, ServiceInstance>();
registry.put(serviceInstance.getServiceName(), serviceInstanceMap);
}
serviceInstanceMap.put(serviceInstance.getServiceInstanceId(), serviceInstance);
System.out.println("服务实例[" + serviceInstance + "]，完成注册......");
System.out.println("注册表：" + registry);
}
//获取服务实例
public synchronized ServiceInstance getServiceInstance(String serviceName, String serviceInstanceId) {
Map<String, ServiceInstance> serviceInstanceMap = registry.get(serviceName);
return serviceInstanceMap.get(serviceInstanceId);
}
//获取整个注册表
public synchronized Map<String, Map<String, ServiceInstance>> getRegistry() {
return registry;
}
//从注册表中删除一个服务实例
public synchronized void remove(String serviceName, String serviceInstanceId) {
System.out.println("服务实例[" + serviceInstanceId + "]，从注册表中进行摘除");
Map<String, ServiceInstance> serviceInstanceMap = registry.get(serviceName);
serviceInstanceMap.remove(serviceInstanceId);
}
//获取服务注册表实例
public static ServiceRegistry getInstance() {
return instance;
}
}
由于ConcurrentHashMap是线程安全的数据结构，所以可以使用ConcurrentHashMap来代替synchronized关键字。通过ConcurrentHashMap的分段加锁机制，能更好地支持并发。
21.微服务注册中心的自我保护机制
(1)为什么要引入自我保护机制
(2)register-server什么情况会判断是自己问题导致无法接收心跳请求
(3)为了实现自我保护机制需要收集心跳总次数
(1)为什么要引入自我保护机制
对于单点的微服务注册中心，一般需要有一个自我保护的机制。如果register-server出现网络故障，导致大量服务实例没办法发送心跳，这时候就不能直接把所有的服务实例都给摘除掉。因为这时候是服务端自己出现问题，客户端服务是正常的，所以就需要引入自我保护的机制了。
(2)register-server什么情况会判断是自己问题导致无法接收心跳请求
可以设定一个比例，比如25%。如果ServiceAliveMonitor发现超过25%的服务实例的心跳没及时更新，那么就可以认为是register-server网络故障而导致接收不到心跳请求。于是就开始自动进入自我保护机制，不再摘除任何的服务实例。从而避免register-server在自己网络故障的情况下，一下子就摘除大量服务实例，导致注册表的数据出现严重丢失。
如果后续发现已有超过75%的服务实例已恢复发送心跳请求，此时ServiceAliveMonitor可以退出自我保护的状态，并继续检查某个服务实例的心跳是否超过90秒还没更新，如果是则认为这个服务实例已宕机，从注册表中进行摘除。
(3)为了实现自我保护机制需要收集心跳总次数
比如有10个服务实例，每分钟应该有20次心跳，但某一分钟只收到8次心跳。此时发现8 < 20 * 0.75，于是有超过25%的服务实例的心跳没有正常发送。此时就可以认为是register-server自己网络故障，从而触发自我保护机制。此时不再摘除任何服务实例，避免注册表的数据出现问题。
如果某一分钟收到的心跳次数达到了18次，18 > 20 * 0.85，此时就可以认为网络恢复了正常，于是就退出自我保护机制，继续检查服务实例的心跳是否在90秒内更新过。如果没更新过，就摘除这个故障的服务实例。
总结：为了实现自我保护机制，register-server需要记录每分钟接收多少心跳请求。所以ServiceAliveMonitor线程每次尝试摘除服务实例时，都会检查上一分钟的心跳次数是否满足超75%的服务实例都正常。如果不满足，就进入自我保护机制，避免摘除大量的服务实例。
22.基于synchronized实现服务心跳计数器
(1)服务端接收到心跳请求时增加记录心跳次数
(2)增加一个心跳请求计数器
(1)服务端接收到心跳请求时增加记录心跳次数
public class RegisterServerController {
private ServiceRegistry registry = ServiceRegistry.getInstance();
//处理发送过来的服务注册请求
public RegisterResponse register(RegisterRequest registerRequest) {
RegisterResponse registerResponse = new RegisterResponse();
try {
ServiceInstance serviceInstance = new ServiceInstance();
serviceInstance.setHostname(registerRequest.getHostname());
serviceInstance.setIp(registerRequest.getIp());
serviceInstance.setPort(registerRequest.getPort());
serviceInstance.setServiceInstanceId(registerRequest.getServiceInstanceId());
serviceInstance.setServiceName(registerRequest.getServiceName());
registry.register(serviceInstance);
registerResponse.setStatus(RegisterResponse.SUCCESS);
} catch (Exception e) {
e.printStackTrace();
registerResponse.setStatus(RegisterResponse.FAILURE);
}
return registerResponse;
}
//处理发送过来的心跳请求
public HeartbeatResponse heartbeat(HeartbeatRequest heartbeatRequest) {
HeartbeatResponse heartbeatResponse = new HeartbeatResponse();
try {
//对服务实例进行续约
ServiceInstance serviceInstance = registry.getServiceInstance(
heartbeatRequest.getServiceName(),
heartbeatRequest.getServiceInstanceId());
serviceInstance.renew();
//记录每分钟的心跳次数
HeartbeatMessuredRate heartbeatMeasuredRate = new HeartbeatMeasuredRate();
heartbeatMeasuredRate.increment();
heartbeatResponse.setStatus(HeartbeatResponse.SUCCESS);
} catch (Exception e) {
e.printStackTrace();
heartbeatResponse.setStatus(HeartbeatResponse.FAILURE);
}
return heartbeatResponse;
}
//拉取服务注册表信息
public Map<String, Map<String, ServiceInstance>> fetchServiceRegistry() {
return registry.getRegistry();
}
}
(2)增加一个心跳请求计数器
//心跳请求计数器
public class HeartbeatMeasuredRate {
//单例实例
private static HeartbeatMeasuredRate instance = new HeartbeatMeasuredRate();
//最近一分钟的心跳次数
private long latestMinuteHeartbeatRate = 0L;
//最近一分钟的时间戳
private long latestMinuteTimestamp = System.currentTimeMillis();
//获取单例实例
public static HeartbeatMeasuredRate getInstance() {
return instance;
}
//增加最近一分钟的心跳次数
public synchronized void increment() {
long currentTime = System.currentTimeMillis();
if (currentTime - latestMinuteTimestamp > 60 * 1000) {
latestMinuteHeartbeatRate = 0L;
this.latestMinuteTimestamp = System.currentTimeMillis();
}
latestMinuteHeartbeatRate++;
}
//获取最近一分钟的心跳次数
public synchronized long get() {
return latestMinuteHeartbeatRate;
}
}
可以使用Atomic来代替synchronized关键字来优化这个心跳计数器。
23.微服务关闭时的服务下线实现
(1)register-client的HttpSender组件增加服务下线接口
(2)register-server的Controller增加对服务下线请求的处理
(1)register-client的HttpSender组件增加服务下线接口
服务关闭时，需要发送一个请求给register-server，通知服务下线了。
//负责发送各种HTTP请求的组件
public class HttpSender {
//发送注册请求
public RegisterResponse register(RegisterRequest request) {
...
}
//发送心跳请求
public HeartbeatResponse heartbeat(HeartbeatRequest request) {
...
}
//拉取服务注册表
public Map<String, Map<String, ServiceInstance>> fetchServiceRegistry() {
...
}
//服务下线
public void cancel(String serviceName, String serviceInstanceId) {
System.out.println("服务实例下线[" + serviceName + ", " + serviceInstanceId + "]");
}
}
(2)register-server的Controller增加对服务下线请求的处理
public class RegisterServerController {
private ServiceRegistry registry = ServiceRegistry.getInstance();
//服务注册
public RegisterResponse register(RegisterRequest registerRequest) {
...
}
//发送心跳
public HeartbeatResponse heartbeat(HeartbeatRequest heartbeatRequest) {
...
}
//拉取服务注册表信息
public Map<String, Map<String, ServiceInstance>> fetchServiceRegistry() {
return registry.getRegistry();
}
//服务下线
public void cancel(String serviceName, String serviceInstanceId) {
registry.remove(serviceName, serviceInstanceId);
}
}
24.基于synchronized修改触发自我保护阈值
(1)新增一个自我保护机制的类
(2)服务注册和下线时更新触发自我保护阈值
(3)监控到服务不存活时更新触发自我保护阈值
目前已经可以记录register-server每分钟收到的心跳请求次数，但还需要知道每分钟收到多少次心跳才不会触发自我保护机制。假设每个服务实例每分钟会发送2个心跳请求给register-server。那么当注册一个服务实例时，要修改触发自我保护机制的阈值，比如加2。当摘除一个服务实例或某个服务实例下线时，也要修改该阈值，比如减2。
(1)新增一个自我保护机制的类
//自我保护机制的类
public class SelfProtectionPolicy {
private static SelfProtectionPolicy instance = new SelfProtectionPolicy();
//期望的心跳次数，如果有10个服务实例，这个数值就是10 * 2 = 20
private long expectedHeartbeatRate = 0L;
//期望的心跳次数的阈值，10 * 2 * 0.75 = 15，每分钟至少有15次心跳才不用开启自我保护机制
private long expectedHeartbeatThreshold = 0L;
//返回实例
public static SelfProtectionPolicy getInstance() {
return instance;
}
public long getExpectedHeartbeatRate() {
return expectedHeartbeatRate;
}
public void setExpectedHeartbeatRate(long expectedHeartbeatRate) {
this.expectedHeartbeatRate = expectedHeartbeatRate;
}
public long getExpectedHeartbeatThreshold() {
return expectedHeartbeatThreshold;
}
public void setExpectedHeartbeatThreshold(long expectedHeartbeatThreshold) {
this.expectedHeartbeatThreshold = expectedHeartbeatThreshold;
}
}
(2)服务注册和下线时更新触发自我保护阈值
public class RegisterServerController {
private ServiceRegistry registry = ServiceRegistry.getInstance();
//服务注册
public RegisterResponse register(RegisterRequest registerRequest) {
RegisterResponse registerResponse = new RegisterResponse();
try {
//在注册表中加入这个服务实例
...
registry.register(serviceInstance);
//更新自我保护机制的阈值
synchronized(SelfProtectionPolicy.class) {
SelfProtectionPolicy selfProtectionPolicy = SelfProtectionPolicy.getInstance();
selfProtectionPolicy.setExpectedHeartbeatRate(selfProtectionPolicy.getExpectedHeartbeatRate() + 2);
selfProtectionPolicy.setExpectedHeartbeatThreshold((long)(selfProtectionPolicy.getExpectedHeartbeatRate() * 0.85));
}
registerResponse.setStatus(RegisterResponse.SUCCESS);
} catch (Exception e) {
e.printStackTrace();
registerResponse.setStatus(RegisterResponse.FAILURE);
}
return registerResponse;
}
...
//服务下线
public void cancel(String serviceName, String serviceInstanceId) {
//从服务注册表中摘除服务实例
registry.remove(serviceName, serviceInstanceId);
//更新触发自我保护机制的阈值
synchronized(SelfProtectionPolicy.class) {
SelfProtectionPolicy selfProtectionPolicy = SelfProtectionPolicy.getInstance();
selfProtectionPolicy.setExpectedHeartbeatRate(selfProtectionPolicy.getExpectedHeartbeatRate() - 2);
selfProtectionPolicy.setExpectedHeartbeatThreshold((long)(selfProtectionPolicy.getExpectedHeartbeatRate() * 0.85));
}
}
}
(3)监控到服务不存活时更新触发自我保护阈值
//微服务存活状态监控组件
public class ServiceAliveMonitor {
//检查服务实例是否存活的间隔
private static final Long CHECK_ALIVE_INTERVAL = 60 * 1000L;
//负责监控微服务存活状态的后台线程
private Daemon daemon;
public ServiceAliveMonitor() {
this.daemon = new Daemon();
daemon.setDaemon(true);
daemon.setName("ServiceAliveMonitor");
}
//启动后台线程
public void start() {
daemon.start();
}
//负责监控微服务存活状态的后台线程
private class Daemon extends Thread {
private ServiceRegistry registry = ServiceRegistry.getInstance();
@Override
public void run() {
Map<String, Map<String, ServiceInstance>> registryMap = null;
while(true) {
try {
registryMap = registry.getRegistry();
for (String serviceName : registryMap.keySet()) {
Map<String, ServiceInstance> serviceInstanceMap = registryMap.get(serviceName);
for (ServiceInstance serviceInstance : serviceInstanceMap.values()) {
//如果服务实例距离上一次发送心跳已经超过90秒，则认为这个服务不存活
//此时需要从注册表中摘除这个服务实例
if (!serviceInstance.isAlive()) {
registry.remove(serviceName, serviceInstance.getServiceInstanceId());
//更新自我保护机制的阈值
synchronized(SelfProtectionPolicy.class) {
SelfProtectionPolicy selfProtectionPolicy = SelfProtectionPolicy.getInstance();
selfProtectionPolicy.setExpectedHeartbeatRate(selfProtectionPolicy.getExpectedHeartbeatRate() - 2);
selfProtectionPolicy.setExpectedHeartbeatThreshold((long)(selfProtectionPolicy.getExpectedHeartbeatRate() * 0.85));
}
}
}
}
Thread.sleep(CHECK_ALIVE_INTERVAL);
} catch (Exception e) {
e.printStackTrace();
}
}
}
}
}
25.基于synchronized开启自我保护机制
(1)增加是否需要开启自我保护机制的方法
(2)心跳请求计数器定时刷新近一分钟心跳次数
(3)监控服务是否存活的后台线程增加判断是否开启自我保护机制
随着服务的注册、下线、故障，触发自我保护机制的阈值会不断变动。在ServiceAliveMonitor摘除故障服务前，可先判断是否要触发自我保护机制。也就是判断是否满足：上一分钟的心跳次数 < 期望的心跳次数 * 0.75。如果小于则认为register-server出现网络故障，无法接收客户端的心跳请求。此时register-server就启动自我保护机制，不再摘除任何服务实例。
(1)增加是否需要开启自我保护机制的方法
//自我保护机制
public class SelfProtectionPolicy {
private static SelfProtectionPolicy instance = new SelfProtectionPolicy();
//期望的心跳次数，如果有10个服务实例，这个数值就是10 * 2 = 20
private long expectedHeartbeatRate = 0L;
//期望的心跳次数的阈值，10 * 2 * 0.75 = 15，每分钟至少得有15次心跳才不用开启自我保护机制
private long expectedHeartbeatThreshold = 0L;
//返回实例
public static SelfProtectionPolicy getInstance() {
return instance;
}
//是否需要开启自我保护机制
public Boolean isEnable() {
HeartbeatMessuredRate heartbeatMessuredRate = HeartbeatMessuredRate.getInstance();
long latestMinuteHeartbeatRate = heartbeatMessuredRate.get();
if (latestMinuteHeartbeatRate < this.expectedHeartbeatThreshold) {
System.out.println("[开启自我保护机制]最近一分钟心跳次数=" + latestMinuteHeartbeatRate + ", 期望心跳次数=" + this.expectedHeartbeatThreshold);
return true;
}
System.out.println("[未开启自我保护机制]最近一分钟心跳次数=" + latestMinuteHeartbeatRate + ", 期望心跳次数=" + this.expectedHeartbeatThreshold);
return false;
}
...
}
(2)心跳请求计数器定时刷新近一分钟心跳次数
//心跳请求计数器
public class HeartbeatMeasuredRate {
//单例实例
private static HeartbeatMeasuredRate instance = new HeartbeatMeasuredRate();
//最近一分钟的心跳次数
private long latestMinuteHeartbeatRate = 0L;
//最近一分钟的时间戳
private long latestMinuteTimestamp = System.currentTimeMillis();
private HeartbeatMeasuredRate() {
Daemon daemon = new Daemon();
daemon.setDaemon(true);
daemon.start();
}
//获取单例实例
public static HeartbeatMeasuredRate getInstance() {
return instance;
}
//增加一次最近一分钟的心跳次数
public void increment() {
synchronized(HeartbeatMeasuredRate.class) {
latestMinuteHeartbeatRate++;
}
}
//获取最近一分钟的心跳次数
public synchronized long get() {
return latestMinuteHeartbeatRate;
}
private class Daemon extends Thread {
@Override
public void run() {
while(true) {
try {
synchronized(HeartbeatMeasuredRate.class) {
long currentTime = System.currentTimeMillis();
if (currentTime - latestMinuteTimestamp > 60 * 1000) {
latestMinuteHeartbeatRate = 0L;
latestMinuteTimestamp = System.currentTimeMillis();
}
}
Thread.sleep(1000);
} catch (Exception e) {
e.printStackTrace();
}
}
}
}
}
(3)监控服务是否存活的后台线程增加判断是否开启自我保护机制
//微服务存活状态监控组件
public class ServiceAliveMonitor {
...
//负责监控微服务存活状态的后台线程
private class Daemon extends Thread {
private ServiceRegistry registry = ServiceRegistry.getInstance();
@Override
public void run() {
Map<String, Map<String, ServiceInstance>> registryMap = null;
while(true) {
try {
//判断是否要开启自我保护机制
SelfProtectionPolicy selfProtectionPolicy = SelfProtectionPolicy.getInstance();
if (selfProtectionPolicy.isEnable()) {
Thread.sleep(CHECK_ALIVE_INTERVAL);
continue;
}
registryMap = registry.getRegistry();
for (String serviceName : registryMap.keySet()) {
Map<String, ServiceInstance> serviceInstanceMap = registryMap.get(serviceName);
for (ServiceInstance serviceInstance : serviceInstanceMap.values()) {
//说明服务实例距离上一次发送心跳已经超过90秒了
if (!serviceInstance.isAlive()) {
registry.remove(serviceName, serviceInstance.getServiceInstanceId());
//更新自我保护机制的阈值
synchronized(SelfProtectionPolicy.class) {
selfProtectionPolicy.setExpectedHeartbeatRate(selfProtectionPolicy.getExpectedHeartbeatRate() - 2);
selfProtectionPolicy.setExpectedHeartbeatThreshold((long)(selfProtectionPolicy.getExpectedHeartbeatRate() * 0.85));
}
}
}
}
Thread.sleep(CHECK_ALIVE_INTERVAL);
} catch (Exception e) {
e.printStackTrace();
}
}
}
}
}
详细介绍后端技术栈的基础内容，包括但不限于：MySQL原理和优化、Redis原理和应用、JVM和G1原理和优化、RocketMQ原理应用及源码、Kafka原理应用及源码、ElasticSearch原理应用及源码、JUC源码、Netty源码、zk源码、Dubbo源码、Spring源码、Spring Boot源码、SCA源码、分布式锁源码、分布式事务、分库分表和TiDB、大型商品系统、大型订单系统等
