多线程与高并发相关

编程入门行业动态更新时间:2024-10-09 18:22:44

多线程与高并发相关

一、线程6个状态

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-7u0nsIEY-1610448936747)(AE35E23516974B748E3DF0FE82E92036)]
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-JuYkXlIo-1610448936750)(8321DCE40FF04C73B5ED87CE243F6ADB)]

线程状态详解

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-4U7G9cce-1610448936751)(40B7AF0DB00F45F88EFE2AFC8ED5018B)]

二、synchronized

1、特性

原子性（原子性就是指一个操作或者多个操作，要么全部执行并且执行的过程不会被任何因素打断，要么就都不执行）
可见性（可见性是指多个线程访问一个资源时，该资源的状态、值信息等对于其他线程都是可见的。）如何保证可见性？
有序性（有序性值程序执行的顺序按照代码先后执行。）
可重入性（通俗一点讲就是说一个线程拥有了锁仍然还可以重复申请锁。）

2、对于对象头的解释

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-o29dOel2-1610448936754)(AA91914CF1FD4B319158F1961662D15C)]
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-IJrQ0hDt-1610448936755)(07121E24995640DCA769E7F5E6A9DD73)]
在对象头的markword中的最低三位代表锁状态，其中一位是是否是偏向锁，两位是普通锁位

3、jdk1.6之后的优化

锁一共有四种状态，级别从低到高：无锁状态，偏向锁状态，轻量级锁状态，重量级锁状态，锁只能升级不能降级。

对于偏向锁的理解：在没有别的线程竞争的时候，一直偏向当前线程，当前线程可以一直执行

大多数情况下，锁不仅不存在多线程竞争，而且总是由同一线程多次获得，为了让线程获得锁的代价更低而引入偏向锁。
当一个线程访问同步代码块并获取锁时，会在对象头和栈帧中的锁记录里存储当前线程id， 
以后该线程再进入和退出同步块的时候不需要进行CAS操作来加锁和解锁，只需要简单测试一下对象头的Markword里是否存储着当前线程的偏向锁。
如果测试成功，表示线程已经获得了锁。
如果测试失败，则需要再测试一下Markword中偏向锁的标识是否为1： 
如果不是，则需要使用CAS去竞争锁；如果设置了，则尝试使用CAS将对象头的偏向锁指向当前线程。

对于轻量级锁（自旋锁）的理解：轻量级锁是由偏向锁升级来的，偏向锁运行在一个线程进入同步块的情况下，当第二个线程加入竞争的时候，偏向锁会升级为轻量级锁

轻量级锁实现方式为：各个线程在自己的线程栈里会生成一个LockRecord 
然后用CAS操作将markword设置为指向自己线程的LockRecord的指针  
设置成功则得到锁，没有设置成功会继续使用SCAS循环操作，默认为10次
jdk6对自旋进行了优化，自旋时间不再固定，由jvm动态监控，叫做自使用自旋

对于重量级锁的理解：因为自旋锁有个缺点就是等待的线程一直占用cpu自旋运行，如果线程数过多或者自旋次过多，这种情况不如向操作系统申请锁，阻塞其他线程来释放cpu资源，这个时候会升级重量级锁。（属于内核态级别，依赖操作系统底层的互斥锁来实现的

锁升级过程：最开始对象是无锁状态 --> 当一个线程准备对这个对象加锁前会去验证这个三个字节，发现此时是无锁状态，则会把是否偏向那一位设置为1，并把markword的线程id设置为当前线程id，此时对象就处于偏向锁状态–>若此时另一个线程来竞争该对象锁，则锁升级为自旋锁，竞争线程自旋等待锁的释放–>若自旋次数过多或者等待线程数过多则升级为重量级锁状态。
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-n6Tn4nmi-1610448936756)(D1402AC2C9444968B540A5896A41814B)]

4、为什么synchronized无法禁止指令重排，却能保证有序性（重点）

