java-07 多线程-并发编程(原子变量、CountDownLatch、Future和CompletableFuture、volatile)

并发编程是指在一个程序中同时执行多个任务或线程。这通常涉及到多线程编程、线程同步、并发容器等技术。这些技术可以用来解决多线程环境中的问题，如线程安全、资源竞争、死锁等问题。在实际的Java并发编程中，还需要考虑到线程池、Future、Callable、ExecutorService等概念。

在另一篇文章介绍了多线程、进程、并发、并行等基本概念，并分析了线程安全问题产生的原因，同时也整理了线程实现的4种方式，并做了对比，请参考 java-06 多线程-4种实现方式

如果你觉得我分享的内容或者我的努力对你有帮助，或者你只是想表达对我的支持和鼓励，请考虑给我点赞、评论、收藏。您的鼓励是我前进的动力，让我感到非常感激。

文章目录

1 线程同步
- 1.1 锁
- 1.2 原子变量
- 1.3 ThreadLocal
- 1.4 CountDownLatch
- 1.5 Future 和 CompletableFuture
- 1.6 volatile

1 线程同步

1.1 锁

java-07 多线程-并发编程(锁)

1.2 原子变量

除了 StampedLock 中的乐观读锁，java.util.concurrent.atomic 包下面 AtomicInteger、AtomicLong、AtomicIntegerArray、AtomicReference 等原子变量类也是基于乐观锁的思想实现的。

不过普通的 AtomicInteger 可能会存在 ABA 问题，此时可以使用 AtomicStampedReference，它内部除了一个对象引用，还维护了一个可以自动更新的整数，通过标识版本来避免 ABA 问题。

AtomicInteger 的基本使用示例：

import java.util.concurrent.atomic.AtomicInteger;public class Main {private static final AtomicInteger counter = new AtomicInteger(0);public static void main(String[] args) {Runnable incrementTask = () -> {for (int i = 0; i < 10000; i++) {counter.incrementAndGet();}};Thread thread1 = new Thread(incrementTask);Thread thread2 = new Thread(incrementTask);thread1.start();thread2.start();try {thread1.join();thread2.join();} catch (InterruptedException e) {e.printStackTrace();}System.out.println("Final counter value: " + counter.get());}
}

AtomicReference 的基本使用示例：

import java.util.concurrent.atomic.AtomicReference;public class Main {private static final AtomicReference<Integer> counterRef = new AtomicReference<>(0);public static void main(String[] args) {Runnable incrementTask = () -> {for (int i = 0; i < 10000; i++) {while (true) {Integer current = counterRef.get();Integer updated = current + 1;if (counterRef.compareAndSet(current, updated)) {break;}}}};Thread thread1 = new Thread(incrementTask);Thread thread2 = new Thread(incrementTask);thread1.start();thread2.start();try {thread1.join();thread2.join();} catch (InterruptedException e) {e.printStackTrace();}System.out.println("Final counter value: " + counterRef.get());}
}

1.3 ThreadLocal

1.4 CountDownLatch

CountDownLatch 是一个同步辅助类，它允许一个或多个线程等待其他线程完成操作。CountDownLatch 通过一个计数器来实现一个线程等待其他线程完成某个操作，当计数器的值减为零时，等待的线程被释放。

CountDownLatch 主要提供了以下方法：

CountDownLatch(int count)：构造方法，用于初始化计数器，指定计数器的初始值为 count。
void await() throws InterruptedException：当调用线程调用此方法时，它会一直等待，直到计数器减为零。如果计数器不为零，线程将被阻塞。
boolean await(long timeout, TimeUnit unit) throws InterruptedException：在指定的时间内等待计数器减为零，如果在指定时间内计数器未减至零，线程将被唤醒。
void countDown()：每个被等待的线程执行完任务后，都应该调用此方法来减小计数器的值。

以下是一个 CountDownLatch 的简单的示例：

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;public class Main {public static void main(String[] args) throws InterruptedException {ExecutorService es = Executors.newFixedThreadPool(10);CountDownLatch latch = new CountDownLatch(10);Runnable task = () -> {System.out.println("task");latch.countDown();};long begin = System.currentTimeMillis();for (int i = 0; i < 10; i++) {es.submit(task);}latch.await();long end = System.currentTimeMillis();System.out.println("time = " + (end - begin));}
}

1.5 Future 和 CompletableFuture

Future 是一个接口，允许我们提交一个任务给线程池或其他异步执行机制，并且在未来获取该任务的结果。它通常用于处理耗时的操作，以便不阻塞主线程，从而提高程序的性能和响应性。

Future 接口相关方法：

