原標(biāo)題：Java并發(fā)編程之CyclicBarrier和線程池的使用

下面我們來講述一下線程池和CyclicBarrier的使用和對(duì)比。

一、場(chǎng)景描述

有四個(gè)游戲玩愛好者玩游戲，游戲中有三個(gè)關(guān)卡，每一個(gè)關(guān)卡必須讓所有玩家到達(dá)后才能允許通過。其實(shí)這個(gè)場(chǎng)景里的玩家中如果有玩家B先到了第一個(gè)關(guān)卡，他必須要等到其他剩余玩家都到達(dá)第一個(gè)關(guān)卡時(shí)才能通過，這也說明了線程之間需要相互等待。這和CountDownLatch的應(yīng)用場(chǎng)景有區(qū)別，CountDownLatch里的線程是到了運(yùn)行的目標(biāo)后繼續(xù)干自己的事情，而不需要管其他玩家，而這里的線程就不同了，必須等待其他玩家。

二、CyclicBarrier介紹

CyclicBarrier 的字面意思是可循環(huán)使用(Cyclic)的屏障(Barrier)。它要做的事情是，讓一組線程到達(dá)一個(gè)屏障(也可以叫同步點(diǎn))時(shí)被阻塞，直到最后一個(gè)線程到達(dá)屏障時(shí)，屏障才會(huì)開門，所有被屏障攔截的線程才會(huì)繼續(xù)干活。CyclicBarrier默認(rèn)的構(gòu)造方法是CyclicBarrier(int parties)，其參數(shù)表示屏障攔截的線程數(shù)量，每個(gè)線程調(diào)用await方法告訴CyclicBarrier我已經(jīng)到達(dá)了屏障，然后當(dāng)前線程被阻塞。

CyclicBarrier類有兩個(gè)常用的構(gòu)造方法：

1. CyclicBarrier(int parties)

這里的parties也是一個(gè)計(jì)數(shù)器，例如，初始化時(shí)parties里的計(jì)數(shù)是3，于是擁有該CyclicBarrier對(duì)象的線程當(dāng)parties的計(jì)數(shù)為3時(shí)就喚醒，注：這里parties里的計(jì)數(shù)在運(yùn)行時(shí)當(dāng)調(diào)用CyclicBarrier:await()時(shí),計(jì)數(shù)就加1，一直加到初始的值

2. CyclicBarrier(int parties, Runnable barrierAction)

這里的parties與上一個(gè)構(gòu)造方法的解釋是一樣的，這里需要解釋的是第二個(gè)入?yún)?Runnable barrierAction),這個(gè)參數(shù)是一個(gè)實(shí)現(xiàn)Runnable接口的類的對(duì)象，也就是說當(dāng)parties加到初始值時(shí)就出發(fā)barrierAction的內(nèi)容。

