java.util.concurrent 包介绍

文章目录

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 实例。

上一篇:大数据之路week04--day06(I/O流阶段一 之异常)


下一篇:day06 is == 编码 解码