get()：用来获取异步任务的结果，如果任务尚未完成，调用 get() 将会阻塞当前线程，直到任务完成并返回结果。
isDone()：用来检查异步任务是否已经完成，如果任务已经完成，它将返回 true，否则返回 false。
cancel(boolean mayInterruptIfRunning)：用于取消异步任务的执行，mayInterruptIfRunning 参数用于指定是否应该中断正在执行的任务。如果任务成功取消，get() 方法将会抛出 CancellationException。

Future 接口通常通过 ExecutorService 接口的实现来使用，ExecutorService 提供了一种提交任务并获取 Future 的方式，从而管理线程池中的任务。

此外，Java 8 中还引入了 CompletableFuture 类，它实现了 Future 接口并提供了更丰富的功能，包括支持函数式编程、组合多个异步任务等。

Future 异步执行任务示例：

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;class Solution {public static void main(String[] args) {ExecutorService executor = Executors.newFixedThreadPool(2);Future<Integer> future = executor.submit(() -> {Thread.sleep(2000);  // 模拟一个耗时操作return 42;});try {System.out.println("等待任务完成...");Integer result = future.get();System.out.println("任务完成，结果为：" + result);} catch (InterruptedException | ExecutionException e) {e.printStackTrace();}executor.shutdown();}
}

CompletableFuture 异步执行任务示例：
分批处理数据

批量删除数据

@Async("commonCmdbExecutor")
public CompletableFuture<CmdbResultVo> multiDeleteObjInstance(CmdbDeleteObjInstanceReqVo body,CountDownLatch latch) {String url = cmdbHost + CmdbConstants.MULTI_DELETE_OBJ_INSTANCE_URL;ResponseEntity<CmdbResultVo> exchange = null;try {exchange = restTemplate.exchange(url, HttpMethod.POST, new HttpEntity<Object>(body, new HttpHeaders()),CmdbResultVo.class);} catch (Exception e) {LOG.error("delete data error", e);} finally {latch.countDown();}return exchange == null ? null : CompletableFuture.completedFuture(exchange.getBody());
}public int deleteCmdbDataAsync(String objCode, List<String> ids) {int groupQty = ids.size() / DELETE_MAX_SIZE;int resultCode = 1;CountDownLatch latch = new CountDownLatch(groupQty + 1);List<CompletableFuture<CmdbResultVo>> futures = new CopyOnWriteArrayList<>();for (int i = 0; i <= groupQty; i++) {CmdbDeleteObjInstanceReqVo deleteBody = new CmdbDeleteObjInstanceReqVo();deleteBody.setObjCode(objCode);List<String> curGroup = getCurGroup(i, DELETE_MAX_SIZE, ids);deleteBody.setInstanceIdList(curGroup);CompletableFuture<CmdbResultVo> cmdbResultVoCompletableFuture =cmdbClient.multiDeleteObjInstance(deleteBody, latch);futures.add(cmdbResultVoCompletableFuture);}try {latch.await(3, TimeUnit.MINUTES);} catch (InterruptedException e) {LOG.error("InterruptedException:", e);Thread.currentThread().interrupt();} catch (Exception e) {LOG.error("Exception:", e);}for (CompletableFuture<CmdbResultVo> future : futures) {try {CmdbResultVo response = future.get();if (response.getResultCode() == 0) {resultCode = 0;}} catch (Exception e) {LOG.error("Exception:", e);}}return resultCode;
}

批量更新数据

    @Async("commonCmdbExecutor")public <T> CompletableFuture<CmdbResultVo> multiUpdateObjInstanceAsync(CmdbUpdateObjInstanceReqVo<T> body,CountDownLatch latch) {String url = cmdbHost + CmdbConstants.MULTI_UPDATE_OBJ_INSTANCE_URL;ResponseEntity<CmdbResultVo> exchange = null;try {exchange = restTemplate.exchange(url, HttpMethod.POST, new HttpEntity<Object>(body, new HttpHeaders()),CmdbResultVo.class);} catch (Exception e) {LOG.error("update data error", e);} finally {latch.countDown();}return exchange == null ? null : CompletableFuture.completedFuture(exchange.getBody());}

1.6 volatile

volatile 关键字主要有两层语义：

保证多线程环境下共享变量操作的可见性：参考 Java 内存模型（JMM）中先行发生（Happens-Before）原则对 volatile 变量规则的描述，对于一个 volatile 变量，如果对于这个变量的写操作先行发生于这个变量的读操作，那么这个写操作所产的影响对于后续的读操作是可见的。
禁止指令重排序：编译器在编译时会在生成的字节码中插入特定的内存屏障指令，确保在 volatile 变量读写操作前后的代码不会被重排序。具体来说，会在 volatile 变量写操作之后，读操作之前插入屏障，因此执行到 volatile 变量读写操作时，前面的操作一定已经执行完成，后面的操作一定还未开始。
voliatile有三大特性：