學習筆記 - java.util.concurrent 多線程架構 [轉]

java concurrent 多線程

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

上面是一個簡單的例子，使用了2個大小的線程池來處理100個線程。但有一個問題：在for 循環的過程中，會等待線程池有空閑的線程，是以主線程會阻塞的。為了解決這個問題，一般啟動一個線程來做for循環，就是為了避免由于線程池滿了造成主線 程阻塞。不過在這裡我沒有這樣處理。[重要修正：經過測試，即使線程池大小小于實際線程數大小，線程池也不會阻塞的，這與Tomcat的線程池不同，它将 Runnable執行個體放到一個“無限”的BlockingQueue中，是以就不用一個線程啟動for循環，Doug Lea果然厲害]

另外它使用了Executors的靜态函數生成一個固定的線程池，顧名思義，線程池的線程是 不會釋放的，即使它是Idle。這就會産生性能問題，比如如果線程池的大小為200，當全部使用完畢後，所有的線程會繼續留在池中，相應的記憶體和線程切換 （while(true)+sleep循環）都會增加。如果要避免這個問題，就必須直接使用ThreadPoolExecutor()來構造。可以像 Tomcat的線程池一樣設定“最大線程數”、“最小線程數”和“空閑線程keepAlive的時間”。通過這些可以基本上替換Tomcat的線程池實作 方案。

需要注意的是線程池必須使用shutdown來顯式關閉，否則主線程就無法退出。shutdown也不會阻塞主線程。

許多長時間運作的應用有時候需要定時運作 任務完成一些諸如統計、優化等工作，比如在電信行業中處理使用者話單時，需要每隔1分鐘處理話單；網站每天淩晨統計使用者通路量、使用者數；大型逾時淩晨3點統 計當天銷售額、以及最熱賣的商品；每周日進行資料庫備份；公司每個月的10号計算工資并進行轉帳等，這些都是定時任務。通過 java的并發庫concurrent可以輕松的完成這些任務，而且非常的簡單。

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秒鐘後運作，并每隔2秒運作一次

final ScheduledFuture beeperHandle = scheduler.scheduleAtFixedRate(
beeper, 1, 2, SECONDS);
           

// 2秒鐘後運作，并每次在上次任務運作完後等待5秒後重新運作

final ScheduledFuture beeperHandle2 = scheduler
.scheduleWithFixedDelay(beeper, 2, 5, SECONDS);
           

// 30秒後結束關閉任務，并且關閉Scheduler

scheduler.schedule(new Runnable() {
public void run() {
beeperHandle.cancel(true);
beeperHandle2.cancel(true);
scheduler.shutdown();
}
}, 30, SECONDS);
}
}
           

為了退出程序，上面的代碼中加入了關閉Scheduler的操作。而對于24小時運作的應用而言，是沒有必要關閉Scheduler的。

在實際應用中，有時候需要多個線程同時工作以完成同一件事情，而且在完成過程中，往往會等待其他線程都完成某一階段後再執行，等所有線程都到達某一個階段後再統一執行。

比如有幾個旅行團需要途經深圳、廣州、韶關、長沙最後到達武漢。旅行團中有自駕遊的，有徒步的，有乘坐旅遊大巴的；這些旅行團同時出發，并且每到一個目的地，都要等待其他旅行團到達此地後再同時出發，直到都到達終點站武漢。

這時候CyclicBarrier就可以派上用場。CyclicBarrier最重要的屬性就是參與者個數，另外最要方法是await()。當所有線程都調用了await()後，就表示這些線程都可以繼續執行，否則就會等待。

并發庫中的BlockingQueue是一個比較好玩的類，顧名思義，就是阻塞隊列。該類主要提供了兩個方法put()和take()，前者将一個對象放到隊列中，如果隊列已經滿了，就等待直到有空閑節點；後者從head取一個對象，如果沒有對象，就等待直到有可取的對象。

下面的例子比較簡單，一個讀線程，用于将要處理的檔案對象添加到阻塞隊列中，另外四個寫線程 用于取出檔案對象，為了模拟寫操作耗時長的特點，特讓線程睡眠一段随機長度的時間。另外，該Demo也使用到了線程池和原子整型 （AtomicInteger），AtomicInteger可以在并發情況下達到原子化更新，避免使用了synchronized，而且性能非常高。由 于阻塞隊列的put和take操作會阻塞，為了使線程退出，特在隊列中添加了一個“辨別”，算法中也叫“哨兵”，當發現這個哨兵後，寫線程就退出。

轉自:http://www.ismayday.com/?p=170