您的位置 首页 java

Java高并发系列-先从上帝的角度看一下并发包JUC

智云科技发布于 2023年5月25日评论关闭阅读(70)

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( () -> {
        // task to be performed
    });
}

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

2.2. ExecutorService

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

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

 public class Task implements Runnable {
    @Override
    public void run() {
        // task details
    }
}

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

 ExecutorService executor = Executors.newFixedThreadPool(10);

如果我们想创建一个单线程的ExecutorService实例，我们可以使用newSingleThreadExecutor( thread Factory 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 void execute() {
    ScheduledExecutorService executorService
      = Executors.newSingleThreadScheduledExecutor();

    Future<String> future = executorService.schedule(() -> {
        // ...
        return "Hello world";
    }, 1, TimeUnit.SECONDS);

    ScheduledFuture<?> scheduledFuture = executorService.schedule(() -> {
        // ...
    }, 1, TimeUnit.SECONDS);

    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（可运行命令，长在initialDelay，长的延迟，TIMEUNIT单元）方法与所述执行的一个的终止和的调用之间的给定的延迟创建并执行所提供的初始延迟后首先调用的周期性动作，并重复地下一个。

2.4. Future

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

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

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

我们可以使用下面的代码片段来创建一个未来的实例： 

 public void invoke() {
    ExecutorService executorService = Executors.newFixed Thread Pool(10);

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

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

 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()方法（称为屏障条件）相互等待。

我们需要创建一个Runnable任务实例来启动屏障条件： 

 public class Task implements Runnable {

    private CyclicBarrier barrier;

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

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

}

现在我们可以调用一些线程来争夺屏障条件： 

 public void start() {

    CyclicBarrier cyclicBarrier = new CyclicBarrier(3, () -> {
        // ...
        LOG.info("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(); 
    }
}

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

2.7. semaphore

信号量 被用于阻挡到物理或逻辑资源的某些部分螺纹级别的访问，每当一个线程试图进入临界区时，它需要检查信号量是否有许可可用。

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

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

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

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

以下代码片段可用于实现信号量： 

 static  Semaphore  semaphore = new Semaphore(10);

public void execute() throws InterruptedException {

    LOG.info("Available permit : " + semaphore.availablePermits());
    LOG.info("Number of threads waiting to acquire: " + 
      semaphore.getQueueLength());

    if (semaphore.tryAcquire()) {
        try {
            // ...
        }
        finally {
            semaphore.release();
        }
    }

}

我们可以使用Semaphore实现类似互斥锁的数据结构。可以在此处找到有关此的更多详细信息。

2.8. ThreadFactory

顾名思义，ThreadFactory充当线程池，可按需创建新线程。它消除了实现高效线程创建机制所需的大量样板代码。

我们可以定义一个ThreadFactory： 

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

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

    @Override
    public Thread newThread(Runnable r) {
        Thread t = new Thread(r, name + "-Thread_" + threadId);
        LOG.info("created new thread with id : " + threadId +
            " and name : " + t.getName());
        threadId++;
        return t;
    }
}

我们可以使用这个newThread(Runnable r)方法在运行时创建一个新线程： 

 BaeldungThreadFactory factory = new BaeldungThreadFactory( 
    "BaeldungThreadFactory");
for (int i = 0; i < 10; i++) { 
    Thread t = factory.newThread(new Task());
    t.start(); 
}

2.9. B Lock ingQueue

在异步编程中，最常见的集成模式之一是生产者-消费者模式。该的java.util.concurrent包带有一个数据结构所知道的BlockingQueue的-它可以在这些异步情况下非常有用的。

2.10. DelayQueue

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

2.11. Locks

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

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

2.12. Phaser

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

3. 结论

在这篇高级概述文章中，我们重点介绍了java.util.concurrent包中可用的不同实用程序，后面的文章，我们会详细的进行分析。

2021-06-08 北京 刘高飞

标签:
打赏 点赞(1)

文章来源：智云一二三科技

文章标题：Java高并发系列-先从上帝的角度看一下并发包JUC

文章地址：https://www.zhihuclub.com/184144.shtml

关于作者: 智云科技

相关文章

热门文章

1分享新浪图床上传接口源码

点赞(946)阅读(61,173)

2PHP简单实现路由Route功能

点赞(1.15K)阅读(35,972)

3Tideways、xhprof 和 xhgui 打造 PHP 非侵入式监控平台

点赞(377)阅读(4,302)

4centos系统如何查看是否安装了mysql

点赞(544)阅读(3,794)

5curl 工具简述

点赞(521)阅读(2,983)
网站地图