# thread-concurrent-lecture **Repository Path**: single_forest/thread-concurrent-lecture ## Basic Information - **Project Name**: thread-concurrent-lecture - **Description**: java并发编程学习 - **Primary Language**: Java - **License**: Not specified - **Default Branch**: master - **Homepage**: None - **GVP Project**: No ## Statistics - **Stars**: 1 - **Forks**: 1 - **Created**: 2020-05-18 - **Last Updated**: 2020-12-19 ## Categories & Tags **Categories**: Uncategorized **Tags**: None ## README 1.java 应用程序的main函数是一个线程,在被JVM启动的时候调用,线程的名字叫main 2.实现一个线程,必须创建Thread实例,Override run方法,并且调用实例的start方法 3.在JVM启动后,实际上有多个线程,但是至少有一个非守护线程 4.当你调用一个线程start方法的时候,至少有两个线程,一个是调用你的线程,还有一个是执行run方法的线程 5.线程的生命周期分为new,runnable,running,block,terminate 临界区 一个程序运行多个线程本身是没有问题的,问题出在多个线程访问共享资源 多个线程读共享资源其实也没有问题,在多个线程对共享资源读写操作时发生指令交错,就会出现问题 一段代码块内如果存在对共享资源的多线程读写操作,称这段代码为临界区 竞态条件 多线程在临界区内执行,由于代码的执行序列不同而导致结果无法预测,称之为发生了竞态条件 静态方法内不能直接调用非静态方法,必须是创建方法所在的类对象,通过类对象调用 synchronized加在成员方法上等价于synchronized(this) synchronized加在静态方法上等价于synchronized(类.class)加在类的class对象上 常见线程安全类 String,Integer,StringBuffer,Random,Vector,Hashtable, 这里的线程安全是指多个线程调用他们同一个实例的某个方法,是线程安全的 Hashtable table = new Hashtable(); new Thread(() -> { table.put("key","value1") },"t1").start(); new Thread(() -> { table.put("key","value1") },"t2").start(); 它们的每个方法都是原子的,任一方法执行完毕前,不允许对象的其他方法调用 但是注意:他们多个方法的组合不是原子的 Hashtable table = new Hashtable(); new Thread(() -> { if(table.get("key") == null){ table.put("key","value1") } },"t1").start(); new Thread(() -> { if(table.get("key") == null){ table.put("key","value2") } },"t2").start(); 线程t1开始执行get("key")操作时不允许table对象的其他方法执行 ,当这个方法执行完成后释放锁,则可能发生上下文切换,开始执行t2的get("key")方法 ,执行完毕后释放锁,这时候,t1的put方法和t2的put方法都有权力竞争锁然后执行,使代码处于临界区 管程monitor 重量级锁(锁对象头状态:10): 当给一个临界区代码块加对象锁时,这个对象的对象头中的Mark Word 会指向一个monitor对象, monitor对象中又有三个部分: 1>owner用于记录当前所有线程的值; 2>entryList用于存储阻塞线程值;(保持顺序) 3>waitSet用于存储获得对象锁,但又执行条件不足的线程,如果条件满足后,会再次进入entryList阻塞(查找效率高) 轻量级锁(锁对象头状态:00): 如果一个对象虽然有多线程访问,但多线程访问的时间是错开的(也就是没有竞争),那么可以使用 轻量级锁来优化(如果有竞争,就会进入锁膨胀,升级为重量级锁) 轻量级锁对使用者是透明的,即语法仍然是synchronized 假设有两个同步代码块,利用同一个对象加锁 static final Object obj = new Object(); public static void method1(){ //第一次利用obj加锁 synchronized(obj){ //同步代码块 method2(); } } public static void method2(){ //第二次利用obj加锁 synchronized(obj){ //同步代码块 } } 轻量级锁底层: 1.创建锁记录(Lock Record)对象:线程调用每一个方法,都会创建这个方法的一个栈帧 ,每个栈帧都会包含一个锁记录的结构,内部可以存储锁对象头的Mark Word以及锁对象的地址(Object reference) 2.让锁记录中Object reference指向锁对象,并尝试用cas替换Object的Mark Word,将Mark Word的值存入锁记录 3.如果cas(compare and set 以原子方式设置为给定值，并返回旧值)替换成功,对象头中存储了(锁记录地址和状态 00) ,表示由该线程给对象加轻量级锁 4.如果cas失败,有如下两种情况: 1>如果是其他线程已经持有了该Object 的轻量级锁,这时表明有竞争,(自旋等待,自旋次数达到上限)进入锁膨胀过程,升级为重量级锁 2>如果是自己执行了synchronized 锁重入,那么再添加一条Lock Record 作为重入的计数 5.当退出synchronized代码块时(解锁时) 如果有取值为null的锁记录,表示有重入,这时重置锁记录,表示重入记录减一 6.当退出synchronized代码块时(解锁时) 锁记录的取值不为null,这时使用cas将Mark Word的值恢复给对象头有以下两种情况: 1>成功,则解锁成功 2>失败,说明轻量级锁进行了锁膨胀或已经升级为重量级锁,进入重量级锁解锁流程 锁膨胀操作: 如果在尝试加轻量级锁的过程中,cas操作无法完成,这时一种情况就是有其他线程为此对象加上了轻量级锁(有竞争) ,这时需要进行锁膨胀,将轻量级锁升级为重量级锁 1.当Thread-1对锁对象进行轻量级加锁时,Thread-0已经对该对象加了轻量级锁 2.然后进入自旋尝试获得锁阶段,当自旋超时后Thread-1加轻量级锁失败,进入锁膨胀流程 1>即为Object对象申请Monitor锁,让Object的对象头中Mark Word指向重量级锁 2>然后自己进入Monitor的EntryList BLOCK 3.当Thread-0退出同步代码块解锁时,使用cas将Mark Word的值恢复给对象头,失败.这时会进入重量级解锁流程 ,即按照锁对象头中的Monitor地址找到Monitor对象,设置Owner为null,(操作系统)唤醒EntryList中的BLOCK线程 自旋优化: 重量级锁竞争的时候,还可以使用自旋来进行优化,如果当前线程重试加锁自旋成功(即这时候持锁线程正好退出了同步块,释放了锁) ,这时当前线程就可以避免阻塞,这样就可以避免cpu进行上下文切换 注:1.在java6之后自旋锁是自适应的,如果线程刚刚一次的自旋成功过,那么认为这次自旋成功的可能性会高,就会增加自旋次数 ,反之就少自旋甚至不自旋 2.自旋会占用CPU时间,单核CPU自旋就是浪费,多核CPU自旋才能发挥优势 3.java7之后不能再人为控制是否开启自旋功能 偏向锁(锁对象头状态:01): 轻量级锁在没有竞争时,只有单一线程执行代码块情况下,每次重入仍然需要执行CAS操作,造成性能浪费 java6中引入了偏向锁来优化:只有第一次使用CAS时会将线程ID设置到对象Mark Word头,之后如果发现只有这个线程来执行代码块, 就表示不存在竞争,不用重新CAS,以后只要不发生竞争(没有其他线程来执行代码块),这个锁对象就归该线程所有 对象头格式 |-----------------------------------------------------------------|--------------------| | Mark Word(64 bits) | status | |-----------------------------------------------------------------|--------------------| | unused:25 | hashcode:31 | unused:1 | age:4 | biased_lock:0 | 01 | Normal | 未加锁 |-----------------------------------------------------------------|--------------------| | thread:54 | epoch:2 | unused:1 | age:4 | biased_lock:1 | 01 | Biased | 偏向锁 |-----------------------------------------------------------------|--------------------| | ptr_to_lock_record:62 | 00 | Lightweight Locked | 轻量级锁 |-----------------------------------------------------------------|--------------------| | ptr_to_heavyweight_monitor:62 | 10 | Heavyweight Locked | 重量级锁 |-----------------------------------------------------------------|--------------------| | | 11 | Marked for GC | 标记回收 |-----------------------------------------------------------------|--------------------| 一个对象创建时: 1.如果开启了偏向锁(默认开启),那么对象创建后,对象头中的mark word值为0x05即后3位为101,这时它的thread,epoch,age都为0 2.偏向锁是默认延迟的,不会在程序启动时立即生效,如果想避免延迟,可以加VM参数 -XX:BiasedLockingStartupDelay=0来禁用延迟 3.如果没有开启偏向锁,那么对象创建后,对象头中的mark word值为0x01即后三位为001,这时它的hashcode,age都为0 ,第一次使用hashcode方法时才会赋值 注: 1.处于偏向锁状态的对象解锁后,线程id仍存储于对象头中 2.调用hashcode方法会禁用掉对象的偏向锁状态而直接使用轻量级锁 3.当存在多个线程交替加锁,那么也会取消掉锁对象的偏向锁状态,进入轻量级锁状态 批量重偏向(针对于一个类的多个对象,如果多次撤销偏向,则在剩下的对象中重新偏向): 如果对象虽然被多个线程访问,但没有竞争,这时偏向了线程T1的对象锁仍有机会重新偏向T2,重偏向会重置对象的ThreadID 条件:当撤销偏向锁阈值超过20次后,jvm会检测到偏向错误,于是会在给对象加锁时,重新偏向至加锁线程 批量撤销(针对于一个类的多个对象,如果更多次撤销偏向,则在剩下的对象设置为不可偏向): 当撤销偏向锁阈值超过40次后,jvm会检测到偏向错误,导致不再发生偏向,于是整个类的所有对象都会加锁为不可偏向,类新建对象也是不可偏向的 锁消除: JIT 即时编译器会对运行时的java字节码进一步优化,如果发现变量不可能被共享,加锁没有意义,就会把锁优化掉 wait/notify 原理: 1.在重量级锁中,Owner线程发现条件不满足,调用wait方法,即可进入WaitSet变为WAITING状态 2.BLOCKED和WAITING的线程都处于阻塞状态,不占用CPU时间 3.BLOCKED线程会在Owner线程释放锁时被唤醒 4.WAITING线程会在Owner线程调用notify或notifyAll时被唤醒,但唤醒后并不是立刻获得锁,仍需进入EntryList重新竞争 API: 1.obj.wait()让成为obj锁owner的线程到waitSet等待 2.obj.notify()在obj锁上正在waitSet中等待的线程中随机挑一个唤醒 3.obj.notifyAll()让obj锁上正在waitSet等待的线程全部唤醒 4.obj.wait(timeout) 让成为obj锁owner的线程到waitSet等待一段时间 注:它们都是线程之间进行协作的手段,都属于Object对象的方法,必须获得此对象的锁,才能调用 使用姿势: //首先创建一个条件标识 flag static final Boolean flag = false synchronized(lock){ while(!flag){ lock.wait(); } //执行业务逻辑 } //另一个线程 synchronized(lock){ flag = true; lock.notifyAll(); } 注:使用以上格式调用wait()方法,可以防止唤醒线程错误,避免虚假唤醒 park,unpark方法 每个线程对象都会有一个park对象,park对象由三部分组成,_counter(标记位),_cond(阻塞条件)和 _mutex(互斥锁) 1.当运行状态的thread调用 park()方法 1>.当前线程调用Unsafe.park()方法 2>.检查_counter,当其值为0,则线程获得互斥锁 3>.线程进入_cond条件变量阻塞 4>.设置_counter = 0 2.当处于park状态的thread 调用 unpark() 方法 1>.调用Unsafe.unpark(thread)方法,设置_counter为1 2>.唤醒_cond条件变量中的thread 3>.thread恢复运行 4>.设置_counter为0 3.当处于运行状态的thread先调用 unpark() 方法,再调用park()方法 1>.调用Unsafe.unpark(thread)方法,设置_counter为1 2>.当前线程调用Unsafe.park()方法 3>.检查_counter,这时_counter的值为1,则线程无需阻塞,继续运行 4>.设置_counter为0 同步模式之保护性暂停: 即Guarded Suspension,用在一个线程等待另一个线程的执行结果 要点: 1.有一个结果需要从一个线程传递到另外一个线程,让他们关联同一个GuardedObject 2.如果有结果不断从一个线程到另外一个线程那么可以使用消息队列(见生产者/消费者) 3.JDK中,join的实现,Future的实现,采用的就是此模式 4.因为要等待另一方的结果,因此归类到同步模式 join方法源码解释 public final synchronized void join(long millis) throws InterruptedException { long base = System.currentTimeMillis(); long now = 0; if (millis < 0) { throw new IllegalArgumentException("timeout value is negative"); } if (millis == 0) { //如果方法参数为0 while (isAlive()) { wait(0); //当调用join方法的线程对象为存活时,执行到这一行代码的线程将陷入循环,直到调用对象状态为死亡 } } else { //如果方法参数不为0,入参为等待时间 while (isAlive()) { long delay = millis - now; if (delay <= 0) { break; } wait(delay);//执行到这一行代码的线程等待的时间为:从进入方法到执行到这一行花费的时间 now = System.currentTimeMillis() - base; } } } 注:在此方法中,等待线程为执行程序的线程,锁对象为调用join方法的线程对象 Object对象的wait()方法和notify()及notifyAll()方法必须获得锁对象之后调用,否则抛出IllegalMonitorStateException(非法的管程状态) 饥饿:一个线程优先级太低,始终得不到CPU调度执行,也不能够结束 死锁问题,可以通过顺序加锁的方式得到解决,但可能引发饥饿问题 活锁问题,指两个线程相互改变对方的终止条件,导致相互无法终止的现象,可通过设置随机睡眠时间的方式解决 ReentrantLock (可重入锁) 相对于synchronized,ReentrantLock具备如下特点: 1.可中断 2.可以设置超时时间 3.可以设置为公平锁(先进先出) 4.支持多个条件变量 与synchronized一样,都支持可重入 基本语法: //获取锁 reentrantLock.lock(); try { //临界区 }finally{ //释放锁 reentrantLock.unlock(); } 可重入: 是指同一个线程如果首次获得了这把锁,那么因为它是这把锁的拥有者,因此有权力再次获取这把锁 如果是不可重入,那么第二次获取锁时,自己也会被挡住 注:将reentrantLock设置为公平锁会降低并发度,不建议使用 条件变量: synchronized中也有条件变量,就是原理中的waitSet休息室,当条件不满足时进入waitSet等待 ReentrantLock的条件变量比synchronized强大之处在于,它是支持多个条件变量的,这就好比 1>synchronized是那些不满足条件的线程都在一间休息室等条件满足,会有虚假唤醒的问题 2>而ReentrantLock支持多间休息室,可以按照已满足的条件,唤醒对应休息室内的线程 使用流程: 1.await前需要获得锁; 2.await执行后,会释放锁,进入conditionObject等待; 3.await的线程被唤醒(或打断,或超时)去重新竞争lock锁; 4.竞争lock锁成功后,从await后继续执行 五.Java内存模型 JMM即Java Memory Model,它定义了主存,工作内存抽象概念,底层对应着CPU寄存器,缓存,硬件内存,CPU指令优化等. JMM体现在以下几个方面 1.原子性-保证指令不会受到线程上下文切换的影响 2.可见性-保证指令不受到cpu缓存的影响 3.有序性-保证指令不会受cpu指令并行优化的影响 可见性(volatile): 它可以用来修饰成员变量和静态变量,可以避免线程从自己的工作缓存中查找变量值,必须到主存中读取 成员变量的值,线程操作volatile变量都是直接操作主存 可见性VS原子性 可见性保证的是在多个线程之间,一个线程对volatile成员变量的修改对另一个线程可见,不能保证原子性,仅用在一个写线程,多个读线程的情况 而原子性主要在解决指令交错问题,保证指令对变量的操作都是顺序执行,当前线程读取到的必须是上个线程执行完的结果 volatile不能保证原子性,而synchronized可以保证原子性和可见性 例如:两个线程,一个i++,另外一个i--,假设成员变量i的初始值为0; 当第一个线程从主存中读取到i为0时,发生了上下文切换 第二个线程也从主存中读取到i的值为0,然后执行了i--操作,值变为-1,但是在线程将i的值写回成员变量之前,发生上下文切换 这时,第一个线程开始执行i++操作,将线程中值变为1,并写回成员变量i,第一个线程执行完毕, 然后第二个线程执行写入操作,将自己线程的结果-1,写回变量i,这样i的最后结果为-1 以上这个例子的产生原因为,线程一的运算操作,执行在线程二的写回操作之前,导致发生并发问题(所以必须要在同步情况下才能保证原子性) 保证可见性原理： 1.写屏障(sfence)保证在该屏障之前的,对共享变量的改动,都同步到主存当中 2.读屏障(lfence)保证在该屏障之后，对共享变量的读取，加载的是主存中最新数据 保证有序性原理： 1.写屏障会确保指令重排序时,不会将写屏障之前的代码排在写屏障之后 2.读屏障会确保指令重排序时,不会将读屏障之后的代码排在读屏障之前 JMM:Java内存模型 happens-before: happens-before规定了对共享变量的写操作对其他线程的读操作可见,它是可见性与有序性的一套规则总结 ,抛开happens-before规则,JMM(Java内存模型)并不能保证一个线程对共享变量的写,对于其它线程对该 共享变量的读可见 1.线程解锁m之前对变量的写,对于接下来对m加锁的其他线程对该变量的读可见 static int x; static Object m = new Object; new Thread(() -> { synchronized(m) { x = 10; } },"t1").start(); new Thread(() -> { synchronized(m){ System.out.println(m); } },"t2").start(); 2.线程对volatile变量的写,对接下来其它线程对该变量的读可见 volatile static int x; new Thread(() -> { x = 10; },"t1").start(); new Thread(() -> { System.out.println(m); },"t2").start(); 3.线程start前对变量的写,对该线程开始后对该变量的读可见 static int x; x = 10; new thread(() -> { System.out.println(x) },"t2").start(); 4.线程结束前对变量的写,对其他线程得知它结束后的读可见(比如其他线程调用t1.isAlive()或t1.join()等待它结束) static int x; Thread t1 = new Thread(() -> { x = 10; },"t1"); t1.start(); t1.join(); System.out.println(x); 5.线程t1打断t2(interrupt) 前对变量的写,对于其它线程得知t2被打断后对变量的读可见(通过t2.interrupted()或t2.isInterrupted()) static int x; Thread t2 = new Thread(() -> { while(true){ if(Thread.currentThread().isInterrupted()){ System.out.println(x); break; } } },"t2"); t2.start(); new Thread(() -> { sleep(1); x = 10; t2.interrupt(); },"t1").start(); while(t2.isInterrupted()){ //使当前线程从执行状态（运行状态）变为可执行态（就绪状态） //这里使为了尽量使t1,t2执行,方便测试 Thread.yield(); } System.out.println(x); 6.对默认变量默认值(0,false,null)的写,对其它线程对该变量的读可见 7.具有传递性,如果x hb -> y 并且y hb -> z 那么有x hb -> z,配合volatile的防止指令重排,有下面的例子 volatile static int x; static int y; new Thread(() -> { y = 10; x = 20; }."t1").start(); new Thread(() -> { // x = 20 对t2 可见,同时 y = 10也对t2可见 System.out.println(x) },"t2"); 分析代码线程安全性 //问题1:为什么类上加final //问题2:如果实现了序列化接口,还要做什么来防止反序列化破坏单例 public final class Singleton implements Serializable { //问题3:为什么设置为私有?是否能防止反射创建新的实例 private Singleton(){} //问题4:这样初始化是否能保证单例对象创建时的线程安全性 private static final Singleton INSTANCE = new Singleton(); //问题5:为什么提供静态方法而不是直接将INSTANCE 设置为public,说出你的理由 public static Singleton getInstance(){ return INSTANCE; } 注:在类中添加该方法可以防止反序列化破坏单例 public Object readResolve(){ return INSTANCE; } } 1.加final防止有子类,子类中不适当的覆盖某些方法,破坏单例 2.如果一个类实现的序列化接口,在反序列化时,会创建新的对象; 可以通过添加注中的readResolve()方法,如果在反序列化过程中,发现readResolve()方法,则直接使用这个方法的返回作为反序列化结果; 3.设置为私有,防止通过new的方式创建对象,破坏单例,但是这种方式并不能防止暴力反射破坏单例 4.没有线程安全问题,因为静态变量初始化操作是在类加载的阶段中完成,JVM会保证代码的线程安全性 5.1>用方法可以有更好的封装性; 2>可以内部实现一些懒惰的初始化; 3>可以创建单例时有更多的控制; 4>可以提供泛型的支持; DCL:double checked locking 双重检测锁,缩小加锁范围 推荐的单例懒汉式实现方式(静态私有内部类方式) public final class Singleton { private Singleton{} private static class LazyHolder { static final Singleton INSTANCE = new Singleton(); } public static Singleton getInstance(){ return LazyHolder.INSTANCE; } } 因为静态内部类的实例化默认是惰性的,必须是调用方法时才会第一次实例化 第五章:无锁并发技术CAS(compare and set) 为什么无锁效率高? 1.无锁情况下,即使重试失败,线程始终在高速运行,没有停歇,而synchronized会让线程在没有获得锁的 时候,发生上下文切换,进入阻塞; 2.但在无锁情况下,因为线程要保持运行,需要额外CPU的支持,CPU在这里就好比高速跑道,没有额外的跑道, 线程想高速运行也无从谈起,虽然不会进入阻塞,但由于没有分到时间片,仍然会进入可运行状态,还是会导致 上下文切换. CAS的特点: 结合CAS和volatile可以实现无锁并发,适用于线程较少,多核CPU的场景下. 1.CAS是基于乐观锁的思想:最乐观的设计,不怕别的线程来修改共享变量,就算改了也没关系,可以在改完的 基础上发起重试 2.synchronized是基于悲观锁的思想:最悲观的设计,得防着其他线程来修改共享变量,我上了锁你们都别 想改,我改完了释放锁,你们才有机会 3.CAS体现的是无锁并发,无阻塞并发,请仔细体会如下两句话: 1>因为没有使用synchronized,所以线程不会陷入阻塞,这是效率提升的因素之一 2>但如果竞争激烈,可以想到重试必然频繁发生,反而会影响效率 4.原子整数 AtomicInteger 5.原子引用 AtomicReference AtomicStampedReference AtomicMarkableReference 6.原子数组 AtomicIntegerArray AtomicLongArray AtomicReferenceArray 7.字段更新器 AtomicReferenceFieldUpdater //域 字段 AtomicIntegerFieldUpdater AtomicLongFieldUpdater 利用字段更新器,可以针对对象的某个域(Field)进行原子操作,只能配合volatile修饰的字段使用,否则会出现异常 8.原子累加器 LongAdder LongAccumulator DoubleAdder DoubleAccumulator 相比于AtomicLong性能提升的原因: CAS在竞争激烈时会导致竞争失败而重试,而LongAdder在有竞争时,设置多个累加单元,Thread-0累加Cell[0],Thread-1 累加Cell[1]...最后将结果汇总,这样在累加时操作不同的Cell变量,因此减少了CAS重试失败,从而提高性能 Contended:防止缓存行伪共享注解(@sun.misc.Contended)[LongAdder的Cell内部类注解] 在此注解的对象或字段的前后各增加128字节大小的padding(空位),从而让CPU将对象预读至缓存时占用不同的缓存行(一般为64byte), 虽然缓存行的失效是以行为整理失效,但是缓存在不同的缓存行,这样就避免了缓存失效的相互影响 注:判断代码块有无竞争,可以把代码块用(原子类[AtomicReference,AtomicStampedReference,AtomicMarkableReference])对象 包裹,判断代码块是否执行成功,如果有失败情况,说明有竞争发生 底层CAS操作类,UnSafe,只能通过反射得到 Field theUnsafe = Unsafe.class.getDeclaredField("theUnsafe"); theUnsafe.setAccessible(true); Unsafe unsafe = (Unsafe) theUnsafe.get(null); 第六章:不可变类 1.final的使用:final关键字可以使用在类,变量上 1>.变量用final修饰保证了该属性是只读的,不能修改 2>.类用final修饰则无法被继承,保证了该类中的方法不能被覆盖,防止子类无意间破坏不可变性 保护性拷贝:通过创建副本对象来避免共享的手段 不可变类的二要素:final修饰类和类中变量,使用保护性拷贝 2.享元模式: 1>.23种设计模式之一,当需要重用数量有限的同一类对象时,最小化内存的使用,尽可能的对相同值的对象共享 2>体现:包装类 在JDK中Boolean, Byte, Short, Integer, Long, Character 等包装类提供了valueOf方法,例如在Long的 valueOf会缓存-128~127之间的Long对象,在这个范围之间会重用对象,大于这个范围,才会新建Long对象 注意: Byte,Short,Long缓存的范围都是-128~127 Character缓存的范围是0~127 Integer的默认范围是-128~127, 最小值不能变,但最大值可以通过调整虚拟机参数-D java.lang.Integer.IntegerCache.high来改变 Boolean缓存了TRUE和FALSE 3.final原理 final变量的赋值操作:使用了putField指令来完成,并在指令之后加入写屏障,保证在其他线程读取时,不会取到变量默认的初始值 final变量的取值操作:获取final修饰的共享成员变量 1>.如果该共享变量的值比较小则直接将该值复制到该方法的栈内存中,并使用BIPush命令获取 2>.如果该共享变量的值大于短整型则会复制该值到引用类的常量池中,并使用LDC指令获取 4.无状态: 在web阶段学习时,设计Servlet时为了保证其线程安全会建议,不要为Servlet设置成员变量,这种没有任何成员变量的类是线程安全的 因为成员变量保存的数据也可以称为状态信息,因此没有成员变量就称之为[无状态] 第七章:线程池 线程池使用了享元模式的思想,对线程进行重用,主要有存储任务的任务队列(Blocking Queue),执行任务的worker集合(Thread Pool)组成; 1.线程池内部结构 1>.任务队列:平衡任务生产者和消费者之间的速度差异 2>.worker集合:存储了执行任务的线程 3>.核心线程数coreSize:记录worker集合中的线程总量 4>.timeout:工作线程获取任务等待时间 5>.timeUnit:等待时间的单位 2.execute方法执行任务逻辑: 1>.当任务数没有超过核心线程数coreSize时(即worker集合的长度小于coreSize时),直接交给worker对象执行 2>.如果任务数超过coreSize时,将任务加入任务队列暂存起来 3.worker线程执行任务逻辑: 1>.当task任务不为空,则执行任务 2>.当task任务执行完毕,再接着从任务队列获取任务并执行 3>.当worker从队列中也获取不到任务了,则从worker集合中移除worker线程对象 代码如下: while (task != null || (task = taskQueue.take()) != null){ try { log.info("正在执行...{}",task); task.run(); }catch (Exception e){ e.printStackTrace(); }finally { //任务执行完毕,将task置为null task = null; } } synchronized (workers){ log.info("worker被移除{}",this); workers.remove(this); } 4.线程池拒绝策略:当任务队列达到最大长度时的策略(避免线程池影响到主线程的执行) 1>.死等 put()------不友好,会阻塞主线程的执行 2>.带超时的等待 offer() 3>.让调用者放弃任务执行 4>.让调用者抛出异常 5>.让调用者自己执行任务 抽象出来一个函数式方法,让线程池创建时传入,由调用者自己决定拒绝策略 @FunctionalInterface interface RejectPolicy{ void reject(BlockingQueue queue,T task); } 5.JDK线程池类结构 ThreadPoolExecutor ExecutorService ScheduledExecutorService ThreadPoolExecutor ScheduledThreadPoolExecutor 1>线程池状态 ThreadPoolExecutor使用int(一共32位)的高3位来标识线程池状态,低29位标识线程数量 状态名 高3位 接受新任务 处理阻塞队列任务 说明 RUNNING 111 Y Y SHUTDOWN 000 N Y 不会接收新任务,但会处理阻塞队列剩余任务 STOP 001 N N 会中断正在执行的任务,并抛弃阻塞队列任务 TIDYING 010 - - 任务全执行完毕,活动线程为0即将进入终结 TERMINATED 011 - - 终结状态 从数字上比较,TERMINATED > TIDYING > STOP > SHUTDOWN > RUNNING 将线程池状态和线程数量存储在一个变量ctl中,目的是将线程状态与线程个数合二为一,就可以通过一次cas原子操作进行赋值 // c为旧值,ctlOf返回结果为新值 ctl.compareAndSet(c,ctlOf(targetState,workerCountOf())); // rs为高3位代表线程池状态, wc为低29位代表线程个数,ctl是合并它们 private static int ctlOf(int rs,int wc){return rs | wc;} 2>构造方法 public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit, BlockingQueue workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler) * corePoolSize 核心线程数目(最多保留的线程数) * maximumPoolSize 最大线程数目 * keepAliveTime 生存时间-针对救急线程 * unit 时间单位-针对救急线程 * workQueue 阻塞队列 * threadFactory 线程工厂-负责创建线程(Executors.defaultThreadFactory) * handler 拒绝策略 3>线程池处理任务 * 线程池中刚开始没有线程,当一个任务提交给线程池后,线程池会创建一个新线程来执行任务. * 当线程数达到corePoolSize,时并且没有空闲线程,再加入新任务,则会被加入workQueue队列排队,直到有空闲的线程. * 如果队列选择了有界队列,那么当任务超过了队列大小时,会创建maximumPoolSize - corePoolSize数目的线程来救急. * 如果线程达到maximumPoolSize 仍然有新任务这时会执行拒绝策略,jdk提供了4种实现. * 当高峰过去后,超过corePoolSize的救急线程如果一段时间没有任务做,需要结束以节省资源,这个存活时间由keepAlive和unit控制. 任务提交优先级:核心线程 > 任务队列 > 救急线程 > 拒绝策略 救急线程的结束时机:救急线程被创建后会协助核心线程处理任务,直到等待队列中任务执行完毕,才会开始进入等待结束阶段 4>jdk提供的四种拒绝策略 * AbortPolicy 让调用者抛出RejectedExecutionException异常(默认) * CallerRunsPolicy 让调用者运行任务 * DiscardPolicy 放弃本次任务 * DiscardOldestPolicy 放弃队列中最早的任务,本任务取而代之 5>其它框架提供的拒绝策略实现 * Dubbo: 抛出RejectedExecutionException异常之前会记录日志,并dump线程栈信息,方便定位问题 * Netty: 创建一个新线程来执行任务 * ActiveMQ: 带超时等待(60s)尝试放入队列 * PinPoint: 使用一个拒绝策略链,会逐一尝试策略链中每一种拒绝策略 6.JDK提供的线程池默认实现(Executors类) 1>.newFixedThreadPool public static ExecutorService newFixedThreadPool(int nThreads){ return new ThreadPoolExecutor(nThreads,nThread, 0L,TimeUnit.MILLISECONDS,new LinkedBlockingQueue()); } 特点: * 核心线程数 = 最大线程数(没有救急线程被创建),因此也无需超时时间 * 阻塞队列是无界的,可以放任意数量的任务 适用于任务量已知,相对耗时的任务 2>.newCachedThreadPool public static ExecutorService newCachedThreadPool(){ return new ThreadPoolExecutor(0,Integer.MAX_VALUE,60L,TimeUnit.SECONDS, new SynchronousQueue()); } 特点: * 核心线程数是0,最大线程数是Integer.MAX_VALUE,救急线程的空闲生存时间是60s,意味着 ** 全部都是救急线程(60s后可以回收) ** 救急线程可以无限创建 * 队列采用了SynchronousQueue 实现特点是,没有容量,没有线程来取是放不进去的(一手交钱,一手交货) * 实现任意数量任务的原理是,会创建任意数量的线程,与固定大小任务的线程池通过无界队列来实现有所不同 整个线程池的表现为线程数会根据任务量不断增长,没有上限,当任务执行完毕,空闲一分钟后释放线程 适合用于任务数比较密集,但每个任务执行时间较短的情况 3>.newSingleThreadExecutor public ExecutorService newSingleThreadExecutor(){ return new FinalizableDelegatedExecutorService( new ThreadPoolExecutor(1,1,0L,TimeUnit.MILLISECONDS,new LinkedBlockingQueue())); } 使用场景: 希望多个任务串行执行.线程数固定为1,任务数多余1时,会放入无界队列排队,任务执行完毕,这唯一的线程也不会被释放 区别: * 自定义一个单线程串行执行任务,如果任务执行失败而终止那么没有任何补偿措施,而JDK线程池还会新建一个线程,保证正常工作 * Executor.newSingleThreadExecutor()线程个数始终为1,不能修改 ** FinalizableDelegatedExecutorService使用了装饰者模式,只对外暴露了ExecutorService接口,因此不能 调用ThreadPoolExecutor中特有的方法 * 而Executor.newFixedThreadPool(1)初始线程个数也为1,但还可以修改 ** 对外暴露的是ThreadPoolExecutor对象,可以强转后调用setCorePoolSize等方法进行修改 7.任务调度 1>执行任务: void execute(Runnable command); 2>提交任务task,用返回值Future获得任务执行结果 Future submit(Callable task); 3>提交tasks中所有任务 List> invokeAll(Collection> tasks) throws InterruptedException; 4>提交tasks中所有任务,带超时时间 List> invokeAll(Collection> tasks,long timeout,TimeUnit unit) throws InterruptedException; 5>提交tasks中所有任务,那个任务先成功执行完毕,返回此任务执行结果,其它任务取消 T invokeAny(Collection> tasks) throws InterruptedException,ExecutionException; 6>提交tasks中所有任务,那个任务先成功执行完毕,返回此任务执行结果,其它任务取消,带超时时间 T invokeAny(Collection> tasks,long timeout,TimeUnit unit) throws InterruptedException,ExecutionException; * execute()方法和submit()方法的区别 ** execute()方法执行任务没有返回值,submit()方法执行任务可以有返回值 ** execute()方法执行任务过程中,如果发生异常,会直接打印异常信息,并销毁此线程,并重新创建一个新线程填充入worker集合 submit()方法执行任务过程中,如果发生异常,不会打印异常信息,当返回的Future对象调用get()方法时,会抛出异常信息 8.关闭线程池 1>.void shutdown() 将线程池状态变为SHUTDOWN(不会接受新任务,但已提交的任务会执行完),此方法不会阻塞调用线程的执行 有以下三步操作: - 修改线程池状态 - 打断空闲线程 - 尝试终结(没有运行的线程,则立即终结线程池,如果还有运行的线程,也不会阻塞等待) 2>.List shutdownNow() 将线程池状态变为STOP(不会接受新任务,会将队列中的任务返回,并用interrupt的方式中断正在执行的任务) 有以下几步操作: - 修改线程池状态 - 打断所有线程 - 获取队列中剩余任务 - 尝试终结 tryTerminate(); 3>.其它方法 boolean isShutdown(); 只要是不在RUNNING状态的线程池,此方法就返回true boolean isTerminated(); 线程池状态是否是 TERMINATED boolean awaitTermination(long timeout,TimeUnit unit) throws InterruptedException; 调用shutdown()后,由于调用线程并不会等待所有任务运行结束,因此如果想在线程池terminated后做些事情,可以利用此方法等待特定时间 9.异步模式之工作线程模式 定义:让有限的工作线程(Worker Thread)来轮流异步处理无限多的任务;也可以将其归类为分工模式,它的典型实现就是线程池,也体现了经典 设计模式中的享元模式. *注意:不同任务类型应该使用不同的线程池,这样能够避免饥饿,并能提升效率 饥饿现象:固定大小的线程池会有饥饿现象 9.线程池设置多少最大线程数量合适 * 数量过小会导致程序不能充分地利用系统资源,容易导致饥饿 * 数量过大或导致更多的线程上下文切换,占用更多内存 1>.CPU密集型运算 通常采用cpu 核数 + 1 能够实现最优的CPU利用率, +1是保证当线程由于页缺失故障(操作系统)或其它原因导致暂停时,额外的这个线程就能 顶上去,保证CPU时钟周期不被浪费 2>.I/O密集型运算 CPU不总是处于繁忙状态,例如,当你执行业务计算时,这时会占用CPU资源,但当你执行I/O操作时,远程RPC调用时(网络I/O),包括进行数据库操作 时,这时候CPU就闲下来了,你可以通过设置合理的线程数量提高它的利用率. 计算公式如下: 线程数 = 核数 * 期望CPU利用率 * 总时间(CPU计算时间 + 等待时间) / CPU 计算时间 例如: 4核CPU计算时间是50%,其它等待时间是50%,期望cpu被100%利用,套用公式 4 * 100% * (100% / 50%) = 8 10.任务调度线程池 在[任务调度线程池]功能加入之前,可以使用java.util.Timer来实现定时功能,Timer的优点在于简单易用,但由于所有任务都是由同一个线程 来调度,因此所有任务都是串行执行的,同一时间只能有一个任务在执行,前一个任务的延迟或异常都会影响到之后的任务 JDK提供了ScheduledExecutorService 带调度功能的线程池 ScheduledExecutorService.schedule() 延时执行任务 ScheduledExecutorService.scheduleAtFixedRate() 以固定的速率执行任务 ScheduledExecutorService.scheduleWithFixedDelay() 以固定间隔执行任务 11.tomcat线程池 tomcat两大组件:1>连接器(Connector),2>servlet容器(Container) tomcat的连接器,其中的NIO EndPoint部分从前到后包含如下结构: 1>.LimitLatch用来限流,可以控制最大连接数,类似J.U.C中的Semaphore 2>.Acceptor只负责[接收新的socket连接] 3>.Poller只负责监听socket channel是否有[可读I/O事件] 4>.一旦发现可读事件,则封装为一个socketProcessor,提交给Executor线程池处理 5>.Executor线程池中的工作线程最终负责[处理请求] tomcat线程池扩展了ThreadPoolExecutor,行为稍微有些不同 * 如果中线程数量达到maximumPoolSize,并且等待队列已满 ** 这时并不会立刻抛出RejectedExecutionException异常 ** 而是自旋一次,尝试再次添加入等待队列,如果还失败,才抛出RejectExecutionException异常 * 线程池中所有线程都为守护线程 * 队列的长度默认是Integer.MAX_VALUE 相当于是个无界队列(建议调整为合理的数字) * 任务提交线程池逻辑有所区别 ** 如果提交的任务数小于核心线程数,则添加任务到队列,由核心线程去执行任务(添加到队列后会自主创建执行线程) ** 如果提交的任务数大于了核心线程数,则比较任务数和最大线程数,如果任务数小于最大线程数,则优先创建救急线程处理任务 ** 如果提交的任务数大于最大线程数,则将任务添加到任务队列 12.Fork/Join线程池 1>概念:Fork/Join是JDK1.7加入的新的线程池实现,它体现的是一种分治思想,适用于能够进行任务拆分的cpu密集型运算 所谓的任务拆分,是将一个大任务拆分为算法上相同的小任务,直至不能拆分可以直接求解,跟递归相关的一些计算,如归并排 序,斐波那契数列,都可以用分治思想进行求解 Fork/Join在分治的基础上加入了多线程,可以把每个任务的分解和合并交给不同的线程来完成,进一步提升了运算效率 因为分治思想适用于cpu密集型运算,所以Fork/Join默认会创建与cpu核心数大小相同的线程池 2>使用见 ForkJoinTest 测试类 第八章:JUC并发工具 1.AQS原理 全称是AbstractQueuedSynchronizer,是阻塞式锁和相关的同步器工具的框架(sync同步器) 内部维护了如下属性: //阻塞队列的头节点 1>.private transient volatile Node head; //阻塞队列的尾节点 2>.private transient volatile Node tail; //锁同步状态 3>.private volatile int state; //记录锁的所属线程对象,在其父类AbstractOwnableSynchronizer中维护 4>.private transient Thread exclusiveOwnerThread; 特点: * 用state属性来标识资源的状态(分独占模式和共享模式),子类需要定义如何维护这个状态,控制如何获取锁和释放锁 ** getState - 获取state状态 ** setState - 设置state状态 ** compareAndSetState - cas 机制设置state状态 ** 独占模式是只有一个线程能够访问资源,而共享模式可以允许多个线程访问资源 * 提供了基于FIFO的等待队列,类似于Monitor的EntryList * 条件变量来实现等待,唤醒机制,支持多个条件变量,类似于Monitor的WaitSet 子类主要实现这样一些方法(默认抛出UnsupportedOperationException) * tryAcquire * tryRelease * tryAcquireShared * isHeldExclusively 获取锁的姿势 // 如果获取锁失败 if(!tryAcquire(arg)){ // 入队,可以选择阻塞当前线程 park unpark } 释放锁的姿势 //如果释放锁成功 if(tryRelease(arg)){ //让阻塞线程恢复运行 } 2.ReentrantLock原理 ReentrantLock内含一个同步器属性sync,sync有两个实现,FairSync(公平同步器),UnfairSync(非公平同步器) 一个线程从执行ReentrantLock的lock()方法开始到进入阻塞状态,一共会进行3-4次尝试获得锁(lock方法,acquire方法,acquireQueue方法(两次尝试)) AQS源码阅读笔录: 1.AbstractQueuedSynchronizer类中维护了一个Node内部类,复用在锁阻塞队列和等待条件(Condition)中 在锁的阻塞队列中,以双向链表的结构维护(当前节点记录前驱节点地址及后继节点地址:通过prev和next标记前驱及后继节点) 在等待条件中,以单向链表的结构维护(当前节点只记录了后继节点的地址:通过nextWater标记下个节点) Node类维护了如下字段: 1>.int waitStatus 等待状态(-1状态为会负责唤醒下一个节点绑定线程) 2>.Node prev 前驱节点 3>.Node next 后继节点 4>.Thread thread 入驻该节点的线程 5>.Node nextWaiter 后继条件等待节点 Node(Thread thread, Node mode) { // Used by addWaiter this.nextWaiter = mode; this.thread = thread; } Node(Thread thread, int waitStatus) { // Used by Condition this.waitStatus = waitStatus; this.thread = thread; } 2.AbstractQueuedSynchronizer类中维护了如下字段: /** * 等待队列的头节点,懒惰初始化,除初始化外,只能通过setHead()方法修改 * 注意：如果head存在，则要保证其waitStatus不为取消状态 */ private transient volatile Node head; /** * 等待队列的尾节点,懒惰初始化,仅通过方法enq进行修改以添加新的等待节点。 */ private transient volatile Node tail; /** * 标记同步状态 */ private volatile int state; 3.addWaiter(Node mode)添加等待队列节点逻辑: 1>.根据当前线程初始化一个Node对象[new Node(Thread.currentThread(), mode)] 2>.获取当前锁对象记录的等待队列的尾节点Node 3>.如果尾节点对象的地址值不为null,则设置新创建的Node对象的前驱节点标记为原尾节点对象 * 调用compareAndSetTail(pred, node)原子操作,设置锁对象的尾节点对象为新的Node节点 * 然后设置原来尾节点Node对象的后继节点标记尾新的Node节点,完成双向链接,并返回Node节点对象 4>.如果尾节点对象的地址值尾null,则调用enq(node)方法,进入初始化等待队列流程 * 再一次获取锁对象的尾节点Node,判断是否尾null,如果为null则尝试初始化 * 调用compareAndSetHead(new Node())原子操作,如果返回true,则初始化成功,设置尾节点等于头节点 * 如果获取到的尾节点Node不为null,说明其它线程已经完成了初始化,则重复3>操作 4.acquireQueued 获得等待(获取锁失败后进入阻塞方法) boolean failed = true; try { boolean interrupted = false; for (;;) { //获取当前节点的前驱节点 final Node p = node.predecessor(); //判断前驱节点是否为头节点,如果为头节点,则开始竞争锁对象,竞争成功,进入下面逻辑,竞争失败,则再次进入下面的 //shouldParkAfterFailedAcquire(p, node)判断是否应该阻塞,结果为true时,进入parkAndCheckInterrupt()阻塞 if (p == head && tryAcquire(arg)) { //从此处可知,线程必须为获得锁后才会更新Node节点属性并释放其前驱节点 //设置头节点为当前节点Node setHead(node); //设置原头节点的后继节点属性为null,断开关联,方便可达性算法回收 p.next = null; // help GC //设置取消标识为false failed = false; //执行返回,true return interrupted; } //进入shouldParkAfterFailedAcquire重新检查等待队列并返回是否应该阻塞,true则进入parkAndCheckInterrupt()阻塞 if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) interrupted = true; } } finally { //如果在竞争锁逻辑中抛出异常,则进去取消竞争锁逻辑 if (failed) cancelAcquire(node); } //线程park,被唤醒后检查线程是否被打断,并清除打断标记 private final boolean parkAndCheckInterrupt() { LockSupport.park(this); return Thread.interrupted(); } 1>.正常获得锁逻辑: * 进入循环,检查当前节点Node是否为头节点的后继节点,如果时后继节点,则尝试获得锁, 获得锁成功,则处理等待队列逻辑并返回,如果获取锁失败,则进行下面逻辑 * 检查当前节点是否应该被park阻塞,如果返回true,则继续下面逻辑 * 线程进入parkAndCheckInterrupt方法并park阻塞 * 当有其它方法唤醒当前线程,则parkAndCheckInterrupt方法返回打断状态 2>.不可打断模式下打断线程阻塞逻辑: * 当线程被从park状态打断,则返回Thread.interrupted()[此方法会清除掉线程的打断状态]为true, 这种情况下,设置局部变量interrupted为true用来标记线程已经被打断(这里为什么清除线程的打断状 态而用局部变量interrupted标识,因为如果不清除线程的Interrupt标记,则调用park方法不生效) * 线程再次循环进入1>的逻辑,并在获得锁之后返回局部变量interrupted标识为线程的打断标识 * 外部方法在获取到线程的打断状态为true时,则调用selfInterrupt(),走打断逻辑流程 线程获取锁失败到陷入阻塞流程总结: 1.调用addWaiter初始化当前线程的Node对象 2.调用acquireQueued将Node对象链接到AQS队列尾部 3.调用shouldParkAfterFailedAcquire将前驱节点的状态更新为-1,作为唤醒依赖标识 5.条件变量实现原理Condition 每个条件变量其实都对应着一个等待队列,其实现类是ConditionObject 中有两个属性 //记录条件队列的头节点Node private transient Node firstWaiter; //记录条件队列的尾节点Node private transient Node lastWaiter; await流程 开始线程持有锁,调用await 1.进入ConditionObject的 addConditionWaiter流程创建新的Node状态为-2(Node.CONDITION),关联当前线程,加入等待队列尾部 2.进入AQS的fullRelease流程,释放同步器上的锁 3.unpark AQS队列中的下一个节点,竞争锁,假设没有其它竞争线程,则第二节点的线程竞争成功 public final void await() throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); Node node = addConditionWaiter(); int savedState = fullyRelease(node); int interruptMode = 0; while (!isOnSyncQueue(node)) { LockSupport.park(this); if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) break; } 注:线程被唤醒后,会携带进入条件队列时的加锁次数,尝试获得锁,并将status设置为加锁次数,用于后面 释放解锁(因为可能线程在发生锁重入之后才进入await,重新唤醒后,一定会进入解锁过程,加锁多少次就调用多少次解锁,不至于将status减为负数) if (acquireQueued(node, savedState) && interruptMode != THROW_IE) interruptMode = REINTERRUPT; if (node.nextWaiter != null) // clean up if cancelled unlinkCancelledWaiters(); if (interruptMode != 0) reportInterruptAfterWait(interruptMode); } signal流程 假设Thread-1要来唤醒Thread-0 1.检查调用方法的线程是不是锁的持有者,否则抛出异常(同synchronized关键字一样,必须获得锁之后才能调用条件变量的方法) 2.进入ConditionObject的doSignal流程,如果等待队列中有多个节点,则获取第一个等待节点Node * 设置条件变量的firstWaiter属性为等待队列节点的下一个节点,如果值为null,则设置lastWaiter也为null * 将Node节点的nextWaiter设置为null,断开和下一个等待节点的关联(出队) * 然后将Node节点转移到AQS队列尾部,如果转移失败,并且条件变量中还有等待线程Node,则尝试获取下一个等待线程Node 3.transferForSignal流程,将该Node加入AQS队列尾部,并将其waitStatus改为0,将其前驱节点的waitStatus状态改为-1 3.读写锁(ReentrantReadWriteLock) 当读操作远远高于写操作时,这时候使用读写锁,让[读-读]可以并发,提高性能;类似于数据库中的:select...from ...lock in share mode 提供一个[数据容器类] 内部分别使用读锁保护数据的read()方法,写锁保护数据的write()方法 代码详见class ReadWriteLockTest 注意事项: 1>.读锁不支持条件变量 2>.重入时升级不支持:即在持有读锁的请况下去获取写锁,会导致获取写锁永久等待 3>.重入时降级支持:即在持有写锁的情况下可以获取到读锁 实例详见class CachedData 拓展:缓存更新策略(更新时,是先清缓存还是先更新数据库) 1>.无锁状态下: * 先清缓存,再更新数据库 存在问题:如果清空缓存后,数据库更新操作还没执行前,正好有线程获取数据,则会把数据库的旧值同步到缓存,导致缓存还为旧值 * 先更新数据库,再清空缓存 存在问题:这种情况,可能在数据库更新完成后,缓存清空前,查询数据的线程会读取到缓存中的旧值,但是持续时间短,优于1>策略 2>.单机(java进程内)缓存数据库一致性问题解决(加读写锁) 还存在以下问题: * 适合读多写少,如果写操作比较频繁,性能较低 * 没有考虑缓存容量 * 没有考虑缓存过期(长时间不用的缓存应该清除掉) * 并发性能还是低,目前只用了一把锁 * 缓存的清除应该确定到唯一需要更新的数据,而不是直接全部清空(考虑按类型分区或重新设计key) 3.读写锁原理(ReentrantReadWriteLock) 读写锁看起来是两个锁,但其实用的是一个Sycn同步器,因此等待队列,state等也是同一个 读写锁底层标记获得锁状态,是通过state字段,其中写锁状态占了state的低16位,读锁状态占了state的高16位 写锁(WriteLock) 上锁流程 1.调用AQS的acquire方法尝试获得锁(内部调用为写锁中的tryAcquire方法) * 检查AQS同步器的state,及锁状态中的独占数量w,及当前锁的owner线程current * 如果state不为0,但是独占数量w为0,则是读锁定状态,无法锁升级,返回false * 如果state不为0,且独占数量w不为0,owner也不为当前线程,则返回false * 如果state不为0,且独占状态w不为0,owner就是current,则为重入情况,直接使用setState(newState)设置state,返回true * 如果state为0,则直接调用CAS操作,尝试修改state,修改成功,设置读写锁的owner线程为current返回true获得锁,否则返回false 2.尝试获得锁失败,创建独占属性Node节点对象(Node对象的nextWaiter属性为null), 添加到AQS队列中,并修改前驱节点waitStatus为-1,然后进入park流程 解锁流程 1.调用AQS的release方法释放锁(内部调用为写锁中的tryRelease方法) * 校验调用解锁的线程是否时锁持有线程,否则抛出异常 * 获取同步器AQS的state状态,减1操作并设置回同步器 * 检查exclusiveCount(独占数量)是否为0,为0说明所有重入数都已释放,设置锁持有线程为null,并返回true,否则返回false * 如果tryRelease方法返回false,则表示是重入锁释放,release方法直接返回false * 如果tryRelease方法返回true,则表示释放锁成功,则获取头节点h,并进入unparkSuccessor(h)流程 唤醒流程: 1.唤醒AQS等待队列中的写线程然后调用上锁流程中的1,尝试获得锁,并在获得锁成功之后,设置AQS的头节点为当前线程Node, 并将原头节点出队列,设置其next属性为null 读锁(ReadLock) 上锁流程 1.调用AQS的acquireShared方法尝试获得锁(内部调用为读锁中的tryAcquireShared方法) * tryAcquireShared方法返回一个整数值,成功返回1失败返回-1 * 检查AQS同步器的state,锁状态中的独占数量w,当前锁的owner线程current * 如果独占状态不为0,则检查锁的持有线程是否为current,如果不是当前线程,则返回-1 * 锁状态中的共享数量r,检查当前线程是否应该阻塞,检查共享数量r是否超出最大值(65530) * 如果检查结果都true,则CAS操作修改state尝试获得锁,如果获取成功,则处理持有锁记录 * if (exclusiveCount(c) != 0 && getExclusiveOwnerThread() != current) return -1; 由此段代码可知,当发生锁降级时,仅仅是设置了当前写锁的持锁对象为当前线程,其它获取读锁的线程仍然被阻塞 2.尝试获得锁失败,则创建共享属性的Node节点对象(Node对象的nextWaiter属性为new Node), 添加到AQS队列中,并修改前驱节点waitStatus为-1,然后进入park流程 释放锁流程 1.调用同步器的releaseShared方法 * 获取同步器的state状态,并进行减一操作,然后调用CAS操作,进行状态码设置,如果设置成功,则返回state==0的结果 * 如果state==0为true,则说明释放锁成功,执行doReleaseShared()(见下面唤醒流程) 唤醒流程 1.唤醒AQS等待队列中的读线程,之后到上锁流程的1,尝试获得锁 2.如果尝试获得锁失败,则继续调用parkAndCheckInterrupt方法阻塞 3.如果获得锁成功,则调用setHeadAndPropagate()方法重置同步器的head为当前Node,并获取当前Node的下一个节点s 判断下一个节点s的share属性为SHARE,则调用doReleaseShared()方法 * doReleaseShared()方法中,循环获取头节点,并检查头节点对象的waitStatus状态 * 如果为-1,则调用CAS方法尝试将头节点的waitStatus状态设置为0,如果成功,则调用unparkSuccessor(h)方法唤醒次节点 * 直到获取到的头节点等于AQS同步器的头节点(说明未能唤醒次节点,否则SHARE属性的Node必定会重置head),则退出循环,代码如下: private void doReleaseShared() { for (;;) { Node h = head; if (h != null && h != tail) { int ws = h.waitStatus; if (ws == Node.SIGNAL) { if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) continue; // loop to recheck cases unparkSuccessor(h); } else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) continue; // loop on failed CAS } if (h == head) break; // loop if head changed } } 总结: 1.读写锁中,读锁上锁调用的是acquireShared方法,而写锁调用的是acquire方法 2.获取读锁失败,线程是阻塞在doAcquireShared方法中,而获取写锁失败,线程是阻塞在acquireQueue方法中 3.因为读写锁中,读-读操作可并发,所有会有多个线程同时释放锁,并唤醒后继节点的操作,所以在进行唤醒操作前, 必须执行CAS操作将头节点的waitStatus状态设置为0,这样其它线程就不会执行唤醒次级节点操作,而是由CAS 成功的线程去执行唤醒操作,保证了不会重复唤醒 4.当读取线程在acquireShared方法中被唤醒后,会尝试获取读锁,如果获取锁成功,则执行setHeadAndPropagate() 方法,重新设置同步器的头节点,之后会判断次级节点的nextWaiter属性是否为Node.SHARE,如果是,则循环去执行 unparkSuccessor(h)唤醒下一个节点的方法,在这过程中,同步检查同步器的head节点和唤醒操作后的head节点是否 一致,如果一致,说明本次唤醒的线程获取锁失败,则停止循环(说明阻塞的一系列读锁都已经并发获得读锁,下一个节点是写锁) 5.readLock不支持条件变量,writeLock支持条件变量,且可重入 4.StampLock 该类自JDK8加入,是为了进一步优化读性能,它的特点是在使用读锁,写锁时,都必须配合[戳]使用 加解读锁 long stamp = lock.readLock() lock.unlockRead(stamp); 加解写锁 long stamp = lock.writeLock() lock.unlockWrite(stamp); 乐观读,StampedLock支持tryOptimisticRead()方法(乐观读),读取完毕后需要一次[验戳]如果验戳通过,表示这期间 确实没有写操作,数据可以安全的使用,如果校验没通过,才需要重新获取读锁,保证数据安全 long stamp = lock.tryOptimisticRead(); //验戳 if(!lock.validate(stamp)){ //锁升级 } 注:存在缺点 1.不支持条件变量 2.不支持可重入 5.Semaphore 信号量,用来限制能同时访问共享资源的线程上限 应用: 1>.使用Semaphore限流,在访问高峰期,让线程阻塞,高峰过去再释放许可,当然它只适合限制单机线程数量, 并且仅是限制线程数量,而不是限制资源数(例如连接数,对比Tomcat LimitLatch的实现) 2>.如果需要限制的线程数和资源数量[一致]的时候,使用Semaphore限流更好 * 用Semaphore实现简单连接池,对比[享元模式]下的实现(用wait/notify),性能和可读性显然更好, 注意:实现中线程数量和数据库连接数是相等的 代码见:CustomPoolTest 实现原理类似读写锁的读锁,只是通过state的值来限制共享线程数量 加锁逻辑: 1>.使用了AQS的同步器类,当调用构造方法创建对象的时候,会将构造参数的permits设置到AQS的state字段 2>.当调用acquire方法获取锁时,内部会调用Semaphore的nonfairTryAcquireShared方法 3>.内部会获取同步器的state值,然后做减法运算(如果是公平锁状态,则先会判断同步器是否有头节点,如果有,则直接返回-1) * 如果运算结果小于0 则直接返回运算结果, * 如果运算结果大于等于0,则尝试CAS加锁,如果成功,则返回运算结果,如果失败,循环执行3>操作 4>.如果返回的运算结果小于0,则调用doAcquireSharedInterruptibly方法,进入阻塞流程 解锁流程: 1>.调用同步器的sync.releaseShared(1)方法,内部调用tryReleaseShared 2>.tryReleaseShared中获取同步器的state值,并做加法运算,然后进行CAS操作,设置state值 * 如果成功,则直接返回true * 如果不成功,则再次进行2>操作,直到成功为止 3>执行tryReleaseShared成功,则调用doReleaseShared()进入循环唤醒Node.SHARE节点流程 5.CountdownLatch(倒计时锁) 用来进行线程间同步协作,等待所有线程完成倒计时 其中构造参数用来初始化等待计数值,await()用来等待计数归零,countDown()用来让计数减一 * 可应用于相互不依赖的流程同时执行,然后最终汇总结果 6.CyclicBarrier 循环栅栏,用来进行线程协作,等待线程满足某个计数.构造时设置[计数个数],每个线程执行到某个需要"同步" 的时刻调用await()方法进行等待,当等待的线程数满足[计数个数]时,继续执行 详细应用见:CyclicBarrierTest 注:要求执行线程数和构造参数计数数保持一致 7.线程安全集合类 目前可以分为三大类 * 遗留的线程安全集合如Hashtable,Vector * 使用Collections装饰的线程安全集合,如: ** Collections.synchronizedCollection ** Collections.synchronizedList ** Collections.synchronizedMap ** Collections.synchronizedSet ** Collections.synchronizedNavigableMap ** Collections.synchronizedNavigableSet ** Collections.synchronizedSortedMap(TreeMap)带排序的map ** Collections.synchronizedSortedSet(TreeSet)带排序的set * java.util.concurrent.* 本大类存在一定规律,包含三个关键词:Blocking,CopyOnWrite,Concurrent ** Blocking大部分实现基于锁,并提供用来阻塞的方法 ** CopyOnWrite之类容器修改开销相对较重 ** Concurrent类型的容器 *** 内部很多操作使用cas优化,一般可以提供较高吞吐量 *** 弱一致性 ***** 遍历时弱一致性,例如:当容器迭代器遍历时,如果容器发生修改,迭代器仍然可以继续进行遍历,这时内容是旧的 ***** 求大小弱一致性,size操作未必是100%准确 ***** 读取弱一致性 注:遍历时如果发生了修改,对于非安全容器来讲,使用fail-fast机制也就是让遍历立刻 失效,抛出ConcurrentModificationException,不再继续遍历 1>HashMap并发死链问题 * HashMap的底层数据结构为数组结构,数组的每个下标位置存储一个Node节点,put操作时,计算key值的hash码,然后与数组的长度 取模运算(底层采用了按位与的计算方式[hash码&size-1]),计算结果为当前Node存储到数组的对应下标,如果发生hash碰撞,则采 用拉链法的方式,将在同一个数组下标的元素串联起来(hash桶)形成链表结构; 注:不同版本JDk区别: ** JDK7中,会将新put进来的元素放在链表头部(导致了并发环境下扩容时的死链) ** JDK8中,会将新put进来的元素放在链表尾部 * 当HashMap中元素数量超过底层数组长度的3/4时,会进行一次扩容(二倍扩容),底层数组长度变为原来的二倍,并重新计算原有元素 的hash桶下标(key值hash码与数组长度取模),并将元素转移到对应数组下标下 注:为什么是二倍扩容? 因为计算桶下标时,使用取模运算,二倍扩容有利于位运算(如扩容前数组长度16,则桶下标为:hash码&[00001111],扩容后数组长度 为32,则桶下标为:hash码&[00011111]) * [并发死链],当两个线程AB同时去扩容时,如果A线程在迁移一个hash桶的元素时,B也同时迁移同一个桶下标的元素,可能造成链表形成 环状,导致并发死链,源码如下: // 将 table 迁移至 newTable void transfer(Entry[] newTable, boolean rehash) { int newCapacity = newTable.length; for (Entry e : table) { while(null != e) { Entry next = e.next; // 1 处 if (rehash) { e.hash = null == e.key ? 0 : hash(e.key); } int i = indexFor(e.hash, newCapacity); // 2 处 // 将新元素加入 newTable[i], 原 newTable[i] 作为新元素的 next e.next = newTable[i]; newTable[i] = e; e = next; } } } 假设链表元素如下 原始链表，格式：[下标] (key,next) [1] (1,35)->(35,16)->(16,null) 线程 a 执行到 1 处 ，此时局部变量 e 为 (1,35)，而局部变量 next 为 (35,16) 线程 a 挂起 线程 b 开始执行(头节点变尾节点) 第一次循环 [1] (1,null) 第二次循环 [1] (35,1)->(1,null) 第三次循环 [1] (35,1)->(1,null) [17] (16,null) 切换回线程 a，此时局部变量 e 和 next 被恢复，引用没变但内容变了：e 的内容被改为 (1,null)，而 next 的内容被改为 (35,1) 并链向 (1,null) 第一次循环 [1] (1,null) 第二次循环，注意这时 e 是 (35,1) 并链向 (1,null) 所以 next 又是 (1,null) [1] (35,1)->(1,null) 第三次循环，e 是 (1,null)，而 next 是 null，但 e 被放入链表头，这样 e.next 变成了 35 （2 处） [1] (1,35)->(35,1)->(1,35) 已经是死链了 * JDK8中对扩容算法做了调整,不再将元素从链表头加入,而是保持与扩容前一样的顺序,但并不能够保证在多线程 环境下能够安全扩容,还会出现其它问题(如扩容丢失数据) 1>ConcurrentHashMap(链表转化为树的阈值为8且数组长度大于等于64,取消树化的阈值为6) * 重要属性和内部类: //默认为0 //当初始化时,值为-1 //当扩容时,值为-(1 + 扩容线程数) //当初始化或扩容完成后,值为 下一次的扩容阈值大小(容量的3/4) private transient volatile int sizeCtl; //静态内部类 整个 ConcurrentHashMap 就是一个Node[](内部属性有,键,值,hash码,nestNode地址) static class Node implements Map.Entry{}; //Node类属性: ** final int hash; 标记Node中key的hash码spread(k.hashCode()),此方法保证了hash码为正整数 ** final K key; 存储元素信息Node的key值 ** volatile V val; 存储元素信息Node的value值 ** volatile Node next; 存储当前节点的下一个节点地址值,如果无下一个节点,则为null; //hash表 transient volatile Node[] table; //扩容时的新hash表 private transient volatile Node nextTable; //静态内部类 扩容时,如果某个bin迁移完毕,用ForwardingNode作为旧table bin的头节点[标记bin处理完成,同时get操作应该去nextTable] //当hash表在扩容时,如果某个bin处理完毕,则将hash表中原来的Node对象替换为ForwardingNode(hash属性值为-2) static final class ForwardingNode extends Node{} //静态内部类 用在compute 以及 computeIfAbsent时,用来占位,计算完成后替换为普通Node static final class ReservationNode extends Node{} //静态内部类 作为treeBin的头节点,存储root和first static final class TreeBin extends Node{} //静态内部类 作为treeBin 的子节点,存储parent,left,right static final class TreeNode extends Node{} * 重要方法: //获取Node[](hash表)中第i个Node static final Node tabAt(Node[] tab,int i) //cas 修改Node[] (hash表) 中第i个 Node 的值,c为旧值, v为新值 static final boolean casTabAt(Node[] tab,int i,Node c,Node v) //直接修改 Node[] 中第i个Node的值,v为新值 static final void setTabAt(Node tab, int t,Node v) * 构造方法: public ConcurrentHashMap(int initialCapacity, float loadFactor, int concurrencyLevel) { if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0) throw new IllegalArgumentException(); if (initialCapacity < concurrencyLevel) // Use at least as many bins initialCapacity = concurrencyLevel; // as estimated threads long size = (long)(1.0 + (long)initialCapacity / loadFactor); //如果超过最大容量,则取最大容量,如果没超过则调用tableSizeFor(int size),保证值的大小是2^n,即:16,32,64... int cap = (size >= (long)MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : tableSizeFor((int)size); this.sizeCtl = cap; } 分析:构造方法只是计算了table的大小,指定了sizeCtl的值,只有在第一次使用时才会创建Node[] * get方法 public V get(Object key) { Node[] tab; Node e, p; int n, eh; K ek; //计算key值的hash码,保证key的hash码是正整数 int h = spread(key.hashCode()); if ((tab = table) != null && (n = tab.length) > 0 && //通过按位与计算当前key所在hash表的下标 (e = tabAt(tab, (n - 1) & h)) != null) { if ((eh = e.hash) == h) { //比对是否为当前node if ((ek = e.key) == key || (ek != null && key.equals(ek))) return e.val; } //检查头node的hash属性是否小于零,-1为正在扩容,并且当前bin已经迁移完毕,需要去newTable查找,-2表示已转为红黑树存储 else if (eh < 0) return (p = e.find(h, key)) != null ? p.val : null; //当前存储结构为链表,则遍历链表查找 while ((e = e.next) != null) { if (e.hash == h && ((ek = e.key) == key || (ek != null && key.equals(ek)))) return e.val; } } return null; } 流程: ** 1.使用spread方法计算key值的hash码,然后同hash表长度减一(tab.length-2)进行按位与运算,得出桶下标 ** 2.使用tabAt(hash表,桶下标)方法获取对应下标下的Node头节点信息 ** 3.如果Node头节点信息为null,则表示无当前key对应的hash码的hash桶存在,则直接返回null ** 4.如果Node不为null,且Node对象的hash属性大于0 *** 则比对Node节点中的key值与查询的key,如果相同则返回,否则用equals方法比对,如果一致则返回 *** 如果头节点匹配失败,则通过头Node,获取nextNode,继续比对,遍历Node单向链表,查询并返回 ** 5.如果Node对象的hash属性小于0,则调用Node的find(tong下标,key)方法,查询并返回 *** 如果头Node的hash值时-1,则表示Node[]正在扩容,当前bin已经迁移完毕,需要取newTable去查找数据 *** 如果头Node的hash值为-1,则表示当前bin已经转为红黑树结构存储 * put(key,val)方法(内部调用putVal(key,val,boolean onlyIfAbsent)) final V putVal(K key, V value, boolean onlyIfAbsent) { //不允许存储null值null键 if (key == null || value == null) throw new NullPointerException(); //计算key的hash码 int hash = spread(key.hashCode()); int binCount = 0; for (Node[] tab = table;;) { Node f; int n, i, fh; //检查hash表是否为空 if (tab == null || (n = tab.length) == 0) //如果为空,初始化hash表 tab = initTable(); //计算桶下标,并获取对应下标位置的Node对象 else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { //如果对应桶下标位置的Node对象为null,则使用put参数初始化一个Node对象 //并使用casTableAt,尝试设置当前桶下标位置为此Node对象 if (casTabAt(tab, i, null, new Node(hash, key, value, null))) //如果设置成功,则break break; // no lock when adding to empty bin //如果失败,则重新进入下一轮循环 } else if ((fh = f.hash) == MOVED) //如果头节点的hash属性为-1状态,说明hash表在扩容状态,则调用帮助扩容方法,帮助扩容 tab = helpTransfer(tab, f); //帮助扩容完毕,重新进入下一轮循环(说明只要ConcurrentHashMap在扩容阶段,就不会有插入操作) else { //如果头节点不为空,且不在扩容阶段 V oldVal = null; //如果key已经存在,则将原val赋值oldVal,并返回 //尝试获得当前头节点对象锁,并执行插入 synchronized (f) { if (tabAt(tab, i) == f) { if (fh >= 0) { //如果数据结构为链表 binCount = 1; for (Node e = f;; ++binCount) { K ek; //检查是否已经存在当前key if (e.hash == hash && ((ek = e.key) == key || (ek != null && key.equals(ek)))) { oldVal = e.val; if (!onlyIfAbsent) e.val = value; break; } Node pred = e; if ((e = e.next) == null) { pred.next = new Node(hash, key, value, null); break; } } } //如果数据结构为树结构 else if (f instanceof TreeBin) { Node p; binCount = 2; if ((p = ((TreeBin)f).putTreeVal(hash, key, value)) != null) { oldVal = p.val; if (!onlyIfAbsent) p.val = value; } } } } if (binCount != 0) { //检查是否需要红黑树化 (树化阈值0 if (binCount >= TREEIFY_THRESHOLD) treeifyBin(tab, i); if (oldVal != null) return oldVal; break; } } } addCount(1L, binCount); return null; }