一文带你深入理解Java多线程与高并发：Atomic类和线程同步新机制

愿天堂没有BUG

发布于 2022-10-28 15:44:29

2620

发布于 2022-10-28 15:44:29

Atomic类和线程同步新机制

今天，我们继续讲一个Atomic的问题，然后开始讲除synchronized之外的别的锁。在前面内容我们讲了synchronized、volatile、Atomic和CAS，Atomic我们只是讲了一个开头还没有讲完，今天我们继续。

像原来我们写m++你得加锁，在多线程访问的情况下，那现在我们可以用AtomicInteger了，它内部就已经帮我们实现了原子操作，直接写 count.incrementAndGet(); //count1++ 这个就相当于count++。原来我们对count是需要加锁的，现在就不需要加锁了。

我们看下面小程序，模拟，我们计一个数，所有的线程都要共同访问这个数count值，大家知道如果所有线程都要访问这个数的时候，如果每个线程给它往上加了10000，你这个时候是需要加锁的，不加锁会出问题。但是，你把它改成AtomicInteger之后就不用在做加锁的操作了，因为incrementAndGet内部用了cas操作，直接无锁的操作往上递增，有同学可能会讲为什么要用无锁操作啊，原因是无锁的操作效率会更高。

/**
* 解决同样的问题的更高效的方法，使用AtomXXX类
* AtomXXX类本身方法都是原子性的，但不能保证多个方法连续调用是原子性的
* @author mashibing
*/package com.mashibing.juc.c_018_00_AtomicXXX;import java.util.ArrayList;import java.util.List;import java.util.concurrent.atomic.AtomicInteger;public class T01_AtomicInteger {/*volatile*/ //int count1 = 0;AtomicInteger count = new AtomicInteger();/*synchronized*/ void m() {for (int i = ; i < ; i++)//if count1.get() < 1000count.incrementAndGet(); //count1++}public static void main(String[] args) {
T01_AtomicInteger t = new T01_AtomicInteger();
List<Thread> threads = new ArrayList<Thread>();for (int i = ; i < ; i++) {
threads.add(new Thread(t::m, "thread-" + i));
}
threads.forEach((o) -> o.start());
threads.forEach((o) -> {try {
o.join();
} catch (InterruptedException e) {
e.printStackTrace();
}
});
System.out.println(t.count);
}
}

大家再看这个小程序，我做了一个很粗糙的测试，这个测试基本上能说明问题，就是我们模拟了很多个线程对一个数进行递增。很多线程对一个数进行递增我们现在学了至少有两种方法：

第一种是我们一个long类型的数，递增的时候我们加锁；

第二种，我们使用AtomicLong可以让它不断的往上递增，这是第二种；

其实，还有第三种叫LongAdder;

由于呢，很多线程对一个数进行递增这个事儿，我们工作当中经常会碰上，比如说在秒杀的时候。那么，这三种哪个效率更高一些呢，很多测试来看AtomicLong还比不上第一种，但是在我的测试上来看，起码像现在这种测试的条件下AtomicLong他的效率还是比synchronized效率要高。我们来看程序，count1、conut2、count3分别是以不同的方式进行实现递增，一上来启动了1000个线程，比较多算是，因为少的话模拟不了那么高的并发。

每一个线程都new出来，之后每一个线程都做了十万次递增，第一种方式，打印起始时间->线程开始->所有线程结束->打印结束时间->计算最后花了多少时间；

第二种方式是我用synchronized，用一个lock，Object lock = new Object();，然后newRunnable(), 依然是一样的，在递增的时候我写的是synchronized (lock)，这里是替代了AtomicLong，上面是AtomicLong，下面是synchronized，在同样的就是计算时间；第三种1000个线程，每个线程十万次递增，第三种呢用的是LongAdder，这个LongAdder里面直接就是count3.increment()；

我们跑起来对比LongAdder是效率最高的，你要是自己做实验的时候，把线程数变小了LongAdder未必有优势，循环数量少了LongAdder也未必有优势，所以，实际当中用哪种你要考虑一下你的并发有多高。

