JDK5中的一個(gè)亮點(diǎn)就是將Doug Lea的并發(fā)庫(kù)引入到Java標(biāo)準(zhǔn)庫(kù)中。Doug Lea確實(shí)是一個(gè)牛人，能教書(shū)，能出書(shū)，能編碼，不過(guò)這在國(guó)外還是比較普遍的，而國(guó)內(nèi)的教授們就相差太遠(yuǎn)了。
一般的服務(wù)器都需要線程池，比如Web、FTP等服務(wù)器，不過(guò)它們一般都自己實(shí)現(xiàn)了線程池，比如以前介紹過(guò)的Tomcat、Resin和Jetty等，現(xiàn)在有了JDK5，我們就沒(méi)有必要重復(fù)造車輪了，直接使用就可以，何況使用也很方便，性能也非常高。
Java代碼
package?concurrent; ??import?java.util.concurrent.ExecutorService; ??import?java.util.concurrent.Executors; ??public?class?TestThreadPool?{ ??public?static?void?main(String?args[])?throws?InterruptedException?{ ????ExecutorService?exec?=?Executors.newFixedThreadPool(2); ??for(int?index?=?0;?index?<?100;?index++)?{ ??Runnable?run?=?new?Runnable()?{ ??public?void?run()?{ ??long?time?=?(long)?(Math.random()?*?1000); ??System.out.println(“Sleeping?”?+?time?+?“ms”); ??try?{ ??Thread.sleep(time); ??}?catch?(InterruptedException?e)?{ ??} ??} ??}; ??exec.execute(run); ??} ????exec.shutdown(); ??} ??}?? package concurrent;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class TestThreadPool {
public static void main(String args[]) throws InterruptedException {
// only two threads
ExecutorService exec = Executors.newFixedThreadPool(2);
for(int index = 0; index < 100; index++) {
Runnable run = new Runnable() {
public void run() {
long time = (long) (Math.random() * 1000);
System.out.println(“Sleeping ” + time + “ms”);
try {
Thread.sleep(time);
} catch (InterruptedException e) {
}
}
};
exec.execute(run);
}
// must shutdown
exec.shutdown();
}
}
上面是一個(gè)簡(jiǎn)單的例子，使用了2個(gè)大小的線程池來(lái)處理100個(gè)線程。但有一個(gè)問(wèn)題：在for循環(huán)的過(guò)程中，會(huì)等待線程池有空閑的線程，所以主線程會(huì)阻塞的。為了解決這個(gè)問(wèn)題，一般啟動(dòng)一個(gè)線程來(lái)做for循環(huán)，就是為了避免由于線程池滿了造成主線程阻塞。不過(guò)在這里我沒(méi)有這樣處理。[重要修正：經(jīng)過(guò)測(cè)試，即使線程池大小小于實(shí)際線程數(shù)大小，線程池也不會(huì)阻塞的，這與Tomcat的線程池不同，它將Runnable實(shí)例放到一個(gè)“無(wú)限”的BlockingQueue中，所以就不用一個(gè)線程啟動(dòng)for循環(huán)，Doug Lea果然厲害]
另外它使用了Executors的靜態(tài)函數(shù)生成一個(gè)固定的線程池，顧名思義，線程池的線程是不會(huì)釋放的，即使它是Idle。這就會(huì)產(chǎn)生性能問(wèn)題，比如如果線程池的大小為200，當(dāng)全部使用完畢后，所有的線程會(huì)繼續(xù)留在池中，相應(yīng)的內(nèi)存和線程切換（while(true)+sleep循環(huán)）都會(huì)增加。如果要避免這個(gè)問(wèn)題，就必須直接使用ThreadPoolExecutor()來(lái)構(gòu)造。可以像Tomcat的線程池一樣設(shè)置“最大線程數(shù)”、“最小線程數(shù)”和“空閑線程keepAlive的時(shí)間”。通過(guò)這些可以基本上替換Tomcat的線程池實(shí)現(xiàn)方案。
需要注意的是線程池必須使用shutdown來(lái)顯式關(guān)閉，否則主線程就無(wú)法退出。shutdown也不會(huì)阻塞主線程。
許多長(zhǎng)時(shí)間運(yùn)行的應(yīng)用有時(shí)候需要定時(shí)運(yùn)行任務(wù)完成一些諸如統(tǒng)計(jì)、優(yōu)化等工作，比如在電信行業(yè)中處理用戶話單時(shí)，需要每隔1分鐘處理話單；網(wǎng)站每天凌晨統(tǒng)計(jì)用戶訪問(wèn)量、用戶數(shù)；大型超時(shí)凌晨3點(diǎn)統(tǒng)計(jì)當(dāng)天銷售額、以及最熱賣的商品；每周日進(jìn)行數(shù)據(jù)庫(kù)備份；公司每個(gè)月的10號(hào)計(jì)算工資并進(jìn)行轉(zhuǎn)帳等，這些都是定時(shí)任務(wù)。通過(guò) java的并發(fā)庫(kù)concurrent可以輕松的完成這些任務(wù)，而且非常的簡(jiǎn)單。
Java代碼
package?concurrent; ??import?static?java.util.concurrent.TimeUnit.SECONDS; ??import?java.util.Date; ??import?java.util.concurrent.Executors; ??import?java.util.concurrent.ScheduledExecutorService; ??import?java.util.concurrent.ScheduledFuture; ??public?class?TestScheduledThread?{ ??public?static?void?main(String[]?args)?{ ??final?ScheduledExecutorService?scheduler?=?Executors ??.newScheduledThreadPool(2); ??final?Runnable?beeper?=?new?Runnable()?{ ??int?count?=?0; ??public?void?run()?{ ??System.out.println(new?Date()?+?”?beep?”?+?(++count)); ??} ??}; ????final?ScheduledFuture?beeperHandle?=?scheduler.scheduleAtFixedRate( ??beeper,?1,?2,?SECONDS); ????final?ScheduledFuture?beeperHandle2?=?scheduler ??.scheduleWithFixedDelay(beeper,?2,?5,?SECONDS); ????scheduler.schedule(new?Runnable()?{ ??public?void?run()?{ ??beeperHandle.cancel(true); ??beeperHandle2.cancel(true); ??scheduler.shutdown(); ??} ??},?30,?SECONDS); ??} ??}?? package concurrent;
import static java.util.concurrent.TimeUnit.SECONDS;
import java.util.Date;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
public class TestScheduledThread {
public static void main(String[] args) {
final ScheduledExecutorService scheduler = Executors
.newScheduledThreadPool(2);
final Runnable beeper = new Runnable() {
int count = 0;
public void run() {
System.out.println(new Date() + ” beep ” + (++count));
}
};
// 1秒鐘后運(yùn)行，并每隔2秒運(yùn)行一次
final ScheduledFuture beeperHandle = scheduler.scheduleAtFixedRate(
beeper, 1, 2, SECONDS);
// 2秒鐘后運(yùn)行，并每次在上次任務(wù)運(yùn)行完后等待5秒后重新運(yùn)行
final ScheduledFuture beeperHandle2 = scheduler
.scheduleWithFixedDelay(beeper, 2, 5, SECONDS);
// 30秒后結(jié)束關(guān)閉任務(wù)，并且關(guān)閉Scheduler
scheduler.schedule(new Runnable() {
public void run() {
beeperHandle.cancel(true);
beeperHandle2.cancel(true);
scheduler.shutdown();
}
}, 30, SECONDS);
}
}
為了退出進(jìn)程，上面的代碼中加入了關(guān)閉Scheduler的操作。而對(duì)于24小時(shí)運(yùn)行的應(yīng)用而言，是沒(méi)有必要關(guān)閉Scheduler的。
在實(shí)際應(yīng)用中，有時(shí)候需要多個(gè)線程同時(shí)工作以完成同一件事情，而且在完成過(guò)程中，往往會(huì)等待其他線程都完成某一階段后再執(zhí)行，等所有線程都到達(dá)某一個(gè)階段后再統(tǒng)一執(zhí)行。
比如有幾個(gè)旅行團(tuán)需要途經(jīng)深圳、廣州、韶關(guān)、長(zhǎng)沙最后到達(dá)武漢。旅行團(tuán)中有自駕游的，有徒步的，有乘坐旅游大巴的；這些旅行團(tuán)同時(shí)出發(fā)，并且每到一個(gè)目的地，都要等待其他旅行團(tuán)到達(dá)此地后再同時(shí)出發(fā)，直到都到達(dá)終點(diǎn)站武漢。
這時(shí)候CyclicBarrier就可以派上用場(chǎng)。CyclicBarrier最重要的屬性就是參與者個(gè)數(shù)，另外最要方法是await()。當(dāng)所有線程都調(diào)用了await()后，就表示這些線程都可以繼續(xù)執(zhí)行，否則就會(huì)等待。
Java代碼
package?concurrent; ??import?java.text.SimpleDateFormat; ??import?java.util.Date; ??import?java.util.concurrent.BrokenBarrierException; ??import?java.util.concurrent.CyclicBarrier; ??import?java.util.concurrent.ExecutorService; ??import?java.util.concurrent.Executors; ??public?class?TestCyclicBarrier?{ ????private?static?int[]?timeWalk?=?{?5,?8,?15,?15,?10?}; ????private?static?int[]?timeSelf?=?{?1,?3,?4,?4,?5?}; ????private?static?int[]?timeBus?=?{?2,?4,?6,?6,?7?}; ????static?String?now()?{ ??SimpleDateFormat?sdf?=?new?SimpleDateFormat(“HH:mm:ss”); ??return?sdf.format(new?Date())?+?“:?“; ??} ????static?class?Tour?implements?Runnable?{ ??private?int[]?times; ??private?CyclicBarrier?barrier; ??private?String?tourName; ??public?Tour(CyclicBarrier?barrier,?String?tourName,?int[]?times)?{ ??this.times?=?times; ??this.tourName?=?tourName; ??this.barrier?=?barrier; ??} ??public?void?run()?{ ??try?{ ??Thread.sleep(times[0]?*?1000); ??System.out.println(now()?+?tourName?+?”?Reached?Shenzhen”); ??barrier.await(); ??Thread.sleep(times[1]?*?1000); ??System.out.println(now()?+?tourName?+?”?Reached?Guangzhou”); ??barrier.await(); ??Thread.sleep(times[2]?*?1000); ??System.out.println(now()?+?tourName?+?”?Reached?Shaoguan”); ??barrier.await(); ??Thread.sleep(times[3]?*?1000); ??System.out.println(now()?+?tourName?+?”?Reached?Changsha”); ??barrier.await(); ??Thread.sleep(times[4]?*?1000); ??System.out.println(now()?+?tourName?+?”?Reached?Wuhan”); ??barrier.await(); ??}?catch?(InterruptedException?e)?{ ??}?catch?(BrokenBarrierException?e)?{ ??} ??} ??} ????public?static?void?main(String[]?args)?{ ????CyclicBarrier?barrier?=?new?CyclicBarrier(3); ??ExecutorService?exec?=?Executors.newFixedThreadPool(3); ??exec.submit(new?Tour(barrier,?“WalkTour”,?timeWalk)); ??exec.submit(new?Tour(barrier,?“SelfTour”,?timeSelf)); ??exec.submit(new?Tour(barrier,?“BusTour”,?timeBus)); ??exec.shutdown(); ??} ??}?? package concurrent;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class TestCyclicBarrier {
// 徒步需要的時(shí)間: Shenzhen, Guangzhou, Shaoguan, Changsha, Wuhan
private static int[] timeWalk = { 5, 8, 15, 15, 10 };
// 自駕游
private static int[] timeSelf = { 1, 3, 4, 4, 5 };
// 旅游大巴
private static int[] timeBus = { 2, 4, 6, 6, 7 };static String now() {
SimpleDateFormat sdf = new SimpleDateFormat(“HH:mm:ss”);
return sdf.format(new Date()) + “: “;
}static class Tour implements Runnable {
private int[] times;
private CyclicBarrier barrier;
private String tourName;
public Tour(CyclicBarrier barrier, String tourName, int[] times) {
this.times = times;
this.tourName = tourName;
this.barrier = barrier;
}
public void run() {
try {
Thread.sleep(times[0] * 1000);
System.out.println(now() + tourName + ” Reached Shenzhen”);
barrier.await();
Thread.sleep(times[1] * 1000);
System.out.println(now() + tourName + ” Reached Guangzhou”);
barrier.await();
Thread.sleep(times[2] * 1000);
System.out.println(now() + tourName + ” Reached Shaoguan”);
barrier.await();
Thread.sleep(times[3] * 1000);
System.out.println(now() + tourName + ” Reached Changsha”);
barrier.await();
Thread.sleep(times[4] * 1000);
System.out.println(now() + tourName + ” Reached Wuhan”);
barrier.await();
} catch (InterruptedException e) {
} catch (BrokenBarrierException e) {
}
}
}public static void main(String[] args) {
// 三個(gè)旅行團(tuán)
CyclicBarrier barrier = new CyclicBarrier(3);
ExecutorService exec = Executors.newFixedThreadPool(3);
exec.submit(new Tour(barrier, “WalkTour”, timeWalk));
exec.submit(new Tour(barrier, “SelfTour”, timeSelf));
exec.submit(new Tour(barrier, “BusTour”, timeBus));
exec.shutdown();
}
}
運(yùn)行結(jié)果：
00:02:25: SelfTour Reached Shenzhen
00:02:25: BusTour Reached Shenzhen
00:02:27: WalkTour Reached Shenzhen
00:02:30: SelfTour Reached Guangzhou
00:02:31: BusTour Reached Guangzhou
00:02:35: WalkTour Reached Guangzhou
00:02:39: SelfTour Reached Shaoguan
00:02:41: BusTour Reached Shaoguan
并發(fā)庫(kù)中的BlockingQueue是一個(gè)比較好玩的類，顧名思義，就是阻塞隊(duì)列。該類主要提供了兩個(gè)方法put()和take()，前者將一個(gè)對(duì)象放到隊(duì)列中，如果隊(duì)列已經(jīng)滿了，就等待直到有空閑節(jié)點(diǎn)；后者從head取一個(gè)對(duì)象，如果沒(méi)有對(duì)象，就等待直到有可取的對(duì)象。
下面的例子比較簡(jiǎn)單，一個(gè)讀線程，用于將要處理的文件對(duì)象添加到阻塞隊(duì)列中，另外四個(gè)寫(xiě)線程用于取出文件對(duì)象，為了模擬寫(xiě)操作耗時(shí)長(zhǎng)的特點(diǎn)，特讓線程睡眠一段隨機(jī)長(zhǎng)度的時(shí)間。另外，該Demo也使用到了線程池和原子整型（AtomicInteger），AtomicInteger可以在并發(fā)情況下達(dá)到原子化更新，避免使用了synchronized，而且性能非常高。由于阻塞隊(duì)列的put和take操作會(huì)阻塞，為了使線程退出，特在隊(duì)列中添加了一個(gè)“標(biāo)識(shí)”，算法中也叫“哨兵”，當(dāng)發(fā)現(xiàn)這個(gè)哨兵后，寫(xiě)線程就退出。
當(dāng)然線程池也要顯式退出了。
Java代碼
package?concurrent; ??import?java.io.File; ??import?java.io.FileFilter; ??import?java.util.concurrent.BlockingQueue; ??import?java.util.concurrent.ExecutorService; ??import?java.util.concurrent.Executors; ??import?java.util.concurrent.LinkedBlockingQueue; ??import?java.util.concurrent.atomic.AtomicInteger; ????public?class?TestBlockingQueue?{ ??static?long?randomTime()?{ ??return?(long)?(Math.random()?*?1000); ??} ????public?static?void?main(String[]?args)?{ ????final?BlockingQueue?queue?=?new?LinkedBlockingQueue(100); ????final?ExecutorService?exec?=?Executors.newFixedThreadPool(5); ??final?File?root?=?new?File(“F://JavaLib”); ????final?File?exitFile?=?new?File(“”); ????final?AtomicInteger?rc?=?new?AtomicInteger(); ????final?AtomicInteger?wc?=?new?AtomicInteger(); ????Runnable?read?=?new?Runnable()?{ ??public?void?run()?{ ??scanFile(root); ??scanFile(exitFile); ??} ????public?void?scanFile(File?file)?{ ??if?(file.isDirectory())?{ ??File[]?files?=?file.listFiles(new?FileFilter()?{ ??public?boolean?accept(File?pathname)?{ ??return?pathname.isDirectory() ??||?pathname.getPath().endsWith(“.java”); ??} ??}); ??for?(File?one?:?files) ??scanFile(one); ??}?else?{ ??try?{ ??int?index?=?rc.incrementAndGet(); ??System.out.println(“Read0:?”?+?index?+?”?“ ??+?file.getPath()); ??queue.put(file); ??}?catch?(InterruptedException?e)?{ ??} ??} ??} ??}; ??exec.submit(read); ????for?(int?index?=?0;?index?<?4;?index++)?{ ????final?int?NO?=?index; ??Runnable?write?=?new?Runnable()?{ ??String?threadName?=?“Write”?+?NO; ??public?void?run()?{ ??while?(true)?{ ??try?{ ??Thread.sleep(randomTime()); ??int?index?=?wc.incrementAndGet(); ??File?file?=?queue.take(); ????if?(file?==?exitFile)?{ ????queue.put(exitFile); ??break; ??} ??System.out.println(threadName?+?“:?”?+?index?+?”?“ ??+?file.getPath()); ??}?catch?(InterruptedException?e)?{ ??} ??} ??} ??}; ??exec.submit(write); ??} ??exec.shutdown(); ??} ??}?? package concurrent;
import java.io.File;
import java.io.FileFilter;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;public class TestBlockingQueue {
static long randomTime() {
return (long) (Math.random() * 1000);
}public static void main(String[] args) {
// 能容納100個(gè)文件
final BlockingQueue queue = new LinkedBlockingQueue(100);
// 線程池
final ExecutorService exec = Executors.newFixedThreadPool(5);
final File root = new File(“F://JavaLib”);
// 完成標(biāo)志
final File exitFile = new File(“”);
// 讀個(gè)數(shù)
final AtomicInteger rc = new AtomicInteger();
// 寫(xiě)個(gè)數(shù)
final AtomicInteger wc = new AtomicInteger();
// 讀線程
Runnable read = new Runnable() {
public void run() {
scanFile(root);
scanFile(exitFile);
}public void scanFile(File file) {
if (file.isDirectory()) {
File[] files = file.listFiles(new FileFilter() {
public boolean accept(File pathname) {
return pathname.isDirectory()
|| pathname.getPath().endsWith(“.java”);
}
});
for (File one : files)
scanFile(one);
} else {
try {
int index = rc.incrementAndGet();
System.out.println(“Read0: ” + index + ” “
+ file.getPath());
queue.put(file);
} catch (InterruptedException e) {
}
}
}
};
exec.submit(read);
// 四個(gè)寫(xiě)線程
for (int index = 0; index < 4; index++) {
// write thread
final int NO = index;
Runnable write = new Runnable() {
String threadName = “Write” + NO;
public void run() {
while (true) {
try {
Thread.sleep(randomTime());
int index = wc.incrementAndGet();
File file = queue.take();
// 隊(duì)列已經(jīng)無(wú)對(duì)象
if (file == exitFile) {
// 再次添加”標(biāo)志”，以讓其他線程正常退出
queue.put(exitFile);
break;
}
System.out.println(threadName + “: ” + index + ” “
+ file.getPath());
} catch (InterruptedException e) {
}
}
}
};
exec.submit(write);
}
exec.shutdown();
}
}
從名字可以看出，CountDownLatch是一個(gè)倒數(shù)計(jì)數(shù)的鎖，當(dāng)?shù)箶?shù)到0時(shí)觸發(fā)事件，也就是開(kāi)鎖，其他人就可以進(jìn)入了。在一些應(yīng)用場(chǎng)合中，需要等待某個(gè)條件達(dá)到要求后才能做后面的事情；同時(shí)當(dāng)線程都完成后也會(huì)觸發(fā)事件，以便進(jìn)行后面的操作。
CountDownLatch最重要的方法是countDown()和await()，前者主要是倒數(shù)一次，后者是等待倒數(shù)到0，如果沒(méi)有到達(dá)0，就只有阻塞等待了。
一個(gè)CountDouwnLatch實(shí)例是不能重復(fù)使用的，也就是說(shuō)它是一次性的，鎖一經(jīng)被打開(kāi)就不能再關(guān)閉使用了，如果想重復(fù)使用，請(qǐng)考慮使用CyclicBarrier。
下面的例子簡(jiǎn)單的說(shuō)明了CountDownLatch的使用方法，模擬了100米賽跑，10名選手已經(jīng)準(zhǔn)備就緒，只等裁判一聲令下。當(dāng)所有人都到達(dá)終點(diǎn)時(shí)，比賽結(jié)束。
同樣，線程池需要顯式shutdown。
Java代碼
package?concurrent; ????import?java.util.concurrent.CountDownLatch; ??import?java.util.concurrent.ExecutorService; ??import?java.util.concurrent.Executors; ????public?class?TestCountDownLatch?{ ??public?static?void?main(String[]?args)?throws?InterruptedException?{ ????final?CountDownLatch?begin?=?new?CountDownLatch(1); ????final?CountDownLatch?end?=?new?CountDownLatch(10); ????final?ExecutorService?exec?=?Executors.newFixedThreadPool(10); ??for(int?index?=?0;?index?<?10;?index++)?{ ??final?int?NO?=?index?+?1; ??Runnable?run?=?new?Runnable(){ ??public?void?run()?{ ??try?{ ??begin.await(); ??Thread.sleep((long)?(Math.random()?*?10000)); ??System.out.println(“No.”?+?NO?+?”?arrived”); ??}?catch?(InterruptedException?e)?{ ??}?finally?{ ??end.countDown(); ??} ??} ??}; ??exec.submit(run); ??} ??System.out.println(“Game?Start”); ??begin.countDown(); ??end.await(); ??System.out.println(“Game?Over”); ??exec.shutdown(); ??} ??}?? package concurrent;import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;public class TestCountDownLatch {
public static void main(String[] args) throws InterruptedException {
// 開(kāi)始的倒數(shù)鎖
final CountDownLatch begin = new CountDownLatch(1);
// 結(jié)束的倒數(shù)鎖
final CountDownLatch end = new CountDownLatch(10);
// 十名選手
final ExecutorService exec = Executors.newFixedThreadPool(10);
for(int index = 0; index < 10; index++) {
final int NO = index + 1;
Runnable run = new Runnable(){
public void run() {
try {
begin.await();
Thread.sleep((long) (Math.random() * 10000));
System.out.println(“No.” + NO + ” arrived”);
} catch (InterruptedException e) {
} finally {
end.countDown();
}
}
};
exec.submit(run);
}
System.out.println(“Game Start”);
begin.countDown();
end.await();
System.out.println(“Game Over”);
exec.shutdown();
}
}
運(yùn)行結(jié)果:
Game Start
No.4 arrived
No.1 arrived
No.7 arrived
No.9 arrived
No.3 arrived
No.2 arrived
No.8 arrived
No.10 arrived
No.6 arrived
No.5 arrived
Game Over
有時(shí)候在實(shí)際應(yīng)用中，某些操作很耗時(shí)，但又不是不可或缺的步驟。比如用網(wǎng)頁(yè)瀏覽器瀏覽新聞時(shí)，最重要的是要顯示文字內(nèi)容，至于與新聞相匹配的圖片就沒(méi)有那么重要的，所以此時(shí)首先保證文字信息先顯示，而圖片信息會(huì)后顯示，但又不能不顯示，由于下載圖片是一個(gè)耗時(shí)的操作，所以必須一開(kāi)始就得下載。
Java的并發(fā)庫(kù)的Future類就可以滿足這個(gè)要求。Future的重要方法包括get()和cancel()，get()獲取數(shù)據(jù)對(duì)象，如果數(shù)據(jù)沒(méi)有加載，就會(huì)阻塞直到取到數(shù)據(jù)，而 cancel()是取消數(shù)據(jù)加載。另外一個(gè)get(timeout)操作，表示如果在timeout時(shí)間內(nèi)沒(méi)有取到就失敗返回，而不再阻塞。
下面的Demo簡(jiǎn)單的說(shuō)明了Future的使用方法：一個(gè)非常耗時(shí)的操作必須一開(kāi)始啟動(dòng)，但又不能一直等待；其他重要的事情又必須做，等完成后，就可以做不重要的事情。
Java代碼
package?concurrent; ????import?java.util.concurrent.Callable; ??import?java.util.concurrent.ExecutionException; ??import?java.util.concurrent.ExecutorService; ??import?java.util.concurrent.Executors; ??import?java.util.concurrent.Future; ????public?class?TestFutureTask?{ ??public?static?void?main(String[]?args)throws?InterruptedException, ??ExecutionException?{ ??final?ExecutorService?exec?=?Executors.newFixedThreadPool(5); ??Callable?call?=?new?Callable()?{ ??public?String?call()?throws?Exception?{ ??Thread.sleep(1000?*?5); ??return?“Other?less?important?but?longtime?things.”; ??} ??}; ??Future?task?=?exec.submit(call); ????Thread.sleep(1000?*?3); ??System.out.println(“Let’s?do?important?things.”); ????String?obj?=?task.get(); ??System.out.println(obj); ????exec.shutdown(); ??} ??}?? package concurrent;import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;public class TestFutureTask {
public static void main(String[] args)throws InterruptedException,
ExecutionException {
final ExecutorService exec = Executors.newFixedThreadPool(5);
Callable call = new Callable() {
public String call() throws Exception {
Thread.sleep(1000 * 5);
return “Other less important but longtime things.”;
}
};
Future task = exec.submit(call);
// 重要的事情
Thread.sleep(1000 * 3);
System.out.println(“Let’s do important things.”);
// 其他不重要的事情
String obj = task.get();
System.out.println(obj);
// 關(guān)閉線程池
exec.shutdown();
}
}
運(yùn)行結(jié)果：
Let’s do important things.
Other less important but longtime things.
考慮以下場(chǎng)景：瀏覽網(wǎng)頁(yè)時(shí)，瀏覽器了5個(gè)線程下載網(wǎng)頁(yè)中的圖片文件，由于圖片大小、網(wǎng)站訪問(wèn)速度等諸多因素的影響，完成圖片下載的時(shí)間就會(huì)有很大的不同。如果先下載完成的圖片就會(huì)被先顯示到界面上，反之，后下載的圖片就后顯示。
Java的并發(fā)庫(kù)的CompletionService可以滿足這種場(chǎng)景要求。該接口有兩個(gè)重要方法：submit()和take()。submit用于提交一個(gè)runnable或者callable，一般會(huì)提交給一個(gè)線程池處理；而take就是取出已經(jīng)執(zhí)行完畢runnable或者callable實(shí)例的Future對(duì)象，如果沒(méi)有滿足要求的，就等待了。 CompletionService還有一個(gè)對(duì)應(yīng)的方法poll，該方法與take類似，只是不會(huì)等待，如果沒(méi)有滿足要求，就返回null對(duì)象。
Java代碼
package?concurrent; ????import?java.util.concurrent.Callable; ??import?java.util.concurrent.CompletionService; ??import?java.util.concurrent.ExecutionException; ??import?java.util.concurrent.ExecutorCompletionService; ??import?java.util.concurrent.ExecutorService; ??import?java.util.concurrent.Executors; ??import?java.util.concurrent.Future; ????public?class?TestCompletionService?{ ??public?static?void?main(String[]?args)?throws?InterruptedException, ??ExecutionException?{ ??ExecutorService?exec?=?Executors.newFixedThreadPool(10); ??CompletionService?serv?= ??new?ExecutorCompletionService(exec); ????for?(int?index?=?0;?index?<?5;?index++)?{ ??final?int?NO?=?index; ??Callable?downImg?=?new?Callable()?{ ??public?String?call()?throws?Exception?{ ??Thread.sleep((long)?(Math.random()?*?10000)); ??return?“Downloaded?Image?”?+?NO; ??} ??}; ??serv.submit(downImg); ??} ????Thread.sleep(1000?*?2); ??System.out.println(“Show?web?content”); ??for?(int?index?=?0;?index?<?5;?index++)?{ ??Future?task?=?serv.take(); ??String?img?=?task.get(); ??System.out.println(img); ??} ??System.out.println(“End”); ????exec.shutdown(); ??} ??}?? package concurrent;import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;public class TestCompletionService {
public static void main(String[] args) throws InterruptedException,
ExecutionException {
ExecutorService exec = Executors.newFixedThreadPool(10);
CompletionService serv =
new ExecutorCompletionService(exec);for (int index = 0; index < 5; index++) {
final int NO = index;
Callable downImg = new Callable() {
public String call() throws Exception {
Thread.sleep((long) (Math.random() * 10000));
return “Downloaded Image ” + NO;
}
};
serv.submit(downImg);
}Thread.sleep(1000 * 2);
System.out.println(“Show web content”);
for (int index = 0; index < 5; index++) {
Future task = serv.take();
String img = task.get();
System.out.println(img);
}
System.out.println(“End”);
// 關(guān)閉線程池
exec.shutdown();
}
}
運(yùn)行結(jié)果:
Show web content
Downloaded Image 1
Downloaded Image 2
Downloaded Image 4
Downloaded Image 0
Downloaded Image 3
End
操作系統(tǒng)的信號(hào)量是個(gè)很重要的概念，在進(jìn)程控制方面都有應(yīng)用。Java并發(fā)庫(kù)的Semaphore可以很輕松完成信號(hào)量控制，Semaphore可以控制某個(gè)資源可被同時(shí)訪問(wèn)的個(gè)數(shù)，acquire()獲取一個(gè)許可，如果沒(méi)有就等待，而release()釋放一個(gè)許可。比如在Windows下可以設(shè)置共享文件的最大客戶端訪問(wèn)個(gè)數(shù)。
Semaphore維護(hù)了當(dāng)前訪問(wèn)的個(gè)數(shù)，提供同步機(jī)制，控制同時(shí)訪問(wèn)的個(gè)數(shù)。在數(shù)據(jù)結(jié)構(gòu)中鏈表可以保存“無(wú)限”的節(jié)點(diǎn)，用Semaphore可以實(shí)現(xiàn)有限大小的鏈表。另外重入鎖ReentrantLock也可以實(shí)現(xiàn)該功能，但實(shí)現(xiàn)上要負(fù)責(zé)些，代碼也要復(fù)雜些。
下面的Demo中申明了一個(gè)只有5個(gè)許可的Semaphore，而有20個(gè)線程要訪問(wèn)這個(gè)資源，通過(guò)acquire()和release()獲取和釋放訪問(wèn)許可。
Java代碼
package?concurrent; ????import?java.util.concurrent.ExecutorService; ??import?java.util.concurrent.Executors; ??import?java.util.concurrent.Semaphore; ????public?class?TestSemaphore?{ ??public?static?void?main(String[]?args)?{ ????ExecutorService?exec?=?Executors.newCachedThreadPool(); ????final?Semaphore?semp?=?new?Semaphore(5); ????for?(int?index?=?0;?index?<?20;?index++)?{ ??final?int?NO?=?index; ??Runnable?run?=?new?Runnable()?{ ??public?void?run()?{ ??try?{ ????semp.acquire(); ??System.out.println(“Accessing:?”?+?NO); ??Thread.sleep((long)?(Math.random()?*?10000)); ????semp.release(); ??}?catch?(InterruptedException?e)?{ ??} ??} ??}; ??exec.execute(run); ??} ????exec.shutdown(); ??} ??}?? package concurrent;import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;public class TestSemaphore {
public static void main(String[] args) {
// 線程池
ExecutorService exec = Executors.newCachedThreadPool();
// 只能5個(gè)線程同時(shí)訪問(wèn)
final Semaphore semp = new Semaphore(5);
// 模擬20個(gè)客戶端訪問(wèn)
for (int index = 0; index < 20; index++) {
final int NO = index;
Runnable run = new Runnable() {
public void run() {
try {
// 獲取許可
semp.acquire();
System.out.println(“Accessing: ” + NO);
Thread.sleep((long) (Math.random() * 10000));
// 訪問(wèn)完后，釋放
semp.release();
} catch (InterruptedException e) {
}
}
};
exec.execute(run);
}
// 退出線程池
exec.shutdown();
}
}
運(yùn)行結(jié)果：
Accessing: 0
Accessing: 1
Accessing: 2
Accessing: 3
Accessing: 4
Accessing: 5
Accessing: 6
Accessing: 7
Accessing: 8
Accessing: 9
Accessing: 10
Accessing: 11
Accessing: 12
Accessing: 13
Accessing: 14
Accessing: 15
Accessing: 16
Accessing: 17
Accessing: 18
Accessing: 19
轉(zhuǎn)載于:https://www.cnblogs.com/liaomin416100569/archive/2010/04/21/9331750.html
總結(jié)
以上是生活随笔為你收集整理的java.util.concurrent 多线程框架的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。
如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。
