【并发编程】Java并发编程核心包

1、简介

java.util.concurrent 是 Java 并发编程的核心包，提供了丰富的工具和框架来支持多线程编程、并发任务执行、线程安全集合、同步机制等。

2、线程池Thread Pool

线程池是并发编程中最重要的工具之一，用于管理和复用线程，避免频繁创建和销毁线程的开销。

最顶层接口Executor，核心接口ExecutorService

2.1、主要类

工厂类（Executors）

工厂类Executors提供了创建线程池的静态方法。

• newFixedThreadPool(int nThreads)：创建固定大小的线程池。

• newCachedThreadPool()：创建可缓存的线程池，线程数根据需要动态调整。

• newSingleThreadExecutor()：创建单线程的线程池。

• newScheduledThreadPool(int corePoolSize)：创建支持定时任务的线程池。

package com.mqtt.mqttproject.test;import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;/*** @ Author : Gridsum* @ Description :*/
public class ThreadPoolExample {public static void main(String[] args) {ExecutorService executor = Executors.newFixedThreadPool(2);for (int i = 0; i< 5; i++){executor.submit(()->{System.out.println("任务执行" + Thread.currentThread().getName());});}executor.shutdown(); //关闭线程池}
}

核心接口（ExecutorService）

ExecutorService接口提供了提供了线程池的管理方法。

ExecutorService接口常用实现类

2.2、ThreadPoolExecutor

 

public ThreadPoolExecutor(int corePoolSize, //核心线程数int maximumPoolSize, //最大线程数long keepAliveTime, //线程空闲时间TimeUnit unit, //时间单位BlockingQueue<Runnable> workQueue, //队列ThreadFactory threadFactory, //线程工厂RejectedExecutionHandler handler//拒绝策略) {if (corePoolSize < 0 ||maximumPoolSize <= 0 ||maximumPoolSize < corePoolSize ||keepAliveTime < 0)throw new IllegalArgumentException();if (workQueue == null || threadFactory == null || handler == null)throw new NullPointerException();this.corePoolSize = corePoolSize;this.maximumPoolSize = maximumPoolSize;this.workQueue = workQueue;this.keepAliveTime = unit.toNanos(keepAliveTime);this.threadFactory = threadFactory;this.handler = handler;}

2.2.1、线程池执行顺序

当线程数小于核心线程数时，创建线程。
当线程数大于等于核心线程数，且任务队列未满时，将任务放入任务队列。
当线程数大于等于核心线程数，且任务队列已满，若线程数小于最大线程数，创建线程。
若线程数等于最大线程数，则执行拒绝策略

2.2.2、workQueue

（1）无界队列
        队列大小无限制，常用的为无界的LinkedBlockingQueue，使用该队列作为阻塞队列时要尤其当心，当任务耗时较长时可能会导致大量新任务在队列中堆积最终导致OOM。阅读代码发现，Executors.newFixedThreadPool 采用就是 LinkedBlockingQueue，而博主踩到的就是这个坑，当QPS很高，发送数据很大，大量的任务被添加到这个无界LinkedBlockingQueue 中，导致cpu和内存飙升服务器挂掉。
        当然这种队列，maximumPoolSize 的值也就无效了。当每个任务完全独立于其他任务，即任务执行互不影响时，适合于使用无界队列；例如，在 Web 页服务器中。这种排队可用于处理瞬态突发请求，当命令以超过队列所能处理的平均数连续到达时，此策略允许无界线程具有增长的可能性。
  （2）有界队列
        当使用有限的 maximumPoolSizes 时，有界队列有助于防止资源耗尽，但是可能较难调整和控制。常用的有两类，一类是遵循FIFO原则的队列如ArrayBlockingQueue，另一类是优先级队列如PriorityBlockingQueue。PriorityBlockingQueue中的优先级由任务的Comparator决定。
使用有界队列时队列大小需和线程池大小互相配合，线程池较小有界队列较大时可减少内存消耗，降低cpu使用率和上下文切换，但是可能会限制系统吞吐量。
  （3）同步移交队列
        如果不希望任务在队列中等待而是希望将任务直接移交给工作线程，可使用SynchronousQueue作为等待队列。SynchronousQueue不是一个真正的队列，而是一种线程之间移交的机制。要将一个元素放入SynchronousQueue中，必须有另一个线程正在等待接收这个元素。只有在使用无界线程池或者有饱和策略时才建议使用该队列。

2.2.3、threadFactory 

线程工厂，用来创建线程， 为了统一在创建线程时设置一些参数，如是否守护线程，线程一些特性等，如优先级。通过这个TreadFactory创建出来的线程能保证有相同的特性。
它是一个接口类，而且方法只有一个，就是创建一个线程。如果没有另外说明，则在同一个ThreadGroup 中一律使用Executors.defaultThreadFactory() 创建线程，并且这些线程具有相同的NORM_PRIORITY 优先级和非守护进程状态。
通过提供不同的 ThreadFactory，可以改变线程的名称、线程组、优先级、守护进程状态，等等。
如果从newThread 返回 null 时ThreadFactory 未能创建线程，则执行程序将继续运行，但不能执行任何任务。 

2.2.4、ThreadPoolExecutor执行逻辑

创建ThreadPoolExecutor实例对象->execute()/submit()->addWorker->runWorker->getTask

 

package com.mqtt.mqttproject.test;import java.util.concurrent.*;/*** @ Author : Gridsum* @ Description :*/
public class ThreadPoolExample {public static void main(String[] args) {LinkedBlockingDeque linkedBlockingDeque = new LinkedBlockingDeque(10);ThreadFactory threadFactory = Executors.defaultThreadFactory();RejectedExecutionHandler rejectedExecutionHandler = new ThreadPoolExecutor.AbortPolicy();ExecutorService executor = new ThreadPoolExecutor(5,10,1000,TimeUnit.MILLISECONDS,linkedBlockingDeque, threadFactory,rejectedExecutionHandler);//如果不需要获取任务执行结果，调用execute方法executor.execute(()->{System.out.println("开始执行任务" + Thread.currentThread().getName());});//如果需要返回任务执行结果，调用submit方法Future<String> future = executor.submit(()->{Thread.sleep(3000);System.out.println("开始执行任务" + Thread.currentThread().getName());return "执行成功";});//添加回调addCallback(future, result -> {System.out.println("回调结果：" + result);});System.out.println("执行其他任务，不需要等待回调结果");executor.shutdown();}public interface Callback{void onComplate(String result);}public static void addCallback(Future<String> future, Callback callback){new Thread(()->{try {callback.onComplate(future.get());} catch (InterruptedException e) {e.printStackTrace();} catch (ExecutionException e) {e.printStackTrace();}}).start();}}

执行结果

 ThreadPoolExecutor核心维护了一个HashSet<Worker>类型的 workers，用来存放线程，维护了BlockingQueue<Runnable> workQueue 队列，存放需要执行的任务task，启动workers中的线程执行workQueue队列中task。

public void execute(Runnable command) {if (command == null)throw new NullPointerException();int c = ctl.get();//workerCountOf获取线程池的当前线程数；小于corePoolSize，执行addWorker创建新线程执行command任务if (workerCountOf(c) < corePoolSize) {if (addWorker(command, true))return;c = ctl.get();}if (isRunning(c) && workQueue.offer(command)) {int recheck = ctl.get();if (! isRunning(recheck) && remove(command))reject(command);else if (workerCountOf(recheck) == 0)addWorker(null, false);}else if (!addWorker(command, false))reject(command);}

private boolean addWorker(Runnable firstTask, boolean core) {retry:for (int c = ctl.get();;) {// Check if queue empty only if necessary.if (runStateAtLeast(c, SHUTDOWN)&& (runStateAtLeast(c, STOP)|| firstTask != null|| workQueue.isEmpty()))return false;for (;;) {if (workerCountOf(c)>= ((core ? corePoolSize : maximumPoolSize) & COUNT_MASK))return false;if (compareAndIncrementWorkerCount(c))break retry;c = ctl.get();  // Re-read ctlif (runStateAtLeast(c, SHUTDOWN))continue retry;// else CAS failed due to workerCount change; retry inner loop}}boolean workerStarted = false;boolean workerAdded = false;Worker w = null;try {w = new Worker(firstTask);final Thread t = w.thread;if (t != null) {final ReentrantLock mainLock = this.mainLock;mainLock.lock();try {// Recheck while holding lock.// Back out on ThreadFactory failure or if// shut down before lock acquired.int c = ctl.get();if (isRunning(c) ||(runStateLessThan(c, STOP) && firstTask == null)) {if (t.isAlive()) // precheck that t is startablethrow new IllegalThreadStateException();workers.add(w);int s = workers.size();if (s > largestPoolSize)largestPoolSize = s;workerAdded = true;}} finally {mainLock.unlock();}if (workerAdded) {t.start();workerStarted = true;}}} finally {if (! workerStarted)addWorkerFailed(w);}return workerStarted;}

Worker类的runworker方法

 private final class Worker extends AbstractQueuedSynchronizer implements Runnable{Worker(Runnable firstTask) {setState(-1); // inhibit interrupts until runWorkerthis.firstTask = firstTask;this.thread = getThreadFactory().newThread(this); // 创建线程}/** Delegates main run loop to outer runWorker  */public void run() {runWorker(this);}// ...}

runWorker方法是线程池的核心:

    线程启动之后，通过unlock方法释放锁，设置AQS的state为0，表示运行可中断；
    Worker执行firstTask或从workQueue中获取任务:
        进行加锁操作，保证thread不被其他线程中断(除非线程池被中断)
        检查线程池状态，倘若线程池处于中断状态，当前线程将中断。
        执行beforeExecute
        执行任务的run方法
        执行afterExecute方法
        解锁操作

通过getTask方法从阻塞队列中获取等待的任务，如果队列中没有任务，getTask方法会被阻塞并挂起，不会占用cpu资源；

final void runWorker(Worker w) {Thread wt = Thread.currentThread();Runnable task = w.firstTask;w.firstTask = null;w.unlock(); // allow interruptsboolean completedAbruptly = true;try {while (task != null || (task = getTask()) != null) {w.lock();// If pool is stopping, ensure thread is interrupted;// if not, ensure thread is not interrupted.  This// requires a recheck in second case to deal with// shutdownNow race while clearing interruptif ((runStateAtLeast(ctl.get(), STOP) ||(Thread.interrupted() &&runStateAtLeast(ctl.get(), STOP))) &&!wt.isInterrupted())wt.interrupt();try {beforeExecute(wt, task);try {task.run();afterExecute(task, null);} catch (Throwable ex) {afterExecute(task, ex);throw ex;}} finally {task = null;w.completedTasks++;w.unlock();}}completedAbruptly = false;} finally {processWorkerExit(w, completedAbruptly);}}

getTask方法：

private Runnable getTask() {boolean timedOut = false; // Did the last poll() time out?for (;;) {int c = ctl.get();// Check if queue empty only if necessary.if (runStateAtLeast(c, SHUTDOWN)&& (runStateAtLeast(c, STOP) || workQueue.isEmpty())) {decrementWorkerCount();return null;}int wc = workerCountOf(c);// Are workers subject to culling?boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;if ((wc > maximumPoolSize || (timed && timedOut))&& (wc > 1 || workQueue.isEmpty())) {if (compareAndDecrementWorkerCount(c))return null;continue;}try {Runnable r = timed ?workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :workQueue.take();if (r != null)return r;timedOut = true;} catch (InterruptedException retry) {timedOut = false;}}}

3、Future和Callable

Future 和 Callable 用于表示异步任务的结果。

主要类：

Callable<V>，类似于 Runnable，但可以返回结果或抛出异常

Future<V>，表示异步计算的结果，提供了检查任务是否完成、获取结果、取消任务等方法。

FutureTask<V>，Future接口的实现类，可以包装 Callable 或 Runnable

详细可以查看上一篇文章：并发编程Future和Callback使用

4、并发集合

java.util.concurrent 提供了线程安全的集合类，适用于多线程环境。

4.1、ConcurrentHashMap

ConcurrentHashMap主要是在多线程情况下提升了性能。

详细查看另一篇内容：HashMap到ConcurrentHashMap原理

5、同步工具

提供了一些高级同步工具，用于控制线程之间的协作。

• CountDownLatch：允许一个或多个线程等待其他线程完成操作。
• CyclicBarrier：让一组线程互相等待，直到所有线程都到达某个屏障点。
• Semaphore：控制同时访问某个资源的线程数量。
• Phaser：更灵活的同步工具，支持分阶段的线程同步。

import java.util.concurrent.CountDownLatch;public class CountDownLatchExample {public static void main(String[] args) throws InterruptedException {CountDownLatch latch = new CountDownLatch(3);for (int i = 0; i < 3; i++) {new Thread(() -> {System.out.println("子线程执行: " + Thread.currentThread().getName());latch.countDown();}).start();}latch.await(); // 主线程等待所有子线程完成System.out.println("所有子线程执行完毕！");}
}

 6、原子变量

提供了一组原子操作类，用于实现无锁的线程安全编程。

• AtomicInteger：支持原子操作的整数。
• AtomicLong：支持原子操作的长整数。
• AtomicReference<V>：支持原子操作的引用类型。

import java.util.concurrent.atomic.AtomicInteger;public class AtomicExample {public static void main(String[] args) {AtomicInteger counter = new AtomicInteger(0);counter.incrementAndGet(); // 原子递增System.out.println("Counter: " + counter.get());}
}

7、锁（Locks）

提供了更灵活的锁机制，替代传统的 synchronized 关键字。

• ReentrantLock：可重入锁，支持公平锁和非公平锁。
• ReadWriteLock：读写锁，允许多个读线程同时访问，但写线程独占。实现类ReentrantReadWriteLock

import java.util.concurrent.locks.ReentrantLock;public class LockExample {public static void main(String[] args) {ReentrantLock lock = new ReentrantLock();lock.lock(); // 加锁try {System.out.println("锁保护的代码块");} finally {lock.unlock(); // 释放锁}}
}

8、定时任务

支持定时或周期性任务的执行。

ScheduledExecutorService支持定时任务的线程池接口，实现类ScheduledThreadPoolExecutor

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;public class ScheduledTaskExample {public static void main(String[] args) {ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);scheduler.scheduleAtFixedRate(() -> {System.out.println("定时任务执行: " + System.currentTimeMillis());}, 0, 1, TimeUnit.SECONDS); // 初始延迟 0 秒，每隔 1 秒执行一次}
}

9、Fork/Join 框架 

用于并行执行任务的框架，适用于分治算法。

• ForkJoinPool：用于执行 ForkJoinTask 的线程池。
• RecursiveTask<V>：用于有返回值的任务。
• RecursiveAction：用于无返回值的任务。

import java.util.concurrent.RecursiveTask;
import java.util.concurrent.ForkJoinPool;public class ForkJoinExample {public static void main(String[] args) {ForkJoinPool pool = new ForkJoinPool();int result = pool.invoke(new FibonacciTask(10));System.out.println("Fibonacci(10) = " + result);}static class FibonacciTask extends RecursiveTask<Integer> {private final int n;FibonacciTask(int n) {this.n = n;}@Overrideprotected Integer compute() {if (n <= 1) return n;FibonacciTask task1 = new FibonacciTask(n - 1);task1.fork();FibonacciTask task2 = new FibonacciTask(n - 2);return task2.compute() + task1.join();}}
}