代碼示例 package com.itmyhome;import java.util.concurrent.BrokenBarrierException;import java.util.concurrent.CyclicBarrier;/**

輸出結(jié)果: 玩家0正在玩第一關(guān)...玩家3正在玩第一關(guān)...玩家2正在玩第一關(guān)...玩家1正在玩第一關(guān)...所有玩家進(jìn)入第二關(guān)！

CyclicBarrier和CountDownLatch的區(qū)別

CountDownLatch: 一個(gè)線程(或者多個(gè))， 等待另外N個(gè)線程完成某個(gè)事情之后才能執(zhí)行。

CyclicBarrier: N個(gè)線程相互等待，任何一個(gè)線程完成之前，所有的線程都必須等待。

CountDownLatch的計(jì)數(shù)器只能使用一次。而CyclicBarrier的計(jì)數(shù)器可以使用reset() 方法重置。所以CyclicBarrier能處理更為復(fù)雜的業(yè)務(wù)場(chǎng)景，比如如果計(jì)算發(fā)生錯(cuò)誤，可以重置計(jì)數(shù)器，并讓線程們重新執(zhí)行一次。

CountDownLatch：減計(jì)數(shù)方式，CyclicBarrier：加計(jì)數(shù)方式

三、線程池使用

在上例中，用到了線程池，采用了Executors提供的靜態(tài)方法初始化線程池。

ExecutorService executorService = Executors.newCachedThreadPool();

newCachedThreadPool()方法實(shí)現(xiàn)如下，即初始線程池沒有創(chuàng)建線程，只有在有新任務(wù)時(shí)才會(huì)創(chuàng)建線程去執(zhí)行任務(wù)，空閑線程等待時(shí)間60秒。

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

}

ExecutorService與ThreadPoolExecutor是什么關(guān)系，有什么差異，因?yàn)橥ㄟ^ThreadPoolExecutor也能夠?qū)崿F(xiàn)線程池。

public class Test { public static void main(String[] args) {

ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10, 200, TimeUnit.MILLISECONDS, new ArrayBlockingQueue(5)); for(int i=0;i<15;i++){

MyTask myTask = new MyTask(i);

executor.execute(myTask);

}

executor.shutdown();

}

}

首先來看看ThreadPoolExecutor的execute函數(shù)，這個(gè)函數(shù)返回void：

void execute(Runnable command)//Executes the given task sometime in the future.

然后再來看看ExecutorService的submit函數(shù)，這個(gè)函數(shù)返回Future，即有返回值，這是兩者的一個(gè)差異之處。

Future> submit(Runnable task)//Submits a Runnable task for execution and returns a Future representing that task.

接下來，通過樹狀圖來看看線程池相關(guān)類間的關(guān)系，可以查閱源碼看之間的關(guān)系：

Executor是頂層接口，僅提供execute方法。

ExecutorService接口繼承了Executor接口，豐富了接口函數(shù)。

AbstractExecutorService抽象類實(shí)現(xiàn)了ExecutorService接口。

ThreadPoolExecutor類繼承了AbstractExecutorService類。

部分源碼

Executor接口

public interface Executor { void execute(Runnable command);

}

ExecutorService接口

public interface ExecutorService extends Executor { void shutdown(); boolean isShutdown(); boolean isTerminated(); boolean awaitTermination(long timeout, TimeUnit unit)

throws InterruptedException;

Future submit(Callable task);

Future submit(Runnable task, T result);

Future> submit(Runnable task);

List> invokeAll(Collection extends Callable> tasks) throws InterruptedException;

List> invokeAll(Collection extends Callable> tasks, long timeout, TimeUnit unit) throws InterruptedException;

T invokeAny(Collection extends Callable> tasks)

throws InterruptedException, ExecutionException;

T invokeAny(Collection extends Callable> tasks, long timeout, TimeUnit unit)

throws InterruptedException, ExecutionException, TimeoutException;

}

AbstractExecutorService抽象類

public abstract class AbstractExecutorService implements ExecutorService { protected RunnableFuture newTaskFor(Runnable runnable, T value) { }; protected RunnableFuture newTaskFor(Callable callable) { }; public Future> submit(Runnable task) {}; public Future submit(Runnable task, T result) { }; public Future submit(Callable task) { }; private T doInvokeAny(Collection extends Callable> tasks, boolean timed, long nanos)

throws InterruptedException, ExecutionException, TimeoutException {

}; public T invokeAny(Collection extends Callable> tasks)

throws InterruptedException, ExecutionException {

}; public T invokeAny(Collection extends Callable> tasks, long timeout, TimeUnit unit)

throws InterruptedException, ExecutionException, TimeoutException {

}; public List> invokeAll(Collection extends Callable> tasks) throws InterruptedException {

}; public List> invokeAll(Collection extends Callable> tasks, long timeout, TimeUnit unit) throws InterruptedException {

};

}

ThreadPoolExecutor類

public class ThreadPoolExecutor extends AbstractExecutorService {

..... public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,

BlockingQueue workQueue); public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,

BlockingQueue workQueue,ThreadFactory threadFactory); public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,

BlockingQueue workQueue,RejectedExecutionHandler handler); public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,

BlockingQueue workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler);

...public void execute(Runnable command) { if (command == null) throw new NullPointerException(); int c = ctl.get(); if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return;

c = ctl.get();

} if (isRunning(c) && workQueue.offer(command)) { int recheck = ctl.get(); if (! isRunning(recheck) && remove(command))

reject(command); else if (workerCountOf(recheck) == 0)

addWorker(null, false);

} else if (!addWorker(command, false)) reject(command);

}

}

歡迎關(guān)注小編，每日更新最新資訊帶給大家。返回搜狐，查看更多

責(zé)任編輯：