首先介绍现代计算机为了提升计算能力，会在硬件层面做很多优化，比如处理器优化和指令重排，这些技术的引入就会导致有序性问题
解决有序性最好的方法就是禁止指令重排序，就像volatile中使用内存屏障（synchronized没使用这些方式进行指令重排）
但是
synchronized为什么没用上述这些却依然能保证有序性？
是因为在java中，根据as-if-serial语义，不管怎么排序都不能影响单线程的执行结果，意思是单线程的执行结果永远不会变，这是前提
而synchronized是一种锁机制，它保证在执行同步代码块的时候只有一个线程能获得对象锁，并进入代码块执行逻辑，这就已经满足了as-if-serial语义（单线程），所以synchronized的有序性自然就保证了

三、volatile

保证内存可见性（保证了每次读写变量都从主内存中读）（缓存一致性协议，MESI）
缓存一致性机制整体来说，是当某块CPU对缓存中的数据进行操作了之后，就通知其他CPU放弃储存在它们内部的缓存，或者从主内存中重新读取，
禁止指令重排序（JMM中有八个指令完成数据的读写，通过其中load和store指令相互组合成的4个内存屏障实现禁止指令重排序）
无法保证原子性

public class DoubleCheckSingle {private static volatile DoubleCheckSingle instance;private DoubleCheckSingle() {}public static DoubleCheckSingle getInstance() {if (instance == null) {synchronized (DoubleCheckSingle.class) {if (instance == null) {instance = new DoubleCheckSingle();}}}return instance;}public static void main(String[] args) {for (int i = 0; i < 20; i++) {new Thread(() -> System.out.println(DoubleCheckSingle.getInstance().hashCode())).start();}}
}

上述双标志检查法中的volatile关键字需要加上，因为在instance = new DoubleCheckSingle();这个过程中有三个步骤：

在堆内存中分配内存
初始化对象
将引用指针指向内存地址

若不加上volatile可能会发生指令重排序的情况，还没初始化的时候就已经让引用指针指向内存地址导致之后的线程读出的对象数据有误。

四、CAS（乐观锁）

cas（内存地址，期望值，更新值）

cpu原语级别的（intel指令为：cmpxchg）需要借用底层本地方法来调用CPU底层指令，它的作用是将指定内存地址的内容与所给的某个值相比，如果相等，则将其内容替换为指令中提供的新值，如果不相等，则更新失败。这一比较并交换的操作是原子的，不可以被中断，而其保证原子性的原理就是上一节提到的“总线锁定和缓存一致性”。

cas通过硬件保证了原子性，硬件层面的操作比高级语言软件层面的操作快得多。

1、ABA问题

一个线程正在执行CAS操作，这时另一个线程执行操作把A变成B，又变为A

解决方案：增加版本号去控制。（如数据库里面增加版本号）

五、Lock

在使用ReentrantLock时，一定要手动释放锁
使用Synchrnized锁定的话如果遇到异常，jvm会自动释放锁，但是lock必须手动释放锁，因此经常在finally中进行锁的释放
可以使用tryLcok进行尝试锁定，在指定时间内会返回是否已经得到锁。不管锁定与否，方法都将继续执行
可以通过使用lock.lockInterruptibly()来实现这个等待中断机制。也就是说正在等待的线程可以选择放弃等待，改为处理其他事情。
可实现公平锁，创建锁的时候指定为true（synchronized只能是非公平锁，所谓公平意思就是先等待的线程优先获得锁，而非公平就是后来的线程也有机会先得到锁）

一些高级的多线程同步工具：

ReentrantLock
CountDownLatch
CyclicBarrier
Phaser
ReadWriteLock
Semaphore
Exchange
以上都是基于AQS
LockSupport

CountDownLatch（门栓）可以实现主线程让所有子线程执行完再继续执行，可在每个线程内countDown()一次（也可以使用join）

