深入理解Java多线程（三）：JUC基础篇
#Java #多线程 #JUC [字体 小·中·大]这篇文章主要侧重讲 JUC 的多数类的使用，文章里贴了很多练习的代码，可以通过代码更加深刻的了解这些类的功能。
这篇文章主要总结了 volatile、原子类、ReentrantLock、CountDownLatch、CyclicBarrier、Phaser、Semphore、Exchanger 的使用，然后进行了一些对比。
volatile
volatile 使用
1/*
2 * 演示volatile的作用
3 * 若running不使用volatile修饰则在主线程（main）修改running=false线程不会停下，而是继续运行
4 *
5 * 结论：通过volatile修饰running可以让线程使用running时获得最新值
6 *
7 * volatile的作用：
8 * 1）使共享变量在线程间可见，通过JMM内存协议，CPU缓存一致性协议使缓存失效，直接读取内存中的数据
9 * 2）阻止指令重排序
10 */
11public class T_VolatileTest {
12
13 static class Task implements Runnable{
14 /*volatile*/ boolean running = true;
15 @Override
16 public void run() {
17 System.out.println("thread start");
18 while (running){
19
20 }
21 System.out.println("thread end");
22 }
23 }
24
25 public static void main(String[] args) {
26 Task t = new Task();
27 new Thread(t).start();
28
29 try {
30 Thread.sleep(1000); //1s
31 } catch (InterruptedException e) {
32 e.printStackTrace();
33 }
34 t.running = false;
35 System.out.println("set running=false");
36 }
37}
volatile 并不是锁
1/*
2 * volatile并不能代替volatile，多并发环境下的m方法如果不同步执行，结果是错的。
3 * 在对m方法同步后，结果输出正确。
4 *
5 * 结论：
6 * 1）volatile不能够做锁
7 * 2）它仅具有使多线程间共享变量可见，但不具有同步的功能
8 */
9public class T_VolatileNotSync {
10 volatile int count = 0;
11 /*synchronized*/ void m() {
12 for(int i=0; i<1000; i++) count++;
13 }
14
15 public static void main(String[] args) {
16 T_VolatileNotSync t = new T_VolatileNotSync();
17
18 List<Thread> threads = new ArrayList<Thread>();
19
20 for(int i=0; i<10; i++) {
21 threads.add(new Thread(t::m, "thread-"+i));
22 }
23
24 // 启动线程
25 for (Thread th : threads) {
26 th.start();
27
28 }
29
30 // 阻塞到所有线程执行完毕
31 threads.forEach((o)->{
32 try {
33 o.join();
34 } catch (InterruptedException e) {
35 e.printStackTrace();
36 }
37 });
38
39 // 输出结果
40 System.out.println(t.count);
41 }
42}
volatile 只能保证引用本身可见，内部数据不能保证
1/*
2 * 在引用上t加volatile，只能保证引用（类、数组实例）本身可见，不能保证内部可见
3 */
4public class T_VolatileReference {
5 static volatile Task t = new Task();
6
7 static class Task implements Runnable{
8 boolean running = true;
9
10 @Override
11 public void run() {
12 System.out.println("run start");
13 while(running) {
14 }
15 System.out.println("run end!");
16 }
17 }
18
19 public static void main(String[] args) {
20 Thread thread = new Thread(t);
21
22 thread.start();
23
24 try {
25 Thread.sleep(1000);
26 } catch (InterruptedException e) {
27 e.printStackTrace();
28 }
29
30 t.running = false;
31 }
32}
原子操作类
原子操作类使java.util.atomic包下的类，这些类为操作提供了原子性，使用时不用进行额外的同步，这些操作仅限于对原子类对象进行操作，如果多个原子类多对象进行操作还是需要额外的同步。
AutomicInteger 使用
在 Counter 类中 AutomicInteger 并没有使用 volatile 修饰，并且 run 中的 integer.incrementAndGet()没有进行同步，最终这些操作并没有在多线程环境下出错。
1// Atomic 操作是原子操作
2public class TestAtomic {
3
4 static class Counter implements Runnable{
5 AtomicInteger integer = new AtomicInteger(0);
6
7 @Override
8 public void run() {
9 for (int i = 0; i < 10000; i++) {
10 integer.incrementAndGet();
11 }
12 }
13 }
14
15 public static void main(String[] args) {
16 List<Thread> threadList = new ArrayList<>(10);
17 Counter c = new Counter();
18
19 // 10个线程对Counter进行类型，每个线程执行run会累加10000此，最终结果为10 0000
20 for (int i = 0; i < 10; i++) {
21 threadList.add(new Thread(c));
22 }
23
24
25 for (Thread thread : threadList) {
26 thread.start();
27 }
28
29 // 阻塞到所有线程执行完毕
30 for (Thread thread : threadList) {
31 try {
32 thread.join();
33 } catch (InterruptedException e) {
34 e.printStackTrace();
35 }
36 }
37
38 // result
39 System.out.println(c.integer.get());
40 }
41}
使用原子类实现 CAS 操作。
1// 使用原子类的CAS
2// 使两个线程协作输出两个字符串 str1=123456 str2=ABCDEF, 要求结果输出为：1A2B3C4D5E6F
3public class TestAtomic2 {
4
5 public static void main(String[] args) {
6 String str1 = "123456";
7 String str2 = "ABCDEF";
8
9 // cas 锁
10 AtomicInteger mutex = new AtomicInteger(0);
11
12 // CAS操作，threa1在0输出，thread2在1时输出
13
14 // thread 1
15 new Thread(() -> {
16 for (int i = 0; i < str1.length(); i++) {
17 // 与期望的值不符进行空循环
18 while (mutex.get()!=0) {
19
20 }
21 System.out.print(str1.charAt(i));
22 mutex.set(1);
23
24 }
25
26 }).start();
27
28 // thread 2
29 new Thread(() -> {
30
31 for (int i = 0; i < str2.length(); i++) {
32 // 与期望的值不符进行空循环
33 while (mutex.get()!=1) {
34
35 }
36 System.out.print(str2.charAt(i));
37 mutex.set(0);
38 }
39 }).start();
40 }
41}
42// 1A2B3C4D5E6F
使用 volatile 变量实现 CAS，与上边的等价。
1// 使用原子类的CAS
2// 使两个线程协作输出两个字符串 str1=123456 str2=ABCDEF, 要求结果输出为：1A2B3C4D5E6F
3public class TestAtomic2 {
4
5 static volatile int mutex = 0;
6
7 public static void main(String[] args) {
8 String str1 = "123456";
9 String str2 = "ABCDEF";
10
11 // CAS操作，threa1在0输出，thread2在1时输出
12
13 // thread 1
14 new Thread(() -> {
15 for (int i = 0; i < str1.length(); i++) {
16 while (mutex != 0) {
17
18 }
19 System.out.print(str1.charAt(i));
20 mutex = 1;
21
22 }
23
24 }).start();
25
26 // thread 2
27 new Thread(() -> {
28 for (int i = 0; i < str2.length(); i++) {
29 while (mutex != 1) {
30
31 }
32 System.out.print(str2.charAt(i));
33 mutex = 0;
34 }
35 }).start();
36 }
37}
LongAdder
1import java.util.ArrayList;
2import java.util.List;
3import java.util.concurrent.atomic.LongAdder;
4
5// 使用LongAdder100个线程累加，这些操作都是原子操作，最终结果正确
6public class TestLongAdder {
7 public static void main(String[] args) {
8 LongAdder adder = new LongAdder();
9
10 List<Thread> threadList = new ArrayList<>(100);
11
12 for (int i = 0; i < 100; i++) {
13 threadList.add(new Thread(()->{
14 for (int j = 0; j < 100; j++) {
15 adder.increment();
16 }
17 }));
18 }
19
20 for (Thread thread : threadList) {
21 thread.start();
22 }
23
24 for (Thread thread : threadList) {
25 try {
26 thread.join();
27 } catch (InterruptedException e) {
28 e.printStackTrace();
29 }
30 }
31
32 System.out.println(adder.longValue());
33 }
34}
AtomicInteger 和 LongAdder 性能对比
1/*
2 * 测试三中实现自增的效率，分别时 sync+volatile、AtomicInteger、LongAdder
3 * sync AtomicInteger LongAdder
4 * 线程数
5 * 10 50 26 30
6 * 100 355 208 55
7 * 500 896 813 127
8 * 900 1286 1783 440
9 * 1200 1245 1977 1976
10 *
11 * 结论：测试结果可以看出，synchronized和AtomicInteger对比当线程数量少时AtomicInteger自旋锁比较快
12 * 当线程较多时，自旋锁性能下降而synchronized性能上升。
13 * LongAdder对CAS进行了优化，线程数量即使较多时也能保持良好的性能，但测试参数波动较大
14 */
15public class LongAdderVsAtomicInteger {
16 static int counterA = 0;
17
18 public static void main(String[] args) throws InterruptedException {
19 int num = 1200;
20 List<Thread> threadList = new ArrayList<>(num);
21
22 Object lockA = new Object();
23 for (int i = 0; i < num; i++) {
24 threadList.add(new Thread(() -> {
25 for (int i1 = 0; i1 < 10_0000; i1++) {
26 synchronized (lockA) {
27 counterA++;
28 }
29 }
30 }));
31 }
32
33 long start = System.currentTimeMillis();
34 for (Thread thread : threadList) {
35 thread.start();
36 }
37 for (Thread thread : threadList) {
38 try {
39 thread.join();
40 } catch (InterruptedException e) {
41 e.printStackTrace();
42 }
43 }
44 System.out.println("synchronized " + counterA + " " + (System.currentTimeMillis() - start));
45
46
47 Thread.sleep(1000);
48
49
50 // 使用AtomicInteger -------------------------------------------------------------
51 AtomicInteger counterB = new AtomicInteger(0);
52 threadList.clear();
53 for (int i = 0; i < num; i++) {
54 threadList.add(new Thread(() -> {
55 for (int i1 = 0; i1 < 10_0000; i1++) {
56 counterB.incrementAndGet();
57 }
58 }));
59 }
60 start = System.currentTimeMillis();
61 for (Thread thread : threadList) {
62 thread.start();
63 }
64 for (Thread thread : threadList) {
65 try {
66 thread.join();
67 } catch (InterruptedException e) {
68 e.printStackTrace();
69 }
70 }
71 System.out.println("AtomicInteger " + counterB.longValue() + " " + (System.currentTimeMillis() - start));
72
73
74 Thread.sleep(1000);
75
76
77 //测试使用LongAdder -------------------------------------------------------------
78 LongAdder counterC = new LongAdder();
79 threadList.clear();
80 for (int i = 0; i < num; i++) {
81 threadList.add(new Thread(() -> {
82 for (int i1 = 0; i1 < 10_0000; i1++) {
83 counterC.increment();
84 }
85 }));
86 }
87 start = System.currentTimeMillis();
88 for (Thread thread : threadList) {
89 thread.start();
90 }
91 for (Thread thread : threadList) {
92 try {
93 thread.join();
94 } catch (InterruptedException e) {
95 e.printStackTrace();
96 }
97 }
98 System.out.println("LongAdder " + counterC.longValue() + " " + (System.currentTimeMillis() - start));
99 }
100}
Reentrant
Reentrant 和 synchronized 功能相似，且都是可重入锁。
常用 api 有：
| 方法名 | 作用 |
|---|---|
| lock() | 加锁, 未获取到锁阻塞 |
| tryLock() | 尝试非阻塞获取锁, 并返回获取锁的情况, 无论是否获得锁都将继续执行, 当返回 false 最后又 unlock()将抛出IllegalMonitorStateException异常 |
| tryLock(times) | 超时获取锁, 超过预设的等待时间不再等待继续执行, 在未获取到锁下执行 unlock()将抛出IllegalMonitorStateException异常 |
| lockInterruptibly() | 可中断获取锁, 和 lock 方法的不同在于该方法响应中断, 当前线程获取到锁后可以被中断 |
| unlock() | 释放锁 |
| newCondition() | 创建锁条件对象, 用于执行监视器的方法阻塞当前线程或发出信号使其他线程获取锁 |
Reentrant 特性
- 是可重入锁
- 必须手动释放锁, 不能依靠异常释放锁
- 公平锁和非公平锁, 默认非公平锁
- 可以尝试锁定"try…"
Reentrant 使用
使用 Reentrant 是都要手动释放锁! 可以按照以下模板写.
1try{
2 lock.lock();
3 // 操作
4}finally{
5 lock.unlock();
6}
使用示例：
1public class ReentrantLock2 {
2 Lock lock = new ReentrantLock();
3
4 void m1() {
5 try {
6 lock.lock(); //synchronized(this)
7 for (int i = 0; i < 10; i++) {
8 TimeUnit.SECONDS.sleep(1);
9
10 System.out.println(i);
11 //if(i==2) m2();// 这里可以证明ReetrantLock是可重入锁
12 }
13 } catch (InterruptedException e) {
14 e.printStackTrace();
15 } finally {
16 lock.unlock();
17 }
18 }
19
20 void m2() {
21 try {
22 lock.lock();
23 System.out.println("m2 ...");
24 } finally {
25 lock.unlock();
26 }
27 }
28
29 public static void main(String[] args) {
30 ReentrantLock2 rl = new ReentrantLock2();
31 new Thread(rl::m1).start();
32 try {
33 TimeUnit.SECONDS.sleep(1);
34 } catch (InterruptedException e) {
35 e.printStackTrace();
36 }
37 new Thread(rl::m2).start();
38 }
39}
Reentrant 中使用 Condition
Condition 的作用是在线程持有 ReentrantLock 后对锁进一步控制，它的功能就像 Thread 的 wait/notify 功能，在线程持有锁后可以进行阻塞或通知其他线程竞争锁。
1// 演示Condition使用
2public class ReentrantLockWithCoundition {
3 ReentrantLock lock = new ReentrantLock();
4 Condition condition = lock.newCondition();
5
6 void m1() {
7 try {
8 lock.lock();
9 System.out.println("m1 start");
10 condition.await();
11 System.out.println("m1 end");
12 } catch (InterruptedException e) {
13 e.printStackTrace();
14 } finally {
15 lock.unlock();
16 }
17
18 }
19
20 void m2() {
21 try {
22 lock.lock();
23 System.out.println("m2 start");
24 condition.signal();
25 System.out.println("m2 end");
26 } finally {
27 lock.unlock();
28 }
29
30 }
31
32 public static void main(String[] args) {
33 ReentrantLockWithCoundition o = new ReentrantLockWithCoundition();
34 new Thread(o::m1,"t1").start();
35 new Thread(o::m2,"t2").start();
36 // t1线程先启动，在执行m1方法时阻塞，线程t2拿到锁执行后通知t1继续执行
37 // 因此Condition的作用是更加细粒度的控制锁，它的作用就像Thead的wait/notify功能，在一个线程获取锁后可以进一步操作
38 }
39}
Reentrant 中使用计数门闩 CountDownLatch
CountDownLatch 能实现阻塞多个线程的 join 方法的功能，在使用为 CountDownLatch 初始一个值（要阻塞的线程数），当线程执行完后值就减 1，当值为 0 时，CountDownLatch#await() 方法就不再阻塞，这跟把所有线程都执行 join 方法的功能时一样的。
1// 两个方法完成的功能时一样的，都是等到所有的线程执行完，最后执行输出语句
2public class TestCountDownLatch {
3 public static void main(String[] args) {
4 usingJoin();
5 usingCountDownLatch();
6 }
7
8 private static void usingCountDownLatch() {
9 Thread[] threads = new Thread[100];
10 CountDownLatch latch = new CountDownLatch(threads.length);
11
12 for(int i=0; i<threads.length; i++) {
13 threads[i] = new Thread(()->{
14 int result = 0;
15 for(int j=0; j<10000; j++) result += j;
16 latch.countDown();
17 });
18 }
19
20 for (int i = 0; i < threads.length; i++) {
21 threads[i].start();
22 }
23
24 try {
25 latch.await();
26 } catch (InterruptedException e) {
27 e.printStackTrace();
28 }
29 System.out.println("latch.getCount() = " + latch.getCount());
30 System.out.println("end latch");
31 }
32
33 private static void usingJoin() {
34 Thread[] threads = new Thread[100];
35
36 for(int i=0; i<threads.length; i++) {
37 threads[i] = new Thread(()->{
38 int result = 0;
39 for(int j=0; j<10000; j++) {
40 result += j;
41 //if(result%100==0) System.out.println(result);
42 }
43 });
44 }
45
46 for (int i = 0; i < threads.length; i++) {
47 threads[i].start();
48 }
49
50 for (int i = 0; i < threads.length; i++) {
51 try {
52 threads[i].join();
53 } catch (InterruptedException e) {
54 e.printStackTrace();
55 }
56 }
57 System.out.println("end join");
58 }
59}
CyclicBarrier
CyclicBarrier 设置一个数量，待阻塞线程到达这个数量时可以触发回调函数。
CyclicBarrier 只有阻塞线程计数功能没有提共同步功能，因此不能做锁。
1public class TestCyclicBarrier {
2 public static void main(String[] args) {
3 // 99个线程，每20个输出一次，共输出4次, 最后一次不足20不输出
4 CyclicBarrier barrier = new CyclicBarrier(20, () -> System.out.println("阻塞线程以达20！！请根据情况进行处理"));
5
6 for(int i=0; i<99; i++) {
7 new Thread(()->{
8 try {
9 // 记录线程阻塞
10 barrier.await();
11 //System.out.println(Thread.currentThread().getName());
12 } catch (InterruptedException e) {
13 e.printStackTrace();
14 } catch (BrokenBarrierException e) {
15 e.printStackTrace();
16 }
17 }, "t-" + i).start();
18 }
19 }
20}
Phaser
Phaser 用来控制完成阶段性的任务。
使用示例如下：
v1 版本，从示例中可以看出 Phaser 通过arriveAndAwaitAdvance来控制走下一个阶段，继承 Phaser 的类重写onAdvance来监控到达了那个阶段。
1/*
2 * 完成一个流程控制: 所有人都到达后发布一个通知,所有人都吃完宴席后发个通知,所有人都离开后发个通知
3 */
4public class TestPhaser {
5
6 static MarryPhaser phase = new MarryPhaser();
7
8 public static void main(String[] args) {
9 phase.bulkRegister(5);
10 for (int i = 0; i < 5; i++) {
11 final int temp = i;
12 new Thread(()->{
13 Person person = new Person("P" + temp);
14 person.arrive();
15 phase.arriveAndAwaitAdvance();
16
17 person.eat();
18 phase.arriveAndAwaitAdvance();
19
20 person.leave();
21 phase.arriveAndAwaitAdvance();
22 }).start();
23 }
24 }
25
26 static class MarryPhaser extends Phaser {
27 @Override
28 protected boolean onAdvance(int phase, int registeredParties) {
29 switch(phase){
30 case 0:
31 System.out.println("大家都到达了!!!");
32 return false;
33 case 1:
34 System.out.println("大家都吃完饭了");
35 return false;
36 case 2:
37 System.out.println("大家都离开了");
38 System.out.println("婚礼结束了");
39 return true;
40 default:
41 return true;
42 }
43 }
44 }
45
46 static class Person {
47 String name;
48
49 public Person(String name) {
50 this.name = name;
51 }
52
53 void arrive(){
54 System.out.printf("%s 到达了\n", name);
55 sleepUtil(1);
56 }
57
58 void eat(){
59 System.out.printf("%s 开始吃宴\n", name);
60 sleepUtil(1);
61 }
62
63 void leave(){
64 System.out.printf("%s 离开了\n", name);
65 sleepUtil(1);
66 }
67
68 }
69
70 static void sleepUtil(int second){
71 try {
72 TimeUnit.SECONDS.sleep(second);
73 } catch (InterruptedException e) {
74 e.printStackTrace();
75 }
76 }
77}
78// output
79P1 到达了
80P4 到达了
81P3 到达了
82P0 到达了
83P2 到达了
84大家都到达了!!!
85P2 开始吃宴
86P3 开始吃宴
87P0 开始吃宴
88P4 开始吃宴
89P1 开始吃宴
90大家都吃完饭了
91P0 离开了
92P3 离开了
93P4 离开了
94P2 离开了
95P1 离开了
96大家都离开了
97婚礼结束了
v2 版本。
1/*
2 * 完成一个流程控制: 所有人都到达后发布一个通知,所有人都吃完宴席后发个通知,所有人都离开后发个通知
3 */
4public class TestPhaser {
5
6 static MarryPhaser phase = new MarryPhaser();
7
8 public static void main(String[] args) {
9 phase.bulkRegister(5);
10 for (int i = 0; i < 5; i++) {
11 final int temp = i;
12 new Thread(new Person("P" + temp)).start();
13 }
14 }
15
16 static class MarryPhaser extends Phaser {
17 @Override
18 protected boolean onAdvance(int phase, int registeredParties) {
19 switch(phase){
20 case 0:
21 System.out.println("大家都到达了!!!");
22 return false;
23 case 1:
24 System.out.println("大家都吃完饭了");
25 return false;
26 case 2:
27 System.out.println("大家都离开了");
28 System.out.println("婚礼结束了");
29 return true;
30 default:
31 return true;
32 }
33 }
34 }
35
36 static class Person implements Runnable{
37 String name;
38
39 public Person(String name) {
40 this.name = name;
41
42 }
43
44 void arrive(){
45 System.out.printf("%s 到达了\n", name);
46 phase.arriveAndAwaitAdvance();
47 sleepUtil(1);
48 }
49
50 void eat(){
51 System.out.printf("%s 开始吃宴\n", name);
52 phase.arriveAndAwaitAdvance();
53 sleepUtil(1);
54 }
55
56 void leave(){
57 System.out.printf("%s 离开了\n", name);
58 phase.arriveAndAwaitAdvance();
59 sleepUtil(1);
60 }
61
62 @Override
63 public void run() {
64 arrive();
65
66 eat();
67
68 leave();
69 }
70 }
71
72 static void sleepUtil(int second){
73 try {
74 TimeUnit.SECONDS.sleep(second);
75 } catch (InterruptedException e) {
76 e.printStackTrace();
77 }
78 }
79}
Semaphore
记录型信号量，用做线程并发流量控制，运行同时运行 xx 个线程，剩下的阻塞。
使用示例：
1public class TestSemaphore {
2 public static void main(String[] args) {
3 // 允许两个线程同时执行，默认非公平锁
4 //Semaphore s = new Semaphore(2);
5
6 // 允许两个线程同时执行，非公平锁
7 Semaphore s = new Semaphore(1, true);
8
9 //只允许一个线程同时执行
10 //Semaphore s = new Semaphore(1);
11
12 new Thread(()->{
13 try {
14 s.acquire();
15
16 System.out.println("T1 running...");
17 Thread.sleep(200);
18 System.out.println("T1 running...");
19
20 } catch (InterruptedException e) {
21 e.printStackTrace();
22 } finally {
23 s.release();
24 }
25 }).start();
26
27 new Thread(()->{
28 try {
29 s.acquire();
30
31 System.out.println("T2 running...");
32 Thread.sleep(200);
33 System.out.println("T2 running...");
34
35 s.release();
36 } catch (InterruptedException e) {
37 e.printStackTrace();
38 }
39 }).start();
40 }
41}
Exchanger
用于两个线程间交换数据。它提供一个同步点，使用 exchange()方法交换数据，当第一个线程执行了 exchange（）后，Exchanger 对象会等待第二个线程执行 exchange（）方法，第二个方法执行后两者交换数据。
使用示例：
1// 交换两个线程中的数据
2public class T12_TestExchanger {
3
4 static Exchanger<String> exchanger = new Exchanger<>();
5
6 public static void main(String[] args) {
7 new Thread(()->{
8 String s = "这是t1中的数据~";
9 try {
10 s = exchanger.exchange(s);
11 } catch (InterruptedException e) {
12 e.printStackTrace();
13 }
14 System.out.println(Thread.currentThread().getName() + " " + s);
15
16 }, "t1").start();
17
18 new Thread(()->{
19 String s = "这是t2中的数据！";
20 try {
21 s = exchanger.exchange(s);
22 } catch (InterruptedException e) {
23 e.printStackTrace();
24 }
25 System.out.println(Thread.currentThread().getName() + " " + s);
26
27 }, "t2").start();
28 }
29}
30// output
31t2 这是t1中的数据~
32t1 这是t2中的数据！
(本文完)