文章目录

• 1.介绍
• 2.主要组件
• • 2.1. Executor
  • 2.2. ExecutorService
  • 2.3. ScheduledExecutorService
  • 2.4. Future
  • 2.5. CountDownLatch
  • 2.6. CyclicBarrier
  • 2.7. Semaphore
  • 2.8. ThreadFactory
  • 2.9.BlockingQueue
  • 2.10. DelayQueue
  • 2.11. Locks
  • 2.12. Phaser

1.介绍

java.util.concurrent 包提供了用于创建并发应用程序的工具。

2.主要组件

java.util.concurrent 包含太多的特性。 在本篇中，主要关注此包中一些最有用的实用程序，

例如：

• Executor
• ExecutorService
• ScheduledExecutorService
• Future
• CountDownLatch
• CyclicBarrier
• Semaphore
• ThreadFactory
• BlockingQueue
• DelayQueue
• Locks
• Phaser

2.1. Executor

Executor是一个表示提供任务对象的接口。

如果任务要在新线程或当前线程上运行，这取决于特定的实现（从哪里启动调用）。 因此，使用此接口，可以将任务执行流程与实际任务执行机制解耦。

这里需要注意的一点是，Executor 并没有严格要求任务执行是异步的。 在最简单的情况下，执行者可以在调用线程中立即调用提交的任务。

创建一个调用程序实例：

public class Invoker implements Executor {
    @Override
    public void execute(Runnable r) {
        r.run();
    }
}

调用程序来执行任务。

public void execute() {
    Executor executor = new Invoker();
    executor.execute(() -> System.out.println("execute task ..."));
}

这里需要注意的是，如果executor不能接受任务执行，就会抛出RejectedExecutionException。

2.2. ExecutorService

ExecutorService 是一个完整的异步处理解决方案。 它管理内存队列并根据线程可用性安排提交的任务。

要使用 ExecutorService，需要创建一个 Runnable 类。

public class Task implements Runnable {
    @Override
    public void run() {
        // 任务细节
    }
}

现在创建 ExecutorService 实例并分配这个任务。 在创建时，需要指定线程池大小。

ExecutorService executor = Executors.newFixedThreadPool(10);

如果想创建一个单线程的ExecutorService实例，可以使用newSingleThreadExecutor(ThreadFactory threadFactory)来创建该实例。

一旦创建了执行器，就可以使用它来提交任务。

public void execute() { 
    executor.submit(new Task()); 
}

还可以在提交任务时创建 Runnable 实例。

executor.submit(() -> {
    new Task();
});

它还带有两种开箱即用的执行终止方法。 第一个是shutdown(); 它等待所有提交的任务完成执行。 另一种方法是shutdownNow()，它立即终止所有挂起/正在执行的任务。

还有一个方法 awaitTermination(long timeout, TimeUnit unit) 强制阻塞直到所有任务在触发关闭事件或执行超时后完成执行，或者执行线程本身被中断，

try {
    executor.awaitTermination( 20l, TimeUnit.NANOSECONDS );
} catch (InterruptedException e) {
    e.printStackTrace();
}

2.3. ScheduledExecutorService

ScheduledExecutorService 是一个类似于 ExecutorService 的接口，但它可以定期执行任务。

Executor 和 ExecutorService 的方法是现场调度的，不会引入任何人为的延迟。 零或任何负值表示需要立即执行请求。

可以同时使用 Runnable 和 Callable 接口来定义任务。

  public static void execute() throws Exception {
        ScheduledExecutorService executorService
                = Executors.newSingleThreadScheduledExecutor();

        Future<String> future = executorService.schedule(() -> "Hello world", 20, TimeUnit.SECONDS);

        ScheduledFuture<?> scheduledFuture = executorService.schedule(() -> {
            System.out.println("延迟5s执行");
        }, 5, TimeUnit.SECONDS);

        while (future.isDone() && !future.isCancelled()) {
            Thread.sleep(2000);
        }
        System.out.println(future.get());

        executorService.shutdown();
    }

ScheduledExecutorService 还可以在给定的固定延迟后执行任务：

executorService.scheduleAtFixedRate(() -> {
    // ...
}, 1, 10, TimeUnit.SECONDS);

executorService.scheduleWithFixedDelay(() -> {
    // ...
}, 1, 10, TimeUnit.SECONDS);

scheduleAtFixedRate( Runnable command, long initialDelay, long period, TimeUnit unit ) 方法创建并执行一个周期性动作，该动作首先在提供的初始延迟后调用，然后在给定的时间段内调用，直到服务实例关闭。

scheduleWithFixedDelay( Runnable command, long initialDelay, long delay, TimeUnit unit ) 方法创建并执行一个周期性动作，该动作在提供的初始延迟后首先被调用，并在执行的终止和调用之间重复给定的延迟 下一个。

2.4. Future

Future 用于表示异步操作的结果。 它带有用于检查异步操作是否完成、获取计算结果等的方法。

更重要的是，cancel(boolean mayInterruptIfRunning) API 取消操作并释放正在执行的线程。 如果 mayInterruptIfRunning 的值为 true，则执行任务的线程将立即终止。

否则，将允许完成正在进行的任务。

public void invoke() {
    ExecutorService executorService = Executors.newFixedThreadPool(10);

    Future<String> future = executorService.submit(() -> {
        // 执行某些任务
        Thread.sleep(10000l);
        return "Hello world";
    });
}

可以使用以下代码片段来检查Future的结果是否准备就绪，如果计算完成则获取数据：

if (future.isDone() && !future.isCancelled()) {
    try {
        str = future.get();
    } catch (InterruptedException | ExecutionException e) {
        e.printStackTrace();
    }
}

还可以为给定操作指定超时。 如果任务花费的时间超过这个时间，则会抛出 TimeoutException：