private static void usingCountDownLatch() {Thread[] threads = new Thread[100];CountDownLatch latch = new CountDownLatch(threads.length);for(int i=0; i<threads.length; i++) {threads[i] = new Thread(()->{int result = 0;for(int j=0; j<10000; j++) result += j;latch.countDown();});}for (int i = 0; i < threads.length; i++) {threads[i].start();}try {latch.await(); //阻塞等待所有子线程执行完} catch (InterruptedException e) {e.printStackTrace();}System.out.println("end latch");}

CyclicBarrier(栅栏)，当线程数满足栅栏设定数值才放行，不然线程一直等待栅栏放行，在每个线程内设置await把线程挡住在栅栏外。（也可以实现主线程让所有子线程执行完再继续执行）

public static void main(String[] args) {//CyclicBarrier barrier = new CyclicBarrier(20);CyclicBarrier barrier = new CyclicBarrier(20, () -> System.out.println("满人"));/*CyclicBarrier barrier = new CyclicBarrier(20, new Runnable() {@Overridepublic void run() {System.out.println("满人，发车");}});*/for(int i=0; i<19; i++) {new Thread(()->{try {barrier.await(); //此处会等待设定线程数到达才放行System.out.println("11111");} catch (InterruptedException e) {e.printStackTrace();} catch (BrokenBarrierException e) {e.printStackTrace();}}).start();}}

phaser指把一个任务分为多个阶段，在每一个阶段只有指定线程到达完后，才能进行下一个阶段，也可以在某一个阶段让指定线程不再继续执行。（CyclicBarrier的升级版）

package com.mashibing.juc.c_020;import java.util.Random;
import java.util.concurrent.Phaser;
import java.util.concurrent.TimeUnit;public class T09_TestPhaser2 {static Random r = new Random();static MarriagePhaser phaser = new MarriagePhaser();static void milliSleep(int milli) {try {TimeUnit.MILLISECONDS.sleep(milli);} catch (InterruptedException e) {e.printStackTrace();}}public static void main(String[] args) {phaser.bulkRegister(7); //注册7个线程，只有等到线程都到达了才继续执行下一个阶段for(int i=0; i<5; i++) {new Thread(new Person("p" + i)).start();}new Thread(new Person("新郎")).start();new Thread(new Person("新娘")).start();}static class MarriagePhaser extends Phaser {@Overrideprotected boolean onAdvance(int phase, int registeredParties) {switch (phase) {case 0:System.out.println("所有人到齐了！" + registeredParties);System.out.println();return false;case 1:System.out.println("所有人吃完了！" + registeredParties);System.out.println();return false;case 2:System.out.println("所有人离开了！" + registeredParties);System.out.println();return false;case 3:System.out.println("婚礼结束！新郎新娘抱抱！" + registeredParties);return true;default:return true;}}}static class Person implements Runnable {String name;public Person(String name) {this.name = name;}public void arrive() {milliSleep(r.nextInt(1000));System.out.printf("%s 到达现场！\n", name);phaser.arriveAndAwaitAdvance();}public void eat() {milliSleep(r.nextInt(1000));System.out.printf("%s 吃完!\n", name);phaser.arriveAndAwaitAdvance();}public void leave() {milliSleep(r.nextInt(1000));System.out.printf("%s 离开！\n", name);phaser.arriveAndAwaitAdvance();}private void hug() {if(name.equals("新郎") || name.equals("新娘")) {milliSleep(r.nextInt(1000));System.out.printf("%s 洞房！\n", name);phaser.arriveAndAwaitAdvance();} else {phaser.arriveAndDeregister(); //让其他线程不再继续执行//phaser.register()}}@Overridepublic void run() {arrive();eat();leave();hug();}}
}

ReadWriteLock 读写锁（重要）

类似数据库里的共享锁和排他锁，加了读锁的操作，允许多个读的线程同时处理，而不允许写操作进行，写操作是排他锁，读操作是共享锁。

    static ReadWriteLock readWriteLock = new ReentrantReadWriteLock();static Lock readLock = readWriteLock.readLock();static Lock writeLock = readWriteLock.writeLock();

信号量 Semaphore（车道和收费站）

限流（同时允许多少个线程运行），也可以设置为公平排队

public static void main(String[] args) {//Semaphore s = new Semaphore(2);Semaphore s = new Semaphore(2, true);//允许一个线程同时执行//Semaphore s = new Semaphore(1);new Thread(()->{try {s.acquire(); //会在这阻塞，只有拿到信号量了才能继续执行System.out.println("T1 running...");Thread.sleep(200);System.out.println("T1 running...");} catch (InterruptedException e) {e.printStackTrace();} finally {s.release();}}).start();new Thread(()->{try {s.acquire();System.out.println("T2 running...");Thread.sleep(200);System.out.println("T2 running...");s.release();} catch (InterruptedException e) {e.printStackTrace();}}).start();}

Exchanger 交换器（两个线程间交换数据，通信）

public class T12_TestExchanger {static Exchanger<String> exchanger = new Exchanger<>();public static void main(String[] args) {new Thread(()->{String s = "T1";try {s = exchanger.exchange(s);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(Thread.currentThread().getName() + " " + s);}, "t1").start();new Thread(()->{String s = "T2";try {s = exchanger.exchange(s);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(Thread.currentThread().getName() + " " + s);}, "t2").start();}
}

LockSupport （使当前线程阻塞等待）

与wait()不同的是，wait()需要一个对象锁，阻塞时需释放锁。

public static void main(String[] args) {Thread t = new Thread(()->{for (int i = 0; i < 10; i++) {System.out.println(i);if(i == 5) {LockSupport.park(); //使线程阻塞}try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}}});t.start();LockSupport.unpark(t); //可用于park之前让其不阻塞
}

六、AQS（重要）

底层就是CAS + volatile。
state是volatile修饰的，核心在于底层维护的双向链表是怎么入队与出队的（通过CAS进行的），添加tail尾部结点的时候，利用CAS比较并交换。
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-zQGvKdQt-1610448936757)(EBCEDA1A815F4EB59837DE4F689CB780)]

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-UGznGmuw-1610448936758)(1D2DE05737484357AF6430A7504BC6F9)]

调用关系图

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-fAN9jX5s-1610448936758)(F35D0DB321274155A39571AE8CB34E01)]

每个node里是一个线程，维护的是一个双向链表

VarHandle ->1、普通属性进行原子操作 2：比反射快，直接操作二进制码

jdk1.9之后的

public class T01_HelloVarHandle {int x = 8;private static VarHandle handle;static {try {handle = MethodHandles.lookup().findVarHandle(T01_HelloVarHandle.class, "x", int.class);} catch (NoSuchFieldException e) {e.printStackTrace();} catch (IllegalAccessException e) {e.printStackTrace();}}public static void main(String[] args) {T01_HelloVarHandle t = new T01_HelloVarHandle();//plain read / writeSystem.out.println((int)handle.get(t));handle.set(t,9); //原子性操作System.out.println(t.x);handlepareAndSet(t, 9, 10);  //CASSystem.out.println(t.x);handle.getAndAdd(t, 10); //增加，也是原子性操作System.out.println(t.x);System.out.println(t.x);}
}

七、ThreadLocal

顾名思义：线程变量

用途：在spring声明式事务中，保证一个线程里是同一个Connection连接对象。

set：把变量值设置到了当前线程的一个map中，Thread.currentThread.map(ThreadLocal,value)
ThreadLocal底层是维护了一个ThreadLocalMap
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-v4WfqkwU-1610448936759)(58B6FF8FD5B54F97AEED1A8A4C467D57)]
而每个线程内部会维护一个ThreadLocalMap
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-PABQTKLX-1610448936760)(C5E58F547F8C4A0D98B0E6ECDF1E7829)]

源码分析：

public void set(T value) {Thread t = Thread.currentThread();ThreadLocalMap map = getMap(t); //获取当前线程的mapif (map != null) {map.set(this, value); //this就是当前ThreadLocal对象} else {createMap(t, value);}}ThreadLocalMap getMap(Thread t) {return t.threadLocals; //返回当前线程的map}