javaEE 初阶 — JUC(java.util.concurrent) 的常见类_java concurrent 常用类

1. Callable 接口

类似于 Runnable 一样。
Runnable 用来描述一个任务，描述的任务没有返回值。
Callable 也是用来描述一个任务，描述的任务是有返回值的。

如果需要使用一个线程单独的计算出某个结果，此时使用 Callable 是比较合适的。

1.1 Callable 的用法

代码示例：创建线程计算 1 + 2 + 3 + … + 1000，不使用 Callable 版本。

• 创建一个类 Result，包含一个 sum 表示最终结果，lock 表示线程同步使用的锁对象。
• main 方法中先创建 Result 实例，然后创建一个线程 t。在线程内部计算 1 + 2 + 3 + … + 1000。
• 主线程同时使用 wait 等待线程 t 计算结束。(注意，如果执行到 wait 之前，线程 t 已经计算完了，就不
  必等待了)。
• 当线程 t 计算完毕后，通过 notify 唤醒主线程，主线程再打印结果。

static class Result {
    public int sum = 0;
    public Object lock = new Object();
}

public static void main(String[] args) throws InterruptedException {
    Result result = new Result();
    
    Thread t = new Thread() {
        @Override
        public void run() {
            int sum = 0;
            for (int i = 1; i <= 1000; i++) {
                sum += i;
            }
            synchronized (result.lock) {
                result.sum = sum;
                result.lock.notify();
            }
        }
    };
    t.start();

    synchronized (result.lock) {
        while (result.sum == 0) {
            result.lock.wait();
        }
        System.out.println(result.sum);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

可以看到，上述代码需要一个辅助类 Result，还需要使用一系列的加锁和 wait notify 操作，代码复
杂，容易出错。


代码示例：创建线程计算 1 + 2 + 3 + … + 1000，使用 Callable 版本

package thread;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;

public class ThreadDemo9 {

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        Callable<Integer> callable = new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                int sum = 0;
                for (int i = 0; i < 1000; i++) {
                    sum += i;
                }
                return sum;
            }
        };

        FutureTask<Integer> futureTask = new FutureTask<>(callable);
        Thread thread = new Thread(futureTask);
        thread.start();

        Integer result = futureTask.get();
        System.out.println(result);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

call() 相当于是 Runnable 的 run() 方法，run方法返回的是 void 此处返回值泛型参数。

FutureTask 表示这是未来的任务。
可以把 FutureTask 简单理解为点餐时的小票，这个小票就是 FutureTask。
后面我们可以随时凭这张小票去查看自己点的餐做出来了没。

get() 方法就是获取结果。
get 会发生阻塞，直到 callable 执行完毕，get 才阻塞完成，才获取到结果。

可以看到，使用 Callable 和 FutureTask 之后，代码简化了很多，也不必手动写线程同步代码了。

2. ReentrantLock

这里的 ReentrantLock 是标准库给我们提供的另一种锁，也是可重入的。

synchronized 是直接基于代码块的方式来加锁和解锁的。
ReentrantLock 使用了 lock 方法和 unlock 方法来加锁和解锁。

package thread;

import java.util.concurrent.locks.ReentrantLock;

public class ThreadDemo10 {
    
    public static void main(String[] args) {
        ReentrantLock reentrantLock = new ReentrantLock();
        reentrantLock.lock();
        
        reentrantLock.unlock();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13

reentrantLock.lock() 和 reentrantLock.unlock() 之间的代码就被锁给保护起来了。
但是这样的写法有很大的弊端：

unlock 有可能会执行不到

2.1 ReentrantLock 的缺陷

如果代码中间存在 return 或者异常，就有可能会导致 unlock 不能顺利执行。

package thread;

import java.util.concurrent.locks.ReentrantLock;

public class ThreadDemo10 {

    public static void main(String[] args) {
        ReentrantLock reentrantLock = new ReentrantLock();
        int num = 0;
        reentrantLock.lock();

        if (num == 0) {
            return;
        }

        if (num == 1) {
            return;
        }

        throw new Exception();

        reentrantLock.unlock();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

这时就要把 unlock 写在 finally 里

package thread;

import java.util.concurrent.locks.ReentrantLock;

public class ThreadDemo10 {

    public static void main(String[] args) {
        ReentrantLock reentrantLock = new ReentrantLock();
        int num = 0;
        try {
            if (num == 0) {
                return;
            }

            if (num == 1) {
                return;
            }

            throw new Exception();
        } finally {
            reentrantLock.unlock();
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

2.1 ReentrantLock 的优势

1、ReentrantLock 提供了公平锁版本的实现。

  ReentrantLock reentrantLock = new ReentrantLock(true);
1

 ReentrantLock reentrantLock = new ReentrantLock();
1

ReentrantLock 括号里加上 true 表示这是一个公平锁。
什么都不加，或者加 false 表示这是一个非公平的锁。


2、更加灵活的阻塞等待方式

对于 synchronized 来说，提供的加锁操作就是 “死等” ，只要获取不到锁，就会一直等待。
而 ReentrantLock 提供更加灵活的等待方式：trylock

这里的 trylock 分为有参数和无参数的版本。

无参数版本：能加锁就加，加不上就放弃。

有参数的版本：指定了超时时间，加不上锁就等待一段时间。如果时间到了也没加上就放弃。


3、ReentrantLock 提供了一个更加强大的等待机制。

synchronized 搭配的是 wait 和 notify ，notify 的时候随机唤醒一个 wait 的线程。

ReentrantLock 搭配的是一个 Condition 类，进行唤醒的时候可以指定唤醒的线程。

3. 原子类

原子类内部用的是 CAS 实现的，所以性能要比加锁实现 i++ 高很多。

原子类有以下几个：

• AtomicBoolean
• AtomicInteger
• AtomicIntegerArray
• AtomicLong
• AtomicReference
• AtomicStampedReferenc

以 AtomicInteger 举例，常见方法有：

• addAndGet(int delta); i += delta;
• decrementAndGet(); --i;
• getAndDecrement(); i–;
• incrementAndGet(); ++i;
• getAndIncrement(); i++

基于 CAS 确实是更高效的解决了线程安全问题，但是 CAS 不能代替锁。
CAS 的适用范围是有限的，不像锁的适用范围那么广。

4. 信号量 Semaphore

操作系统上提到的信号量和此处这个信号量是一个东西，只不过此处的这个信号量是 java 把操作系统原生的信号量封装了一下。

信号量在生活中经常可以见到。

比如说停车场，因为停车场的空闲位置个数都是固定的，当位置到达上限的时候就不能停车了。

停车场的入口位置会有一个牌子，牌子上显示空闲位置的个数。
每当有车进去，牌子上的显示的个数就减少一个；有车出来，个数就加一个。
这个牌子就相当于是一个计数器，当这个计数器为 0 的时候，也就是停车场空闲位置达到上限了。

当没有位置的时候要停车，就只能在这里等待或者去别处找停车场。

信号量本质上就是一个 计数器，描述了 “可用资源的个数”

P操作（acquire） ：申请一个可用资源，计时器就要 -1。
V操作（） ：释放一个可用资源，计数器个数就要 +1。

如果此时的计数器为 0 了，继续执行 P 操作，就会发生阻塞等待。

考虑一个计数初始值为 1 的信号量。
针对这个信号量的值，就只有1 和 0 两种取值（信号量不能是负的）

执行一次 P 操作，1 就变成了 0 。
执行一次 V 操作，0 就变成了 1 。

如果已经执行过一次 P 操作了，继续执行 P 操作，就会阻塞等待。

有没有让你想到锁，（锁可以视为是计数器为 1 的信号量，二元信号量）
锁是一种信号量的特殊情况，信号量是锁的一般表达。

Semaphore 的使用

package thread;

import java.util.concurrent.Semaphore;

public class ThreadDemo11 {

    public static void main(String[] args) throws InterruptedException{
        Semaphore semaphore = new Semaphore(3); //指定计数器个数是3
        semaphore.acquire(); //执行P操作
        System.out.println("P操作一次");

        semaphore.acquire(); //执行P操作
        System.out.println("P操作一次");

        semaphore.acquire(); //执行P操作
        System.out.println("P操作一次");

        semaphore.acquire(); //执行P操作
        System.out.println("P操作一次");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

当前的计数器个数是 3 ，当计数器为 0 的时候继续执行 acquire 操作就会阻塞等待。

package thread;

import java.util.concurrent.Semaphore;

public class ThreadDemo11 {

    public static void main(String[] args) throws InterruptedException{
        Semaphore semaphore = new Semaphore(3); //指定计数器个数是3
        semaphore.acquire(); //执行P操作
        System.out.println("P操作一次");

        semaphore.acquire(); //执行P操作
        System.out.println("P操作一次");

        semaphore.acquire(); //执行P操作
        System.out.println("P操作一次");

        semaphore.release(); //执行V操作

        semaphore.acquire(); //执行P操作
        System.out.println("P操作一次");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

当计数器为 0 的时候，执行 V 操作后，计数器的个数就加了一个。

5. CountDownLatch

假如有一个跑步比赛。



这场跑步比赛，开始的时间确定的（发号枪）
但是结束的时间是不明确的。（所有的选手都冲过终点后）

为了等待跑步比赛的结束，就引入了这个 CountDownLatch。
主要是有两个方法：

1、await （a 是 all wait 是等待），主线程来调用这个方法。

2、countDown 表示选手冲过了终点线。

CountDownLatch 在构造的时候，会指定一个计数（选手的个数）

例如，指定四个选手进行比赛，初始情况下，调用 await 就会阻塞。
每个选手都冲过终点，都会调用 countDown 方法。

第三次调用 countDown ，await 没有任何影响。
第四次调用 countDown ，await 就会被唤醒，返回。（解除阻塞队列）
此时就可以认为是整个比赛都结束了。

package thread;

import java.util.concurrent.CountDownLatch;

public class ThreadDemo12 {

    public static void main(String[] args) throws InterruptedException{
        CountDownLatch latch = new CountDownLatch(10);
        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                try {
                    Thread.sleep(1000);
                    latch.countDown();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        };
        for (int i = 0; i < 10; i++) {
            new Thread(runnable).start();
        }

        // 必须等到 10 人全部回来
        latch.await();
        System.out.println("比赛结束");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

6. 相关面试题

1、介绍下 Callable 是什么

Callable 是一个 interface，相当于把线程封装了一个 “返回值”。
方便程序猿借助多线程的方式计算结果。

Callable 和 Runnable 相对，都是描述一个 “任务”。
Callable 描述的是带有返回值的任务，Runnable 描述的是不带返回值的任务。

Callable 通常需要搭配 FutureTask 来使用。FutureTask 用来保存 Callable 的返回结果。
因为 Callable 往往是在另一个线程中执行的，啥时候执行完并不确定。
FutureTask 就可以负责这个等待结果出来的工作。


2、线程同步的方式有哪些？

synchronized，ReentrantLock，Semaphore 等都可以用于线程同步。


3、为什么有了 synchronized 还需要 juc 下的 lock？

以 juc 的 ReentrantLock 为例。

• synchronized 使用时不需要手动释放锁，ReentrantLock 使用时需要手动释放，使用起来更
  灵活。
• synchronized 在申请锁失败时，会死等，ReentrantLock 可以通过 trylock 的方式等待一段时
  间就放弃。
• synchronized 是非公平锁，ReentrantLock 默认是非公平锁。可以通过构造方法传入一个
  true 开启公平锁模式。
• synchronized 是通过 Object 的 wait / notify 实现等待-唤醒。每次唤醒的是一个随机等待的
  线程。ReentrantLock 搭配 Condition 类实现等待-唤醒，可以更精确控制唤醒某个指定的线
  程。

4、AtomicInteger 的实现原理是什么？

基于 CAS 机制。

伪代码如下：

class AtomicInteger {
    private int value;
    public int getAndIncrement() {
        int oldValue = value;
        while ( CAS(value, oldValue, oldValue+1) != true) {
            oldValue = value;
        }
       return oldValue;
    }
}
1
2
3
4
5
6
7
8
9
10

5、信号量听说过么？之前都用在过哪些场景下？

信号量，用来表示 “可用资源的个数”，本质上就是一个计数器。

使用信号量可以实现 “共享锁”，比如某个资源允许 3 个线程同时使用，那么就可以使用 P 操作作为
加锁，V 操作作为解锁，前三个线程的 P 操作都能顺利返回，后续线程再进行 P 操作就会阻塞等待，
直到前面的线程执行了 V 操作。


6、解释一下 ThreadPoolExecutor 构造方法的参数的含义

参考关于 ThreadPoolExecutor 的篇章

篇章链接：

https://blog.csdn.net/m0_63033419/article/details/128586070?spm=1001.2014.3001.5501