文章目錄

• • 主要的組件
  • Executor
  • ExecutorService
  • ScheduledExecutorService
  • Future
  • CountDownLatch
  • CyclicBarrier
  • Semaphore
  • ThreadFactory

java.util.concurrent簡介

java.util.concurrent包提供了很多有用的類，方便我們進行并發(fā)程序的開發(fā)。本文將會做一個總體的簡單介紹。

主要的組件

java.util.concurrent包含了很多內(nèi)容， 本文將會挑選其中常用的一些類來進行大概的說明：

• Executor
• ExecutorService
• ScheduledExecutorService
• Future
• CountDownLatch
• CyclicBarrier
• Semaphore
• ThreadFactory

Executor

Executor是一個接口，它定義了一個execute方法，這個方法接收一個Runnable，并在其中調(diào)用Runnable的run方法。

我們看一個Executor的實現(xiàn)：

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

現(xiàn)在我們可以直接調(diào)用該類中的方法：

public void execute() {Executor executor = new Invoker();executor.execute( () -> {log.info("{}", Thread.currentThread().toString());});}

注意，Executor并不一定要求執(zhí)行的任務是異步的。

ExecutorService

如果我們真正的需要使用多線程的話，那么就需要用到ExecutorService了。

ExecutorService管理了一個內(nèi)存的隊列，并定時提交可用的線程。

我們首先定義一個Runnable類：

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

我們可以通過Executors來方便的創(chuàng)建ExecutorService：

ExecutorService executor = Executors.newFixedThreadPool(10);

上面創(chuàng)建了一個ThreadPool， 我們也可以創(chuàng)建單線程的ExecutorService：

ExecutorService executor =Executors.newSingleThreadExecutor();

我們這樣提交task：

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

因為ExecutorService維持了一個隊列，所以它不會自動關(guān)閉， 我們需要調(diào)用executor.shutdown() 或者executor.shutdownNow()來關(guān)閉它。

如果想要判斷ExecutorService中的線程在收到shutdown請求后是否全部執(zhí)行完畢，可以調(diào)用如下的方法：

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

ScheduledExecutorService

ScheduledExecutorService和ExecutorService很類似，但是它可以周期性的執(zhí)行任務。

我們這樣創(chuàng)建ScheduledExecutorService：

ScheduledExecutorService executorService= Executors.newSingleThreadScheduledExecutor();

executorService的schedule方法，可以傳入Runnable也可以傳入Callable：

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

還有兩個比較相近的方法：

scheduleAtFixedRate( Runnable command, long initialDelay, long period, TimeUnit unit )scheduleWithFixedDelay( Runnable command, long initialDelay, long delay, TimeUnit unit )

兩者的區(qū)別是前者的period是以任務開始時間來計算的，后者是以任務結(jié)束時間來計算。

Future

Future用來獲取異步執(zhí)行的結(jié)果?？梢哉{(diào)用cancel(boolean mayInterruptIfRunning) 方法來取消線程的執(zhí)行。

我們看下怎么得到一個Future對象：

public void invoke() {ExecutorService executorService = Executors.newFixedThreadPool(10);Future<String> future = executorService.submit(() -> {// ...Thread.sleep(10000l);return "Hello world";}); }

我們看下怎么獲取Future的結(jié)果：

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

future還可以接受一個時間參數(shù)，超過指定的時間，將會報TimeoutException。

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

CountDownLatch

CountDownLatch是一個并發(fā)中很有用的類，CountDownLatch會初始化一個counter，通過這個counter變量，來控制資源的訪問。我們會在后面的文章詳細介紹。

CyclicBarrier

CyclicBarrier和CountDownLatch很類似。CyclicBarrier主要用于多個線程互相等待的情況，可以通過調(diào)用await() 方法等待，知道達到要等的數(shù)量。

public class Task implements Runnable {private CyclicBarrier barrier;public Task(CyclicBarrier barrier) {this.barrier = barrier;}@Overridepublic 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(); } }

Semaphore

Semaphore包含了一定數(shù)量的許可證，通過獲取許可證，從而獲得對資源的訪問權(quán)限。通過 tryAcquire()來獲取許可，如果獲取成功，許可證的數(shù)量將會減少。

一旦線程release()許可，許可的數(shù)量將會增加。

我們看下怎么使用：

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();}}}

ThreadFactory

ThreadFactory可以很方便的用來創(chuàng)建線程：

public class ThreadFactoryUsage implements ThreadFactory {private int threadId;private String name;public ThreadFactoryUsage(String name) {threadId = 1;this.name = name;}@Overridepublic 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;} }

本文的例子可以參考https://github.com/ddean2009/learn-java-concurrency/tree/master/concurrent-overview

更多精彩內(nèi)容且看：

• 區(qū)塊鏈從入門到放棄系列教程-涵蓋密碼學,超級賬本,以太坊,Libra,比特幣等持續(xù)更新
• Spring Boot 2.X系列教程:七天從無到有掌握Spring Boot-持續(xù)更新
• Spring 5.X系列教程:滿足你對Spring5的一切想象-持續(xù)更新
• java程序員從小工到專家成神之路（2020版）-持續(xù)更新中,附詳細文章教程

更多教程請參考 flydean的博客

總結(jié)

以上是生活随笔為你收集整理的java.util.concurrent简介的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。