try {
    future.get(10, TimeUnit.SECONDS);
} catch (InterruptedException | ExecutionException | TimeoutException e) {
    e.printStackTrace();
}

2.5. CountDownLatch

CountDownLatch（在 JDK 5 中引入）是一个实用程序类，它阻塞一组线程直到某些操作完成。

一个 CountDownLatch 用一个计数器（整数类型）初始化； 该计数器随着依赖线程完成执行而递减。 但是一旦计数器达到零，其他线程就会被释放。

2.6. CyclicBarrier

CyclicBarrier 的工作原理与 CountDownLatch 几乎相同，只是可以重用它。 与 CountDownLatch 不同，它允许多个线程在调用最终任务之前使用 await() 方法（称为屏障条件）相互等待。

public class CyclicBarrierTest {
    public static void main(String[] args) {
        start();
    }

    public static void start() {

        CyclicBarrier cyclicBarrier = new CyclicBarrier(3, () -> {
            // ...
            System.out.println("All previous tasks are completed");
        });

        Thread t1 = new Thread(new Task(cyclicBarrier), "T1");
        Thread t2 = new Thread(new Task(cyclicBarrier), "T2");
        Thread t3 = new Thread(new Task(cyclicBarrier), "T3");

        if (!cyclicBarrier.isBroken()) {
            t1.start();
            t2.start();
            t3.start();
        }

    }
}

class Task implements Runnable {

    private CyclicBarrier barrier;

    public Task(CyclicBarrier barrier) {
        this.barrier = barrier;
    }

    @Override
    public void run() {
        try {
            System.out.println(Thread.currentThread().getName() +
                    " is waiting");
            barrier.await();
            System.out.println(Thread.currentThread().getName() +
                    " is released");
        } catch (InterruptedException | BrokenBarrierException e) {
            e.printStackTrace();
        }
    }

}

输出

T1 is waiting
T2 is waiting
T3 is waiting
All previous tasks are completed
T3 is released
T1 is released
T2 is released

isBroken() 方法检查在执行期间是否有任何线程被中断。 在执行实际过程之前，我们应该始终执行此检查。

2.7. Semaphore

Semaphore(信号量)用于阻止线程级访问物理或逻辑资源的某些部分。 信号量包含一组许可； 每当一个线程试图进入临界区时，它需要检查信号量是否有许可可用。

如果许可不可用（通过 tryAcquire()），则不允许线程跳入临界区； 但是，如果许可可用，则准予访问，并且许可计数器减少。

一旦执行线程释放临界区，许可计数器再次增加（由 release() 方法完成）。

可以使用 tryAcquire(long timeout, TimeUnit unit) 方法指定获取访问的超时时间。

还可以检查可用许可的数量或等待获取信号量的线程数量。

public class SemaphoreTest {
    static Semaphore semaphore = new Semaphore(10);

    public static void main(String[] args) throws InterruptedException {
        for (int i = 0; i < 20; i++) {
            Thread.sleep(500);
            Thread thread = new Thread(() -> {
                try {
                    execute();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
            thread.setName("Thread[" + i + "]");
            thread.start();

        }
    }


    public static void execute() throws InterruptedException {

        System.out.println("可用的信号量 : " + semaphore.availablePermits());
        System.out.println("等待获取的线程数: " +
                semaphore.getQueueLength());

        if (semaphore.tryAcquire()) {
            try {
                System.out.println(Thread.currentThread().getName() + "：正在执行任务...");
                Thread.sleep(new Random().nextInt(10) * 1000);
            } finally {
                semaphore.release();
            }
        } else {
            System.out.println(Thread.currentThread().getName() + "：没有执行任务权限");
        }

    }
}

可以使用信号量实现类似互斥锁的数据结构。

2.8. ThreadFactory

ThreadFactory 充当线程（不存在）池，可按需创建新线程。 它消除了实现高效线程创建机制所需的大量样板代码。

public class CustomThreadFactory  implements ThreadFactory {
    private int threadId;
    private String name;

    public CustomThreadFactory(String name) {
        threadId = 1;
        this.name = name;
    }

    @Override
    public Thread newThread(Runnable r) {
        Thread t = new Thread(r, name + "-Thread_" + threadId);
        System.out.println("新建线程id为 : " + threadId +
                " name为 : " + t.getName());
        threadId++;
        return t;
    }

    public static void main(String[] args) {
        CustomThreadFactory factory = new CustomThreadFactory(
                "CustomThreadFactory");
        for (int i = 0; i < 10; i++) {
            int finalI = i;
            Runnable runnable = () -> System.out.println(finalI + " runable thread");
            Thread t = factory.newThread(runnable);
            t.start();
        }
    }
}

2.9.BlockingQueue

在异步编程中，最常见的集成模式之一是生产者-消费者模式。 java.util.concurrent 包带有一个称为 BlockingQueue 的数据结构——它在这些异步场景中非常有用。

2.10. DelayQueue

DelayQueue 是一个无限大小的元素阻塞队列，其中一个元素只有在它的到期时间（称为用户定义的延迟）完成时才能被拉取。 因此，最顶部的元素（头部）将具有最大的延迟量，并且将最后轮询。

2.11. Locks

Lock 是一个实用的类，用于阻止其他线程访问特定代码段，除了当前正在执行它的线程。

Lock 和 Synchronized 块之间的主要区别在于，同步块完全包含在方法中； 但是，可以在不同的方法中使用 Lock API 的 lock() 和 unlock() 操作。

2.12. Phaser

Phaser 是一个比 CyclicBarrier 和 CountDownLatch 更灵活的解决方案——用作可重用的屏障，动态数量的线程在继续执行之前需要等待。 可以协调执行的多个阶段，为每个程序阶段重用一个 Phaser 实例。