JUC（java.util.concurrent）的常见类

小猪同学hy 2024-08-22 14:35:02 阅读 89

前言

在前面我们已经学习java.util.concurrent包（实现多线程并发编程常用的包）中的一些操作，例如线程池、阻塞队列等，那么本篇我们就来学习一下JUC中几种其他常见的类。

Callable和Future

Callable

Callable是一个泛型接口，只有一个方法 call().用于创建可以返回结果的任务，与Runnable不同，Callable可以返回一个结果，并且可以抛出异常，需要依赖FutureTask类来获取返回结果。

我们查看Callable的接口定义：

在使用Callable接口的时候，我们需要将我们的任务需求编写到call()方法中。

示例：现在要计算从1加到1000的和

<code>class Demo1 {

public static void main(String[] args) throws ExecutionException, InterruptedException {

Callable<Integer> call = new Callable<Integer>() {

@Override

public Integer call() throws Exception {

int sum = 0;

for (int i = 0; i <= 1000; i++) {

sum += i;

}

return sum;

}

};

}

在这个代码中，我们实现了从1加到1000的和的计算，当在完成计算之后，会返回一个Integer类型的数据。

Future

Future同样是一个泛型接口，用来表示异步计算的结果，在接口中提供了一些方法来检查任务是否完成、获取计算结果以及取消任务的执行。以下是Future接口的定义：

public interface Future<V> {

boolean cancel(boolean mayInterruptIfRunning);

boolean isCancelled();

boolean isDone();

V get() throws InterruptedException, ExecutionException;

V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;

}

cancel((boolean mayInterruptIfRunning):尝试取消任务的执行；isCancelled()：检查任务是否已经被取消；isDone()：检查任务是否已经完成；get()：获取任务的结果，如果任务尚未完成，就会阻塞当前线程，直到任务完成；get(long timeout, TimeUnit unit)：在指定的时间内获取任务的结果，如果任务尚未完成，则阻塞当前线程。

Future有它的实现类FutureTask供我们使用，我们可以看其文档解释，这段话的意思就是说：可取消的异步计算。FutureTask类提供了Future的基本实现，其中包括启动和取消计算、查询计算是否完成以及检索计算结果的方法。只有在计算完成后才能检索结果;如果计算尚未完成，get方法将阻塞。一旦计算完成，就不能重新启动或取消计算(除非使用runAndReset调用计算)。

FutureTask可以用来包装一个Callable或Runnable对象。因为FutureTask实现了Runnable，所以FutureTask可以提交给Executor执行。

除了作为一个独立的类，这个类还提供了受保护的功能，这在创建自定义任务类时可能很有用。

我们可以在java8文档中查看相关方法：

使用Callable和FutureTask

示例一：创建线程计算1+2+...+1000的和，使用Callable.

<code> /**

* 主函数，用于演示如何通过Callable接口实现多线程计算

* 本函数的目标是计算1到1000的和，通过创建一个Callable任务，并在新线程中执行该任务来实现

* @param args 命令行参数

* @throws ExecutionException 如果任务未能成功完成，则抛出此异常

* @throws InterruptedException 如果当前线程在等待任务完成时被中断，则抛出此异常

*/

public static void main(String[] args) throws ExecutionException, InterruptedException {

// 创建一个Callable对象，用于计算1到1000的和

Callable<Integer> call=new Callable<Integer>() {

/**

* 执行计算任务的方法

* 本方法通过循环计算1到1000的和

* @return 计算结果，即1到1000的和

* @throws Exception 如果在计算过程中发生错误，则抛出此异常

*/

@Override

public Integer call() throws Exception {

int sum=0;

for(int i=0;i<=1000;i++){

sum+=i;

}

return sum;

}

};

// 使用Callable对象创建一个FutureTask，以便可以在新线程中执行任务

FutureTask<Integer> futureTask=new FutureTask<>(call);

// 创建一个新线程，并将FutureTask对象作为任务传递给该线程

Thread t=new Thread(futureTask);

// 启动新线程，开始执行计算任务

t.start();

// 获取计算任务的结果，主线程将在这里等待直到任务完成

int result=futureTask.get();

// 打印计算结果

System.out.println(result);

}

结果

示例二：使用单个线程的线程池

<code> public static void main(String[] args) throws ExecutionException, InterruptedException {

// 创建一个固定大小的线程池，这里只创建一个线程

ExecutorService ex=Executors.newFixedThreadPool(1);

// 创建一个Callable对象，用于执行计算任务

Callable<Integer> calls=new Callable<Integer>() {

@Override

public Integer call() throws Exception {

// 初始化求和变量

int sum=0;

// 计算从0到1000的累加和

for(int i=0;i<=1000;i++){

sum+=i;

}

// 返回计算结果

return sum;

}

};

// 创建一个FutureTask对象，并将Callable对象作为参数传递给它

FutureTask<Integer> futureTask= (FutureTask<Integer>) ex.submit(calls);

// 获取计算任务的结果，主线程将在这里等待直到任务完成

int result=futureTask.get();

// 打印计算结果

System.out.println(result);

// 关闭ExecutorService

ex.shutdown();

}

在这段代码中，我们创建了一个只有一个线程的线程池，并且创建了一个Callable对象并重写其中的call方法。通过调用submit方法提交任务并获取FutureTask对象。但由于此时返回的是Futue类型的结果，所以我们需要将其强转为FutureTask<Integer>。同时我们通过调用其中的get方法来等待任务执行完成并获取到结果。需要注意，这里我们需要手动关闭线程池。释放资源。

示例三：拥有多个线程的线程池

class Demo4{

// 主函数，展示如何使用固定大小的线程池执行Callable任务

public static void main(String[] args) throws ExecutionException, InterruptedException {

// 创建一个固定大小为4的线程池

ExecutorService ex=Executors.newFixedThreadPool(4);

// 创建一个可返回结果的Callable任务

Callable<String> calls=new Callable<String>() {

// 实现call方法，当任务被线程池执行时，此方法将被调用

@Override

public String call() throws Exception {

// 输出当前线程的名称和执行的信息

System.out.println(Thread.currentThread().getName()+"执行了call方法");

// 返回任务执行结果

return "result";

}

};

// 循环提交4个任务到线程池

for(int i=0;i<4;i++){

// 提交任务并获取FutureTask对象，用于查询任务执行结果

FutureTask<String> futureTask= (FutureTask<String>) ex.submit(calls);

// 获取并输出任务的执行结果

System.out.println(futureTask.get());

}

// 关闭线程池，不再接受新的任务

ex.shutdown();

}

ReentrantLock

ReentrantLock是一个可重入互斥锁，允许一个线程多次获取同一个锁而不会产生死锁，和synchronized类似，都是用来保证线程安全的。

ReentrantLock加锁有两种方式：

lock()：加锁，如果获取不到锁就进入阻塞等待。trylock()：加锁，如果获取不到锁，就会放弃加锁。unlock()：解锁。

在使用ReentrantLock的时候需要手动释放锁unlock(),如果忘记解锁，可能会带来比较严重的后果。所以我们可以使用try-finally来进行结果操作。

<code> ReentrantLock lock=new ReentrantLock();

try{

lock.lock();

//working

}finally {

lock.unlock();

}

ReentrantLock和synchronized的区别

ReentrantLock可以使用lock()和tryLock()进行加锁，使用lock()的时候，若没有获取到锁就会进入阻塞等待，而使用tryLock()的时候，如果没有获取到锁，就会放弃获取而不是阻塞等待。synchronized是一个关键字，是JVM内部实现的；ReentrantLock是标准库中的一个类，在JVM外部实现的（基于Java实现）。synchronized不需要手动解锁，而ReentrantLock需要手动释放锁，使用起来更加灵活，但是也容易遗漏unlock，所以最好在加上try-finally使用。synchronized在申请锁失败时,会死等.ReentrantLock可以通过trylock的⽅式等待⼀段时间就放弃.synchronized是⾮公平锁,ReentrantLock默认是⾮公平锁.可以通过构造⽅法传⼊⼀个true开启公平锁模式.更强⼤的唤醒机制.synchronized是通过Object的wait/notify实现等待-唤醒.每次唤醒的是⼀个随机等待的线程.而ReentrantLock搭配Condition类实现等待-唤醒,可以更精确控制唤醒某个指定的线程.

如何选择使用哪个锁？

锁竞争不激烈的时候,使⽤synchronized,效率更⾼,⾃动释放更⽅便

锁竞争激烈的时候,使⽤ReentrantLock,搭配trylock更灵活控制加锁的⾏为,⽽不是死等.

如果需要使⽤公平锁,使⽤ReentrantLock.

虽然ReentrantLock使用起来比较灵活，但一般情况下建议使用synchronized。

原子类

原子类内部其实是使用CAS实现的，CAS是一个原子的操作，不需要加锁，性能比加锁的要好多。

• AtomicBoolean

• AtomicInteger

• AtomicIntegerArray

• AtomicLong

• AtomicReference

• AtomicStampedReference

CAS的原理以及如何使用CAS在上一篇我已经讲过，想了解更多的可以看看CAS原理

线程池

线程池是为了解决频繁创建和销毁线程带来的性能和资源浪费问题。

在java中使用线程池我们需要用到ExecutorService和Executors。

ExecutorService 表示一个线程池实例.Executors 是一个工厂类, 能够创建出几种不同风格的线程池.ExecutorService 的 submit 方法能够向线程池中提交若干个任务

Executors是一个工厂类，里面提供了创建不同风格线程池的方法。

newFixedThreadPool: 创建固定线程数的线程池newCachedThreadPool: 创建线程数目动态增长的线程池.newSingleThreadExecutor: 创建只包含单个线程的线程池.newScheduledThreadPool: 设定延迟时间后执行命令，或者定期执行命令. 是进阶版的 Timer。

Executors本质上是对ThreradPoolExecutor的封装。

感兴趣的可以看看这一篇【JavaEE】线程池-CSDN博客，里面讲解了关于ThreadPoolExecutor构造方法的相关参数的含义。

示例：

class Demo5{

/**

* 程序入口

* @param args 命令行参数

*/

public static void main(String[] args) {

// 创建一个固定大小的线程池，用于执行任务

ExecutorService ex=Executors.newFixedThreadPool(4);

// 提交一个运行时打印"hello"的无返回值任务到线程池

ex.submit(()->{

System.out.println("hello");

});

// 关闭线程池，不再接受新的任务提交，但已提交的任务将继续执行完成

ex.shutdown();

}

Semaphore（信号量）

信号量（Semaphore）又称为信号灯。用来表示“可用资源的个数”，本质上就是一个计数器。在多线程环境下用于协调各个线程, 以保证它们能够正确、合理的使用公共资源。

信号量是一个非负整数，当我们申请资源的时候，计数器就会-1，称为“P操作”，释放资源的时候，计数器就会+1，也称为“V操作”。当计数器为0时，如果再申请资源，就会阻塞等待，直到有其他线释放资源。

Semaphore的PV操作中的加减计数器操作都是原⼦的,可以在多线程环境下直接使用.

信号量可以分为二元信号量和计数信号量两种，二元信号量就相当于一把锁，当有线程申请资源时，就相当于加锁，此时计数器-1，信号量就为0；当有其他线程想要申请资源就会陷入阻塞等待，直到占用资源的线程释放，才能获取到。计数信号量就表示此时有多少可以被申请的资源，当有线程申请资源，计数器就-1，当有线程释放资源，计数器就+1。

加锁和解锁的操作就可以看做，加锁时信号量为0，解锁信号量为1.

在java中，把信号量的相关操作封装在Semaphore中，acquire()方法表示申请资源，release()表示释放资源。availablePermits()可以查看信号量中还有多少资源可用。

示例：

/**

* SemaphoreDemo类用于演示信号量的应用

*/

class SemaphoreDemo{

// 用于计数的共享变量

static int count=0;

/**

* 程序的入口点

* @param args 命令行参数

* @throws InterruptedException 如果线程被中断

*/

public static void main(String[] args) throws InterruptedException {

// 创建一个信号量，初始值为1，用于控制同时只能有一个线程执行临界区代码

Semaphore semaphore=new Semaphore(1);

// 创建一个公平的可重入锁，本例中未直接使用，但展示了如何创建

ReentrantLock locker=new ReentrantLock(true);

// 创建第一个线程，用于增加计数器

Thread t1=new Thread(()->{

for (int i=0;i<1000;i++) {

try {

// 获取信号量，允许进入临界区

semaphore.acquire();

// 执行临界区操作：增加计数器

count++;

// 释放信号量，允许其他线程进入临界区

semaphore.release();

} catch (InterruptedException e) {

// 如果线程被中断，则抛出运行时异常

throw new RuntimeException(e);

}

});

// 创建第二个线程，同样用于增加计数器

Thread t2=new Thread(()->{

for (int i=0;i<1000;i++) {

try {

// 获取信号量，允许进入临界区

semaphore.acquire();

// 执行临界区操作：增加计数器

count++;

// 释放信号量，允许其他线程进入临界区

semaphore.release();

} catch (InterruptedException e) {

// 如果线程被中断，则抛出运行时异常

throw new RuntimeException(e);

}

});

// 启动第一个线程

t1.start();

// 启动第二个线程

t2.start();

// 等待第一个线程结束

t1.join();

// 等待第二个线程结束

t2.join();

// 输出计数结果

System.out.println(count);

}

CountDownLatch

CountDownLatch是java中的一个同步工具类，用来协调多个线程之间的同步，初始值为线程的数量。

CountDownLatch通过一个计数器功能，当一个线程完成了自己的任务，计数器的值就-1，当计数器为0时，说明所有的线程都完成任务，此时在CountDownLatch等待的线程就可以恢复执行。

CountDownLatch一个典型用法就是把一个大任务拆分成N个小任务，让多个线程来执行小任务，每个线程执行完自己的任务计数器就-1，当所有小任务都完成后，等待所有小任务完成的线程才继续往下执行。就好比富士康手机加工的流水线一样，组装一步手机需要一条条的流水线来相互配合完成。一条条流水线（Worker），每条线都干自己的活。有的流水线是贴膜的，有的流水线是打螺丝的，有的流水线是质检的、有的流水线充电的、有的流水线贴膜的。等这些流水线都干完了就把一部手机组装完成了。

方法：

CountDownLatch(int count)：count为计数器的初始值（一般需要多少个线程执行，count就设为几）。

countDown(): 每调用一次计数器值-1，直到count被减为0，代表所有线程全部执行完毕。

getCount()：获取当前计数器的值。

await(): 等待计数器变为0，即等待所有异步线程执行完毕。

boolean await(long timeout, TimeUnit unit)：

此方法至多会等待指定的时间，超时后会自动唤醒，若 timeout 小于等于零，则不会等待boolean 类型返回值：若计数器变为零了，则返回 true；若指定的等待时间过去了，则返回 false

<code>class CountDownLatchDemo{

public static void main(String[] args) throws InterruptedException {

//创建一个固定大小的线程池，用于处理下载任务

ExecutorService service= Executors.newFixedThreadPool(4);

//创建一个计数器，用于等待所有下载任务完成

CountDownLatch count=new CountDownLatch(20);//20个任务

//提交20个下载任务到线程池

for(int i=1;i<=20;i++){

int id=i;

//使用lambda表达式创建匿名内部类，定义每个下载任务的执行逻辑

service.submit(()->{

System.out.println("下载任务"+id+"正在执行");

try {

//模拟下载任务的执行时间

Thread.sleep(3000);

} catch (InterruptedException e) {

//处理中断异常

throw new RuntimeException(e);

}

//任务完成

System.out.println("下载任务"+id+"执行完毕");

//计数器减1，表示一个任务完成

count.countDown();

});

}

//等待所有任务完成，计数器归零

count.await();

System.out.println("所有下载任务已完成");

//关闭线程池

service.shutdown();

}

当调用了20次 countDown() 方法之后，await() 方法才会结束等待，继续执行后面的代码。

需要注意的是，CountDownLatch是一次性的，一旦计数器的值达到0，就不能再次使用。如果需要多次使用类似的功能，可以考虑使用CyclicBarrier等其他同步工具类。