《java併發程式設計實戰》筆記(一) 結構化併發應用程式
阿新 • • 發佈：2018-12-01
下載地址
連結：https://pan.baidu.com/s/1i6FlscH 密碼：m21n
1.任務執行
任務是一組邏輯執行單元，執行緒是使得任務非同步執行的機制
不可取的所謂執行緒開啟的方式：
1.所有任務放在單個執行緒中序列執行
2.每一個任務都開啟一個執行緒，無限制，非常浪費資源
有效方法：通過有界佇列防止高負荷的應用程式把記憶體耗盡
使用執行緒池 java.util.concurrent Executor框架
1.1 執行緒池
好處：
- web伺服器不會在高負載的情況下失敗
- 伺服器不會建立上千的執行緒來爭奪有限的cpu和記憶體資源
- Executor 可實現各種調優，管理，監控，日誌
介面
public interface Executor{
void executor(Runnable command);
}
| 執行緒池型別 | 作用 |
|---|---|
| newFixedThreadPool | 固定長度的執行緒池，自動維持固定數量的執行緒 |
| newCachedThreadPool | 可快取的執行緒池，自動回收空閒的執行緒，如不足，會新增執行緒，執行緒池規模不存在限制 |
| newSingleThreadExecutor | 單執行緒的Executor,保持單執行緒序列執行 |
| newScheduledThreadPool | 固定數量的執行緒池，延時或者定時執行任務 |
建立執行緒池
import java.io.IOException; import java.util.concurrent.Executor; import java.util.concurrent.Executors; public class ExecutorTest { private static final int NUM = 100; private static final Executor exec = Executors.newFixedThreadPool(NUM); private static int index=0; private static int count=0; public static void main (String[] args) throws IOException { while(count++ < 100){ Runnable task = new Runnable(){ public void run(){ //TODO System.out.println(Thread.currentThread().getName() +" :"+index++); } }; exec.execute(task); } System.exit(1); } } result: pool-1-thread-1 :0 pool-1-thread-3 :2 pool-1-thread-2 :1 pool-1-thread-4 :3 ... pool-1-thread-98 :97 pool-1-thread-99 :98 pool-1-thread-100 :99 不存在超出100的執行緒。
Executor的生命週期
狀態
- 執行
- 關閉
- 已終止
介面
public interface ExecutorService extends Executor{
void shutdown(); //平緩關閉，不接受新的任務，完成執行中的任務再關閉
List<Runnable> shutdownNow; //粗暴關閉，嘗試關閉所有執行中的任務，不接受新的任務
isTerminated(); //檢視執行緒池是否關閉狀態
//....
}
class Demo{
private static final Executor exec = ...
public void start(){
while(!exec.isShutdown){
try{
Runnable task = new Runable(){
public void run(){
//TODO
}
};
exec.execute(task);
}
}catch(RejectExecutorException e){
if(!exec.isShutdown()){
log.info("task rejected",e);
}
}
}
public void stop(){
exec.shutdown();
}
}
延時任務與週期任務
Timer類負責管理這型別任務，但表現非常差，缺點如下：
- 執行只建立一個執行緒處理所有定時任務，無法保證每個任務的定時準確性。
- 不捕獲執行中的異常，直接終止定時執行緒，不會恢復執行緒執行。“執行緒洩漏”
正確使用：
執行緒池：ScheduledThreadPoolExector
方法： 構造方法 或者 newScheduledThreadPool()
排程服務：
1.DelyQueue(實現 BlockingQueue) 為 ScheduledThreadPoolExector提供排程服務
2.每個Deplyed物件都有一個相應的延遲時間
3.在DelyQueue中，只有某個物件逾期了，才能執行take操作。
Runnable的缺點
Runnable 不能返回一個值或者丟擲受檢查的異常
攜帶結果的Callable和Future
介面
public interface Callable<V>{
V call throws Exception; //V == void ,則表示無返回結果
}
public interface Future<V>{
boolean cancel(boolean mayinterruptIfRunning);
boolean isCancelled();
boolean isDone();
V get() throws InterruptedException, ExecutionException, CancellationException; //任務完成立刻返回，或者等待任務完成返回，或者丟擲異常
V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, CancellationException, TimeoutException;
}
使用方式
1.使用Runnable或Callable建立一個例項
2.使用 ExecutorService 的 submit 提交例項 得到 一個 Future
3.通過返回的Future呼叫相應的方法獲得資訊
使用頁面渲染影象(序列下載)
public class FutureRender{
public static final ExecutorService exec = Executors.newFixedThreadPool(100);
/*
*渲染頁面
*/
void renderPage(CharSequence source){
final List<ImageInfo> imageInfos = scanForImageInfo(source); //獲得相關影象的資訊
//圖片的下載構建任務 V = List<ImageDate>
Callable<List<ImageDate>> task = new Callable(){
public List<ImageDate> call(){
List<ImageDate> result = new ArrayList<ImageDate>(); //影象資訊
for(ImageInfo imageInfo : result){
result.add(imageInfo.downImage());
}
return result;
}
};
rendTexe(source); //渲染文字
Future<List<ImageDate>> future = exec.submit(task); //獲得 Future物件
try{
List<ImageDate> imageData = future.get(); //future.get( 指定時間，timeUnit) 限時獲得
for(ImageData iamge : imageData){
//渲染影象
}
}catch(InterruptedException e){
//重新設定執行緒中斷狀態
Thread.currentThread.interrupt();
//取消任務
future.cancel(true);
}catch(ExecutiontException e){
throw launcherThrowable(e.getCause());
}
}
}
優點：
文字和圖片並行處理
缺點：
必須等待所有照片下載完
更高效的並行下載
CompletionService 融合了Executor 和 BlockingQueue
可執行Callable任務
最後通過佇列的類似方式獲得Future 結果
ExectorCompletionService 實現了CompletionService
1.提交的任務被包裝成QueueFuture
2.計算結果放入BlockingQueue
3.方法阻塞，等待所有結果算出
public class Render{
public static final ExecutorService exec = Executors.newFixedThreadPool(100);
/*
*渲染頁面
*/
void renderPage(CharSequence source){
final List<ImageInfo> imageInfos = scanForImageInfo(source); //獲得相關影象的資訊
CompletionService<List<ImageDate>> completionService = new ExectorCompletionService(exec);
for(ImageInfo iamgeInfo : imageInfos){
completionService.submit(new Callable<ImageDate>(){
public ImageDate call(){
return imageInfo.downImage());
}
});
};
rendTexe(source); //渲染文字
Future<List<ImageDate>> future = exec.submit(task); //獲得 Future物件
try{
for(int i=0; i<imageInfos.size(); i++){
Future<ImageData> future = completionService.take();
ImageData imageData = future.get();
//渲染圖片
}
}catch(InterruptedException e){
//重新設定執行緒中斷狀態
Thread.currentThread.interrupt();
}catch(ExecutiontException e){
throw launcherThrowable(e.getCause());
}
}
}
invokeAll 批量獲得future結果
1.支援限時返回任務結果
2.invokeAll 引數：一組任務，結果：一組Future，有序
3.某個任務未完成則取消，通過get() 或者isCancelled判斷情況
案例：在預定時間內獲得旅遊網站報價
//報價任務
//TravelQuote 相關公司的報價資訊
public class QuoteTask implements Callable<TravelQuote>{
private final TravelCompany company; //旅遊公司
private final TravelInfo travelInfo; //旅遊資訊條件
public TravelQuote call(){
return company.quote(travelInfo);
}
}
//獲得多個公司旅遊報價
public List<TravelQuote> getTravelQuotes(TravelInfo travelInfo, set<TravelCompany> companies, Comparator<TravelQuote> ranking, long outtime, TimeUnit unit) throws InterruptedExecption{
List<QuoteTask> tasks = new ArrayList<QuoteTask>(); // 任務列表
for(TravelCompany company : companies){
tasks.add(new QuoteTask(company, travelInfo));
}
List<Future<TravelQuote>> futures = exec.invokeAll(tasks, outtime, unit); //執行緒池執行報價任務返回future list
List<TravelQuote> quotes = new ArrayList<TravelQuote>(tasks.size());
Iterator<QuoteTask> taskIter = tasks.iterator();
for(Future<TravelQuote> f : futures){
QuoteTask task = taskIter.next();
try{
quotes.add(f.get());
}catch(ExecutionException e){
quotes.add(task.getFailureQuote(e.getCause()));
}catch(CanellationExecption e){
quotes.add(task.getTimeOutQuote(e));
}
}
Collections.sort(quotes, ranking);
return quotes;
}