package com.mashibing.juc.c_018_00_AtomicXXX;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAdder;public class T02_AtomicVsSyncVsLongAdder {static long count2 = 0L;static AtomicLong count1 = new AtomicLong(0L);static LongAdder count3 = new LongAdder();public static void main(String[] args) throws Exception {
Thread[] threads = new Thread[];for(int i=; i<threads.length; i++) {
threads[i] =new Thread(()-> {for(int k=; k<; k++) count1.incrementAndGet();
});
}long start = System.currentTimeMillis();for(Thread t : threads ) t.start();for (Thread t : threads) t.join();long end = System.currentTimeMillis();//TimeUnit.SECONDS.sleep(10);System.out.println("Atomic: " + count1.get() + " time " + (end-start));//-----------------------------------------------------------Object lock = new Object();for(int i=; i<threads.length; i++) {
threads[i] =new Thread(new Runnable() {
@Overridepublic void run() {for (int k = ; k < ; k++)
synchronized (lock) {
count2++;
}
}
});
}
start = System.currentTimeMillis();for(Thread t : threads ) t.start();for (Thread t : threads) t.join();
end = System.currentTimeMillis();
System.out.println("Sync: " + count2 + " time " + (end-start));//----------------------------------for(int i=; i<threads.length; i++) {
threads[i] =new Thread(()-> {for(int k=; k<; k++) count3.increment();
});
}
start = System.currentTimeMillis();for(Thread t : threads ) t.start();for (Thread t : threads) t.join();
end = System.currentTimeMillis();//TimeUnit.SECONDS.sleep(10);System.out.println("LongAdder: " + count1.longValue() + " time " + (end-
start));
}static void microSleep(int m) {try {
TimeUnit.MICROSECONDS.sleep(m);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

为什么Atomic要比Sync快？

因为不加锁，刚刚我们说了synchronized是要加锁的，有可能它要去操作系统申请重量级锁，所以synchronized效率偏低，在这种情形下效率偏低。

LongAdder为什么要比Atomicx效率要高呢？

是因为LongAdder的内部做了一个分段锁，类似于分段锁的概念。在它内部的时候，会把一个值放到一个数组里，比如说数组长度是4，最开始是0，1000个线程，250个线程锁在第一个数租元素里，以此类推，每一个都往上递增算出来结果在加到一起。

间歇性复习

我们讲了synchronized的一些细节；

volatile；

atomicXXX他的原理就是CAS;

在这个基础之上有给大家讲了递增increment，他有好几种方式sync、atomicXXX、LongAdder;

并不是很多，我们继续...

老类型的锁就是synchronized我们已经讲完了，下面我们来讲基于CAS的一些新类型的锁，讲这些锁的用法，再来讲这些锁的原理。

ReentrantLock

第一种锁比较新鲜可重入锁ReentranLlock，synchronized本身就是可重入锁的一种，什么叫可重入，意思就是我锁了一下还可以对同样这把锁再锁一下，synchronized必须是可重入的，不然的话子类调用父类是没法实现的，看下面这个小程序是这样写的，m1方法里面做了一个循环每次睡1秒钟，每隔一秒种打印一个。接下来调m2，是一个synchronized方法也是需要加锁的，我们来看主程序启动线程m1，一秒钟后再启动线程m2。分析下这个执行过程在第一个线程执行到一秒钟的时候第二个线程就会起来，假如我们这个锁是不可重入的会是什么情况，第一个线程申请这把锁，锁的这个对象，然后这里如果是第二个线程来进行申请的话，他start不了，必须要等到第一个线程结束了，因为这两个是不同的线程。两个线程之间肯定会有争用，可以在m1里面调用m2就可以，synchronized方法是可以调用synchronized方法的。锁是可重入的。

/**
* reentrantlock用于替代synchronized
* 本例中由于m1锁定this，只有m1
ִ
执
行完毕的时候,m2才能执行
* 这里是复习synchronized最原始的语义
* @author mashibing
*/package com.mashibing.juc.c_020;

import java.util.concurrent.TimeUnit;子类和父类如果是synchronized(this)就是同一把锁，同一个this当然是同一把锁。

ReentrantLock是可以替代synchronized的，怎么替代呢，看如下代码，原来写synchronized的地方换写lock.lock()，加完锁之后需要注意的是记得lock.unlock()解锁，由于synchronized是自动解锁的，大括号执行完就结束了。lock就不行，lock必须得手动解锁，手动解锁一定要写在try...fifinally里面保证最好一定要解锁，不然的话上锁之后中间执行的过程有问题了，死在那了，别人就永远也拿不到这把锁了。

public class T01_ReentrantLock1 {synchronized void m1() {for(int i=; i<; i++) {try {
TimeUnit.SECONDS.sleep();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(i);if(i == ) m2();
}
}synchronized void m2() {
System.out.println("m2 ...");
}public static void main(String[] args) {
T01_ReentrantLock1 rl = new T01_ReentrantLock1();new Thread(rl::m1).start();try {
TimeUnit.SECONDS.sleep();
} catch (InterruptedException e) {
e.printStackTrace();
}//new Thread(rl::m2).start();}
}

/**
* reentrantlock用于替代synchronized
* 由于m1锁定this，只有m1
ִ
执
行完毕的时候,m2才能执行
* 这个里是复习synchronized最原始的语义
*
* 使用reentrantlock可以完成同样的功能
* 需要注意的是，必须要必须要必须要手动释放锁（重要的事情说三遍）
* 使用syn锁定的话如果遇到异常，jvm会手动释放锁，但是lock必须手动释放锁
* @author mashibing
*/package com.mashibing.juc.c_020;import java.util.concurrent.TimeUnit;import java.util.concurrent.locks.Lock;import java.util.concurrent.locks.ReentrantLock;public class T02_ReentrantLock2 {
Lock lock = new ReentrantLock();void m1() {try {
lock.lock(); //synchronized(this)for (int i = ; i < ; i++) {
TimeUnit.SECONDS.sleep();
System.out.println(i);
}
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}void m2() {try {
lock.lock();
System.out.println("m2 ...");
} finally {
lock.unlock();
}
}public static void main(String[] args) {
T02_ReentrantLock2 rl = new T02_ReentrantLock2();new Thread(rl::m1).start();try {
TimeUnit.SECONDS.sleep();
} catch (InterruptedException e) {
e.printStackTrace();
}new Thread(rl::m2).start();
}
}

可能有同学会说reentrantlock既然和synchronized差不多的话，那我们要它有什么用呢，ReentrantLock有一些功能还是要比synchronized强大的，强大的地方，你可以使用tryLock进行尝试锁定，不管锁定与否，方法都将继续执行，synchronized如果搞不定的话他肯定就阻塞了，但是用ReentrantLock你自己就可以决定你到底要不要wait。

下面程序就是说比如5秒钟你把程序执行完就可能得到这把锁，如果得不到就不行。由于我的第一个线程跑了10秒钟，所以你在第二个线程里申请5秒肯定是那不到的，把循环次数减少就可以能拿到了。

package com.mashibing.juc.c_020;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;public class T03_ReentrantLock3 {
Lock lock = new ReentrantLock();void m1() {try {lock.lock();for (int i = ; i < ; i++) {
TimeUnit.SECONDS.sleep();
System.out.println(i);
}
} catch (InterruptedException e) {
e.printStackTrace();
} finally {lock.unlock();
}
}

/**
* 使用tryLock进行尝试锁定，不管锁定与否，方法都将继续执行
* 可以根据tryLock的返回值来判断是否锁定
* 也可以指定tryLock的时间
*/void m2() {/*
boolean locked = lock.tryLock();
System.out.println("m2 ..." + locked);
if(locked) lock.unlock();
*/boolean locked = false;try {
locked = lock.tryLock(, TimeUnit.SECONDS);
System.out.println("m2 ..." + locked);
} catch (InterruptedException e) {
e.printStackTrace();
} finally {if(locked) lock.unlock();
}
}public static void main(String[] args) {
T03_ReentrantLock3 rl = new T03_ReentrantLock3();new Thread(rl::m1).start();try {
TimeUnit.SECONDS.sleep();
} catch (InterruptedException e) {
e.printStackTrace();
}new Thread(rl::m2).start();
}
}

当然除了这个之外呢，ReentrantLock还可以用lock.lockInterruptibly() 这个类，对interrupt()方法做出相应，可以被打断的加锁，如果以这种方式加锁的话我们可以调用一个 t2.interrupt(); 打断线程2的等待。线程1 上来之后加锁，加锁之后开始睡，睡的没完没了的，被线程1拿到这把锁的话，线程2如果说在想拿到这把锁不太可能，拿不到锁他就会在哪儿等着，如果我们使用原来的这种lock.lock()是打断不了它的，那么我们就可以用另外一种方式lock.lockInterruptibly() 这个类可以被打断的，当你要想停止线程2就可以用 interrupt() ，这也是ReentrantLock比synchronized好用的一个地方。

package com.mashibing.juc.c_020;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Function;public class T04_ReentrantLock4 {public static void main(String[] args) {
Lock lock = new ReentrantLock();
Thread t1 = new Thread(()->{try {lock.lock();
System.out.println("t1 start");
TimeUnit.SECONDS.sleep(Integer.MAX_VALUE);
System.out.println("t1 end");
} catch (InterruptedException e) {
System.out.println("interrupted!");
} finally {lock.unlock();
}
});
t1.start();
Thread t2 = new Thread(()->{try {//lock.lock();lock.lockInterruptibly(); //可以对interrupt()方法做出相应System.out.println("t2 start");
TimeUnit.SECONDS.sleep();
System.out.println("t2 end");
} catch (InterruptedException e) {
System.out.println("interrupted!");
} finally {lock.unlock();
}
});
t2.start();try {
TimeUnit.SECONDS.sleep();
} catch (InterruptedException e) {
e.printStackTrace();
}
t2.interrupt();//打断线程2的等待}
}

ReentrantLock还可以指定为公平锁，公平锁的意思是当我们new一个ReentrantLock你可以传一个参数为true，这个true表示公平锁，公平锁的意思是谁等在前面就先让谁执行，而不是说谁后来了之后就马上让谁执行。如果说这个锁不公平，来了一个线程上来就抢，它是有可能抢到的，如果说这个锁是个公平锁，这个线程上来会先检查队列里有没有原来等着的，如果有的话他就先进队列里等着别人先运行，这是公平锁的概念。

ReentrantLock默认是非公平锁。

/**
* ReentrantLock还可以指定为公平锁
* @author mashibing
*/package com.mashibing.juc.c_020;
import java.util.concurrent.locks.ReentrantLock;public class T05_ReentrantLock5 extends Thread {private static ReentrantLock lock=new ReentrantLock(true); //参数为true表示为公平锁，请对比输出结果public void run() {for(int i=; i<; i++) {lock.lock();try{
System.out.println(Thread.currentThread().getName()+"获得锁");
}finally{lock.unlock();
}
}
}public static void main(String[] args) {
T05_ReentrantLock5 rl=new T05_ReentrantLock5();
Thread th1=new Thread(rl);
Thread th2=new Thread(rl);
th1.start();
th2.start();
}
}

我们稍微回顾一下: Reentrantlock vs synchronized

ReentrantLock可以替代synchronized这是没问题的，他也可以重入，可以锁定的。本身的底层是castrylock：自己来控制，我锁不住该怎么办lockInterruptibly：这个类，中间你还可以被打断它还可以公平和非公平的切换现在除了synchronized之外，多数内部都是用的cas。当我们聊这个AQS的时候实际上它内部用的是park和unpark，也不是全都用的cas,他还是做了一个锁升级的概念，只不过这个锁升级做的比较隐秘，在你等待这个队列的时候如果你拿不到还是进入一个阻塞的状态，前面至少有一个cas的状态，他不像原先就直接进入阻塞状态了。

CountDownLatchCountDown叫倒数，Latch是门栓的意思（倒数的一个门栓，5、4、3、2、1数到了，我这个门栓就开了）

看下面的小程序，这小程序叫usingCountDownLatch，new了100个线程，接下来，又来了100个数量的CountDownLatch，什么意思，就是，这是一个门栓，门栓上记了个数threads.length是100，每一个线程结束的时候我让 latch.countDown()，然后所有线程start()，再latch.await()，最后结束。那CountDown是干嘛使得呢，看latch.await()，它的意思是说给我看住门，给我插住不要动。每个线程执行到latch.await()的时候这个门栓就在这里等着，并且记了个数是100，每一个线程结束的时候都会往下CountDown，CountDown是在原来的基础上减1，一直到这个数字变成0的时候门栓就会被打开，这就是它的概念，它是用来等着线程结束的。

用join实际上不太好控制，必须要你线程结束了才能控制，但是如果是一个门栓的话我在线程里不停的CountDown，在一个线程里就可以控制这个门栓什么时候往前走，用join我只能是当前线程结束了你才能自动往前走，当然用join可以，但是CountDown比它要灵活。

package com.mashibing.juc.c_020;
import java.util.concurrent.CountDownLatch;public class T06_TestCountDownLatch {public static void main(String[] args) {
usingJoin();
usingCountDownLatch();
}private static void usingCountDownLatch() {
Thread[] threads = new Thread[];
CountDownLatch latch = new CountDownLatch(threads.length);for(int i=; i<threads.length; i++) {
threads[i] = new Thread(()->{int result = ;for(int j=; j<; j++) result += j;
latch.countDown();
});
}for (int i = ; i < threads.length; i++) {
threads[i].start();
}try {
latch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("end latch");
}private static void usingJoin() {
Thread[] threads = new Thread[];for(int i=; i<threads.length; i++) {
threads[i] = new Thread(()->{int result = ;for(int j=; j<; j++) result += j;});
}for (int i = ; i < threads.length; i++) {
threads[i].start();
}for (int i = ; i < threads.length; i++) {try {
threads[i].join();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
System.out.println("end join");
}
}

CyclicBarrier

来讲这个同步工具叫CyclicBarrier，意思是循环栅栏，这有一个栅栏，什么时候人满了就把栅栏推倒，哗啦哗啦的都放出去，出去之后扎栅栏又重新起来，再来人，满了，推倒之后又继续。

下面程序，两个参数，第二个参数不传也是可以的，就是满了之后不做任何事情。第一个参数是20，满了之后帮我调用第二个参数指定的动作，我们这个指定的动作就是一个Runnable对象，打印满人，发车。什么barrier.await()会被放倒，就是等够20个人了，后面也可以写你要做的操作 s。什么时候满了20人了就发车。下面第一种写法是满了之后我什么也不做，第二种写法是用Labda表达式的写法。这个意思就是线程堆满了，我们才能往下继续执行。

举例：CyclicBarrier的概念呢比如说一个复杂的操作，需要访问数据库，需要访问网络，需要访问文件，有一种方式是顺序执行，挨个的都执行完，效率非常低，这是一种方式，还有一种可能性就是并发执行，原来是1、2、3顺序执行，并发执行是不同的线程去执行不同的操作，有的线程去数据库找，有的线程去网络访问，有的线程去读文件，必须是这三个线程全部到位了我才能去进行，这个时候我们就可以用CyclicBarrier。

package com.mashibing.juc.c_020;import java.util.concurrent.BrokenBarrierException;import java.util.concurrent.CyclicBarrier;public class T07_TestCyclicBarrier {public static void main(String[] args) {//CyclicBarrier barrier = new CyclicBarrier(20);CyclicBarrier barrier = new CyclicBarrier(, () ->
System.out.println("满人"));/*CyclicBarrier barrier = new CyclicBarrier(20, new Runnable() {
@Override
public void run() {
System.out.println("满人，发车");
}
});*/for(int i=; i<; i++) {new Thread(()->{try {
barrier.await();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
}).start();
}
}
}

Phaser

Phaser它就更像是结合了CountDownLatch和CyclicBarrier，翻译一下叫阶段。这个稍微复杂一些，如果有面试官问到你能说出来，一定会高看你一眼的。

Phaser是按照不同的阶段来对线程进行执行，就是它本身是维护着一个阶段这样的一个成员变量，当前我是执行到那个阶段，是第0个，还是第1个阶段啊等等，每个阶段不同的时候这个线程都可以往前走，有的线程走到某个阶段就停了，有的线程一直会走到结束。你的程序中如果说用到分好几个阶段执行，而且有的人必须得几个人共同参与的一种情形的情况下可能会用到这个Phaser。

有种情形很可能用到，如果你写的是遗传算法，遗传算法是计算机来模拟达尔文的进化策略所发明的一种算法，当你去解决这个问题的时候这个Phaser是有可能用的上的。这个东西更像是CyclicBarrier，栅栏这个东西是一个一个的栅栏，他原来是一个循环的栅栏，循环使用，但是这个栅栏是一个栅栏一个栅栏的。

好，来看我们自己模拟的一个小例子。模拟了一个结婚的场景，结婚是有好多人要参加的，因此，我们写了一个类Person是一个Runnable可以new出来，扔给Thread去执行，模拟我们每个人要做一些操作，有这么几种方法，arrive()到达、eat()吃、leave()离开、hug()拥抱这么几个。作为一个婚礼来说它会分成好几个阶段，第一阶段大家好都得到齐了，第二个阶段大家开始吃饭，三阶段大家离开，第四个阶段新郎新娘入洞房，那好，每个人都有这几个方法，在方法的实现里头我就简单的睡了1000个毫秒，我自己写了一个方法，把异常处理写到了方法里了。

在看主程序，一共有五个人参加婚礼了，接下来新郎，新娘参加婚礼，一共七个人。它一start就好调用我们的run（）方法，它会挨着牌的调用每一个阶段的方法。那好，我们在每一个阶段是不是得控制人数，第一个阶段得要人到期了才能开始，二阶段所有人都吃饭，三阶段所有人都离开，但是，到了第四阶段进入洞房的时候就不能所有人都干这个事儿了。所以，要模拟一个程序就要把整个过程分好几个阶段，而且每个阶段必须要等这些线程给我干完事儿了你才能进入下一个阶段。

那怎么来模拟过程呢，我定义了一个phaser，我这个phaser是从Phaser这个类继承，重写onAdvance方法，前进，线程抵达这个栅栏的时候，所有的线程都满足了这个第一个栅栏的条件了onAdvance会被自动调用，目前我们有好几个阶段，这个阶段是被写死的，必须是数字0开始，onAdvance会传来两个参数phase是第几个阶段，registeredParties是目前这个阶段有几个人参加，每一个阶段都有一个打印，返回值false，一直到最后一个阶段返回true，所有线程结束，整个栅栏组，Phaser栅栏组就结束了。

我怎么才能让我的线程在一个栅栏面前给停住呢，就是调用 phaser.arriveAndAwaitAdvance()这个方法，这个方法的意思是到达等待继续往前走，直到新郎新娘如洞房，其他人不在参与，调用phaser.arriveAndDeregister() 这个方法。还有可以调用方法phaser.register()往上加，不仅可以控制栅栏上的个数还可以控制栅栏上的等待数量，这个就叫做phaser。是给大家拓宽知识面用的。

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();for(int i=; i<; 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 :
System.out.println("所有人都到齐了！" + registeredParties);
System.out.println();return false;case :
System.out.println("所有人都吃完了！" + registeredParties);
System.out.println();return false;case :
System.out.println("所有人都离开了！" + registeredParties);
System.out.println();return false;case :
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());
System.out.printf("%s 到达现场！\n", name);
phaser.arriveAndAwaitAdvance();
}public void eat() {
milliSleep(r.nextInt());
System.out.printf("%s 吃完!\n", name);
phaser.arriveAndAwaitAdvance();
}public void leave() {
milliSleep(r.nextInt());
System.out.printf("%s 离开！\n", name);
phaser.arriveAndAwaitAdvance();
}private void hug() {if(name.equals("新郎") || name.equals("新娘")) {
milliSleep(r.nextInt());
System.out.printf("%s 洞房！\n", name);
phaser.arriveAndAwaitAdvance();
} else {
phaser.arriveAndDeregister();//phaser.register()}
}
@Overridepublic void run() {
arrive();
eat();
leave();
hug();
}
}
}

ReadWriteLock

这个ReadWriteLock 是读写锁。读写锁的概念其实就是共享锁和排他锁，读锁就是共享锁，写锁就是排他锁。那这个是什么意思，我们先要来理解这件事儿，读写有很多种情况，比如说你数据库里的某条儿数据你放在内存里读的时候特别多，而改的时候并不多。举一个简单的例子，我们公司的组织结构，我们要想显示这组织结构下有哪些人在网页上访问，所以这个组织结构被访问到会读，但是很少更改，读的时候多写的时候就并不多，这个时候好多线程来共同访问这个结构的话，有的是读线程有的是写线程，要求他不产生这种数据不一致的情况下我们采用最简单的方式就是加锁，我读的时候只能自己读，写的时候只能自己写，但是这种情况下效率会非常的低，尤其是读线程非常多的时候，那我们就可以做成这种锁，当读线程上来的时候加一把锁是允许其他读线程可以读，写线程来了我不给它，你先别写，等我读完你在写。读线程进来的时候我们大家一块读，因为你不改原来的内容，写线程上来把整个线程全锁定，你先不要读，等我写完你在读。

我们看这个读写锁怎么用，我们这有两个方法，read()读一个数据，write()写一个数据。read这个数据的时候我需要你往里头传一把锁，这个传那把锁你自己定，我们可以传自己定义的全都是排他锁，也可以传读写锁里面的读锁或写锁。write的时候也需要往里面传把锁，同时需要你传一个新值，在这里值里面传一个内容。我们模拟这个操作，读的是一个int类型的值，读的时候先上锁，设置一秒钟，完了之后read over，最后解锁unlock。再下面写锁，锁定后睡1000毫秒，然后把新值给value，write over后解锁，非常简单。

我们现在的问题是往里传这个lock有两种传法，第一种直接new ReentrantLock()传进去，分析下这种方法，主程序定义了一个Runnable对象，第一个是调用read() 方法，第二个是调用write() 方法同时往里头扔一个随机的值，然后起了18个读线程，起了两个写线程，这个两个我要想执行完的话，我现在传的是一个ReentrantLock，这把锁上了之后没有其他任何人可以拿到这把锁，而这里面每一个线程执行都需要1秒钟，在这种情况下你必须得等20秒才能干完这事儿；

第二种，我们换了锁 new ReentrantReadWriteLock() 是ReadWriteLock的一种实现，在这种实现里头我又分出两把锁来，一把叫readLock，一把叫writeLock，通过他的方法readWriteLock.readLock()来拿到readLock对象，读锁我就拿到了。通过readWriteLock.writeLock()拿到writeLock对象。这两把锁在我读的时候扔进去，因此，读线程是可以一起读的，也就是说这18个线程可以一秒钟完成工作结束。

所以使用读写锁效率会大大的提升。

package com.mashibing.juc.c_020;
import java.util.Random;
import java.util.concurrent.atomic.LongAdder;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;public class T10_TestReadWriteLock {static Lock lock = new ReentrantLock();private static int value;static ReadWriteLock readWriteLock = new ReentrantReadWriteLock();static Lock readLock = readWriteLock.readLock();static Lock writeLock = readWriteLock.writeLock();public static void read(Lock lock) {try {lock.lock();
Thread.sleep();
System.out.println("read over!");//模拟读取操作} catch (InterruptedException e) {
e.printStackTrace();
} finally {lock.unlock();
}
}public static void write(Lock lock, int v) {try {lock.lock();
Thread.sleep();value = v;
System.out.println("write over!");//模拟写操作} catch (InterruptedException e) {
e.printStackTrace();
} finally {lock.unlock();
}
}public static void main(String[] args) {//Runnable readR = ()-> read(lock);Runnable readR = ()-> read(readLock);//Runnable writeR = ()->write(lock, new Random().nextInt());Runnable writeR = ()->write(writeLock, new Random().nextInt());for(int i=; i<; i++) new Thread(readR).start();for(int i=; i<; i++) new Thread(writeR).start();
}
}

以后还写不写synchronized？分布式锁咋实现？

以后一般不用这些新的锁，多数都用synchronized。只有特别特别追求效率的时候才会用到这些新的锁。现在用布式锁很多，分布式锁也是面试必问的，它也不难比较简单，学了redis，redis有两种方法可以实现分布式锁，还有学完 ZooKeeper也可以实现分布式锁，还有数据库也可以实现，但数据库实现的效率就比较低了。

给大家举一个简单的例子，就说秒杀这个事情。在开始秒杀之前它会从数据库里面读某一个数据，比如所一个电视500台，只能最对售卖500台，完成这件事得是前面的线程访问同一个数最开始是0一直涨到500就结束，需要加锁，从0递增，如果是单机LongAdder或AtomicInteger搞定。分布式的就必须得用分布式锁，对一个数进行上锁。redis是单线程的所以扔在一台机器上就ok了。

Semaphore

我们来聊这个Semaphore，信号灯。可以往里面传一个数，permits是允许的数量，你可以想着有几盏信号灯，一个灯里面闪着数字表示到底允许几个来参考我这个信号灯。

s.acquire()这个方法叫阻塞方法，阻塞方法的意思是说我大概acquire不到的话我就停在这，acquire的意思就是得到。如果我 Semaphore s = new Semaphore(1) 写的是1，我取一下，acquire一下他就变成0，当变成0之后别人是acquire不到的，然后继续执行，线程结束之后注意要s.release()，执行完该执行的就把他release掉，release又把0变回去1，还原化。

Semaphore的含义就是限流，比如说你在买票，Semaphore写5就是只能有5个人可以同时买票。

acquire的意思叫获得这把锁，线程如果想继续往下执行，必须得从Semaphore里面获得一个许可，他一共有5个许可用到0了你就得给我等着。

例如，有一个八条车道的机动车道，这里只有两个收费站，到这儿，谁acquire得到其中某一个谁执行。默认Semaphore是非公平的，new Semaphore(2, true)第二个值传true才是设置公平。公平这个事儿是有一堆队列在哪儿等，大家伙过来排队。用这个车道和收费站来举例子，就是我们有四辆车都在等着进一个车道，当后面在来一辆新的时候，它不会超到前面去，要在后面排着这叫公平。所以说内部是有队列的，不仅内部是有队列的，这里面用到的东西，我今天将的所有的从头到尾reentrantlock、

CountDownLatch、CyclicBarrier、Phaser、ReadWriteLock、Semaphore还有后面要讲的Exchanger都是用同一个队列，同一个类来实现的，这个类叫AQS。

package com.mashibing.juc.c_020;
import java.util.concurrent.Semaphore;public class T11_TestSemaphore {public static void main(String[] args) {//Semaphore s = new Semaphore(2);Semaphore s = new Semaphore(, true);//允许一个线程同时执行//Semaphore s = new Semaphore(1);new Thread(()->{try {
s.acquire();
System.out.println("T1 running...");
Thread.sleep();
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();
System.out.println("T2 running...");
s.release();
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
}
}

Exchanger

这个Exchanger是给大家扩宽知识面用的，看下面这个小程序，这里我们定义了一个Exchanger，

Exchanger叫做交换器，俩人之间互相交换个数据用的。怎么交换呢，看这里，我第一个线程有一个成员变量叫s，然后exchanger.exchange(s)，第二个也是这样，t1线程名字叫T1，第二个线程名字叫T2。

到最后，打印出来你会发现他们俩交换了一下。线程间通信的方式非常多，这只是其中一种，就是线程之间交换数据用的。exchanger你可以把它想象成一个容器，这个容器有两个值，两个线程，有两个格的位置，第一个线程执行到exchanger.exchange的时候，阻塞，但是要注意我这个exchange方法的时候是往里面扔了一个值，你可以认为吧T1扔到第一个格子了，然后第二个线程开始执行，也执行到这句话了，exchange，

他把自己的这个值T2扔到第二个格子里。接下来这两个哥们儿交换一下，T1扔给T2，T2扔给T1，两个线程继续往前跑。exchange只能是两个线程之间，交换这个东西只能两两进行。

exchange的使用场景，比如在游戏中两个人装备交换。

package com.mashibing.juc.c_020;import java.util.concurrent.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();
}
}

内容回顾

我们首先讲了ReentrantLock和synchronized一个区别，ReentrantLock更灵活，更方便；讲了CountDownLatch的用法，就是倒计时，什么时候计数完了，门栓打开，程序继续往下执行；
CycliBarrier一个栅栏，循环使用，什么时候人满了，栅栏放倒大家冲过去；
Phaser分阶段的栅栏；
ReadWriteLock读写锁，重点掌握；
Semaphore限流用的；
Exchanger两个线程之间互相交换数据；