并发工具类及线程池

    技术2025-12-27  10

    同步辅助类


    CyclicBarrier

    是一个同步辅助工具类,它允许一组线程互相等待直到所有的线程达到一个公共屏障点。在程序中由固定的线程数量,这些线程有时必须等待彼此。

    应用场景:

    分组计算然后统计。 理解:人满发车(20)

    底层设计: ReentrantLock 和 condition

    final ReentrantLock lock = this.lock; lock.lock();

    CountDownLatch

    允许一个或多个线程一直等待,直到其他线程的操作执行完后在执行

    应用场景

    分类统计,合并结果

    区别: CountDownLatch :允许一个或多个线程一直等待,直到其他线程的操作执行完后在执行 cyclicBarrier:多个线程相互等待,直到达到同一个不同点,在继续执行。


    Semaphore

    信号量:用来控制同时访问特定资源的线程数量,通过协调各个线程以保证合理的使用公共资源。 创建Semaphore对象实例的时候传人信号量的值。 acquire();获取信号量。 release();释放信号量。 # 同步使用。


    Exchanger

    用于进行两个线程之间数据的交换。它提供一个同步点,在同步点上,两个线程可以彼此交换数据。


    线程池

    开发者的困境:

    线程管理: 线程的创建、启动、销毁。 线程复用:如何减少频繁重复创建线程的开销 弹性伸缩: 拒绝策略: AbortPolicy:终止,丢弃任务,直接抛出异常,调用者将会获得异常 DisCardPolicy:抛弃,线程池将会悄悄的丢弃任务而不被调用者知道 CallerRunsPolicy:调用者运行,不抛弃任务也不抛出异常,而是叫任务退回给调用者 DisCardOldestPolicy:抛弃最旧的 异常处理: 任务分配:优先级的策略…

    线程池参数详解:

    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; } corePoolSize(核心线程数最大值)

    该线程池中核心线程数最大值 。线程池新建线程的时候,如果当前线程总数小于corePoolSize,则新建的是核心线程,如果超过corePoolSize,则新建的是非核心线程。核心线程默认情况下会一直存活在线程池中,即使这个核心线程啥也不干(闲置状态)。

    maximumPoolSize(线程总数最大值)

    该线程池中线程总数最大值,线程总数 = 核心线程数 + 非核心线程数。

    keepAliveTime(非核心线程闲置超时时长)

    该线程池中非核心线程闲置超时时长。一个非核心线程,如果不干活(闲置状态)的时长超过这个参数所设定的时长,就会被销毁掉。

    unit (时间单位)

    时间单位。为 keepAliveTime 指定时间单位。

    workQueue(任务队列)

    该线程池中的任务队列。维护着等待执行的Runnable对象,当所有的核心线程都在干活时,新添加的任务会被添加到这个队列中等待处理,如果队列满了,则新建非核心线程执行任务。

    threadFactory(线程工厂)

    创建线程的工厂类。可以通过指定线程工厂为每个创建出来的线程设置更有意义的名字,如果出现并发问题,也方便查找问题原因。

    handler(拒绝策略)

    执行拒绝策略的对象。当线程池的阻塞队列已满和指定的线程都已经开启,说明当前线程池已经处于饱和状态了,那么就需要采用一种策略来处理这种情况

    线程池的作用:

    就是限制在系统中可以执行的线程的数量。根据系统环境的情况,自动或手动设置线程数量,以达到运行的最佳效果。 用线程池控制线程的数量,多余的线程必须排队等候,一个任务执行完毕,再从队列中取最前面的任务开始执行。如果队列中没有等待的线程,线程池的资源就处于等待的状态。当一个新任务需要运行时,如果线程池中有等待的工作线程,就可以运行,否则进入等待队列。

    Executor接口

    /* * @since 1.5 * @author Doug Lea */ public interface Executor { void execute(Runnable command); }

    Executor接口只有一个方法,它的子类ExecutorService提供了更加丰富的功能

    ExecutorService

    public interface ExecutorService extends Executor { /** *关闭线程池,已提交的任务继续执行,不接受继续提交新任务 */ void shutdown(); /** * 关闭线程池,尝试停止正在执行的所有任务,不接受继续提交新任务 * 它和前面的方法相比,加了一个单词now,区别在于他会停止当前正在进行的任务 */ List<Runnable> shutdownNow(); /** * 线程池是否已经关闭 */ boolean isShutdown(); /** * 如果调用了 shutdown() 或 shutdownNow() 方法后,所有任务结束了,那么返回true * 这个方法必须在调用shutdown或shutdownNow方法之后调用才会返回true */ boolean isTerminated(); /** * 等待所有任务完成,并设置超时时间 我们这么理解,实际应用中是,先调用 shutdown 或 shutdownNow, * 然后再调这个方法等待所有的线程真正地完成,返回值意味着有没有超时 */ boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException; /** * 提交一个 Callable 任务 */ <T> Future<T> submit(Callable<T> task); /** * 提交一个 Runnable 任务,第二个参数将会放到 Future 中,作为返回值 * 因为 Runnable 的 run 方法本身并不返回任何东西 */ <T> Future<T> submit(Runnable task, T result); /** * 提交一个Runnable任务 */ Future<?> submit(Runnable task); /** * 执行所有任务,返回 Future 类型的一个 list */ <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException; /** * 也是执行所有任务,但是这里设置了超时时间 */ <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException; /** * 只有其中的一个任务结束了,就可以返回,返回执行完的那个任务的结果 */ <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException; /** * 同上一个方法,只有其中的一个任务结束了,就可以返回,返回执行完的那个任务的结果, * 不过这个带超时,超过指定的时间,抛出 TimeoutException 异常 */ <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException; }

    常见创建线程池的方式:

    FixedThreadPool:创建固定长度的线程池,每次提交任务创建一个线程,直到达到线程池的最大数量,线程池的大小不再变化 public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()); }

    参数详解:

    FixedThreadPool的corePoolSize和maxiumPoolSize都被设置为创建FixedThreadPool时指定的参数nThreads。0L则表示当线程池中的线程数量操作核心线程的数量时,多余的线程将被立即停止最后一个参数表示FixedThreadPool使用了无界队列LinkedBlockingQueue作为线程池的工作队列,由于是无界的,当线程池的线程数达到corePoolSize后,新任务将在无界队列中等待,因此线程池的线程数量不会超过corePoolSize,同时maxiumPoolSize也就变成了一个无效的参数,并且运行中的线程池并不会拒绝任务

    执行流程

    如果当前工作中的线程数量少于corePool的数量,就创建新的线程来执行任务。当线程池的工作中的线程数量达到了corePool,则将任务加入LinkedBlockingQueue。线程执行完1中的任务后会从队列中去任务。 注意LinkedBlockingQueue是无界队列,所以可以一直添加新任务到线程池。 SingleThreadExecutor:是使用单个worker线程的Executor。特点是使用单个工作线程执行任务。

    类似于FixedThreadPool,只是用一个线程去执行任务。

    CachedThreadPool

    CachedThreadPool是一个”无限“容量的线程池,它会根据需要创建新线程。特点是可以根据需要来创建新的线程执行任务,没有特定的corePool

    public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>()); }

    CachedThreadPool的corePoolSize被设置为0,即corePool为空;maximumPoolSize被设置为Integer.MAX_VALUE,即maximum是无界的。这里keepAliveTime设置为60秒,意味着空闲的线程最 多可以等待任务60秒,否则将被回收。 CachedThreadPool使用没有容量的SynchronousQueue作为主线程池的工作队列,它是一个没有容量 的阻塞队列。每个插入操作必须等待另一个线程的对应移除操作。这意味着,如果主线程提交任务的速 度高于线程池中处理任务的速度时,CachedThreadPool会不断创建新线程。极端情况下,CachedThreadPool会因为创建过多线程而耗尽CPU资源。

    newScheduledThreadPool

    创建一个定长线程池,支持定时及周期性任务执行。

    FutureTask

    Future接口和实现Future接口的FutureTask类,代表异步计算的结果。一个可取消的异步计算。 FutureTask提供了对Future的基本实现,可以调用方法去开始和取消一个计算,可以查询计算是否完成并且获取计算结果。


    Processed: 0.010, SQL: 10