java并發(fā)編程--線程池初步

三里屯搖滾

java并發(fā)編程--線程池初步

[coolxing按: 轉(zhuǎn)載請(qǐng)注明作者和出處, 如有謬誤, 歡迎在評(píng)論中指正.]



服務(wù)器應(yīng)用程序經(jīng)常需要處理執(zhí)行時(shí)間很短而數(shù)目巨大的請(qǐng)求, 如果為每一個(gè)請(qǐng)求創(chuàng)建一個(gè)新的線程, 會(huì)導(dǎo)致一些問(wèn)題的出現(xiàn), 如:

1. 性能瓶頸. 線程的創(chuàng)建和銷毀需要執(zhí)行大量的后臺(tái)操作, 如果單個(gè)請(qǐng)求的執(zhí)行時(shí)間很短, 有可能花在創(chuàng)建和銷毀線程上的時(shí)間大于真正執(zhí)行請(qǐng)求的時(shí)間.

2. 可能會(huì)導(dǎo)致資源不足. 大量的并發(fā)請(qǐng)求意味著需要?jiǎng)?chuàng)建大量的線程, 過(guò)多的線程存在會(huì)吞噬大量的系統(tǒng)資源, 而且CPU需要在這些線程間不斷切換, 這可能引發(fā)"切換過(guò)度"的問(wèn)題.

為了適應(yīng)上述場(chǎng)合, java在JDK1.5中引入了線程池的概念. 線程池中存放著一定數(shù)量的已創(chuàng)建好的線程, 當(dāng)一個(gè)請(qǐng)求到來(lái)時(shí), 只需從線程池中取出一個(gè)線程來(lái)執(zhí)行請(qǐng)求, 請(qǐng)求完成后再將線程歸還給線程池. 同時(shí), 我們可以為線程池指定最大的線程數(shù)量, 當(dāng)池中所有線程都處于活動(dòng)狀態(tài)下, 新的任務(wù)會(huì)排隊(duì)等候, 直到之前的某個(gè)任務(wù)處理完成后, 新的任務(wù)才能得到處理.  



創(chuàng)建線程池. java.util.concurrent.Executors類提供了多個(gè)靜態(tài)方法用于創(chuàng)建線程池.

|--public static ExecutorService newFixedThreadPool(int nThreads): 創(chuàng)建一個(gè)可重用的固定線程數(shù)的線程池. 如果池中所有的nThreads個(gè)線程都處于活動(dòng)狀態(tài)時(shí)提交任務(wù)(任務(wù)通常是Runnable或Callable對(duì)象), 任務(wù)將在隊(duì)列中等待, 直到池中出現(xiàn)可用線程.

|--public static ExecutorService newCachedThreadPool(): 調(diào)用此方法創(chuàng)建的線程池可根據(jù)需要自動(dòng)調(diào)整池中線程的數(shù)量. 執(zhí)行任務(wù)時(shí)將重用存在先前創(chuàng)建的線程(如果池中存在可用線程的話). 如果池中沒(méi)有可用線程, 將創(chuàng)建一個(gè)新的線程, 并將其添加到池中. 池中的線程超過(guò)60秒未被使用就會(huì)被銷毀, 因此長(zhǎng)時(shí)間保持空閑的CachedThreadPool不會(huì)消耗額外的資源.

|--public static ExecutorService newSingleThreadExecutor(): 創(chuàng)建一個(gè)單線程的Executor. 這個(gè)Executor保證按照任務(wù)提交的順序依次執(zhí)行任務(wù).

|--public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize): 創(chuàng)建一個(gè)可重用的固定線程數(shù)的線程池. ScheduledExecutorService是ExecutorService的子接口, 調(diào)用ScheduledExecutorService的相關(guān)方法, 可以延遲或定期執(zhí)行任務(wù).

以上靜態(tài)方法均使用默認(rèn)的ThreadFactory(即Executors.defaultThreadFactory()方法的返回值)創(chuàng)建線程, 如果想要指定ThreadFactory, 可調(diào)用他們的重載方法.通過(guò)指定ThreadFactory, 可以定制新建線程的名稱, 線程組, 優(yōu)先級(jí), 守護(hù)線程狀態(tài)等.

如果Executors提供的創(chuàng)建線程池的方法無(wú)法滿足要求, 可以使用ThreadPoolExecutor類創(chuàng)建線程池.



提交任務(wù). 所有的線程池都是ExecutorService及其子類的對(duì)象, 因此, 可以調(diào)用ExecutorService的相關(guān)方法提交任務(wù).

|--void execute(Runnable command): 使用池中已存在的線程或新建一個(gè)線程執(zhí)行command.



Java代碼

1. 1.public class ExecutorsDemo {   
   
2. 2.      
   
3. 3.    public static void main(String[] args) throws Exception {   
   
4. 4.        System.out.println("----------------FixedThreadPool---------------------");   
   
5. 5.        ExecutorService fixedPool = getFixedThreadPool();   
   
6. 6.        executeThread(fixedPool);   
   
7. 7.           
   
8. 8.        // 為了避免混淆, 需要等待executeThread(fixedPool)執(zhí)行完成   
   
9. 9.        Thread.sleep(3000);   
   
10. 10.           
    
11. 11.        System.out.println("----------------CashedThreadPool---------------------");   
    
12. 12.        ExecutorService cashedPool = getCashedThreadPool();   
    
13. 13.        executeThread(cashedPool);   
    
14. 14.           
    
15. 15.        // 為了避免混淆, 需要等待executeThread(cashedPool)執(zhí)行完成   
    
16. 16.        Thread.sleep(3000);   
    
17. 17.           
    
18. 18.        System.out.println("----------------SingleThreadExecutor---------------------");   
    
19. 19.        ExecutorService singleExecutor = getSingleThreadExecutor();   
    
20. 20.        executeThread(singleExecutor);   
    
21. 21.           
    
22. 22.    }   
    
23. 23.      
    
24. 24.    // 只存在一個(gè)線程的線程池   
    
25. 25.    private static ExecutorService getSingleThreadExecutor() {   
    
26. 26.        return Executors.newSingleThreadExecutor();   
    
27. 27.    }   
    
28. 28.  
    
29. 29.    // 創(chuàng)建一個(gè)根據(jù)需要自動(dòng)調(diào)整大小的線程池   
    
30. 30.    private static ExecutorService getCashedThreadPool() {   
    
31. 31.        return Executors.newCachedThreadPool();   
    
32. 32.    }   
    
33. 33.  
    
34. 34.    // 創(chuàng)建一個(gè)可重用的固定線程數(shù)的線程池   
    
35. 35.    private static ExecutorService getFixedThreadPool() {   
    
36. 36.        return Executors.newFixedThreadPool(2);   
    
37. 37.    }   
    
38. 38.      
    
39. 39.    private static void executeThread(ExecutorService pool) {   
    
40. 40.        Thread t1 = new MyThread();   
    
41. 41.        Thread t2 = new MyThread();   
    
42. 42.        Thread t3 = new MyThread();   
    
43. 43.        Thread t4 = new MyThread();   
    
44. 44.        // 將Tread放入線程池中執(zhí)行   
    
45. 45.        // MyThread類繼承自Thread, 而Thread類實(shí)現(xiàn)了Runnable接口   
    
46. 46.        pool.execute(t1);   
    
47. 47.        pool.execute(t2);   
    
48. 48.        pool.execute(t3);   
    
49. 49.        pool.execute(t4);   
    
50. 50.        // 關(guān)閉線程池   
    
51. 51.        pool.shutdown();   
    
52. 52.    }   
    
53. 53.      
    
54. 54.    private static final class MyThread extends Thread {   
    
55. 55.        @Override  
    
56. 56.        public void run() {   
    
57. 57.            super.run();   
    
58. 58.            System.out.println(Thread.currentThread().getName() + ": is running!");   
    
59. 59.        }   
    
60. 60.    }   
    
61. 61.}  
    
62. public class ExecutorsDemo {
    
63.        
    
64.         public static void main(String[] args) throws Exception {
    
65.                 System.out.println("----------------FixedThreadPool---------------------");
    
66.                 ExecutorService fixedPool = getFixedThreadPool();
    
67.                 executeThread(fixedPool);
    
68.                
    
69.                 // 為了避免混淆, 需要等待executeThread(fixedPool)執(zhí)行完成
    
70.                 Thread.sleep(3000);
    
71.                
    
72.                 System.out.println("----------------CashedThreadPool---------------------");
    
73.                 ExecutorService cashedPool = getCashedThreadPool();
    
74.                 executeThread(cashedPool);
    
75.                
    
76.                 // 為了避免混淆, 需要等待executeThread(cashedPool)執(zhí)行完成
    
77.                 Thread.sleep(3000);
    
78.                
    
79.                 System.out.println("----------------SingleThreadExecutor---------------------");
    
80.                 ExecutorService singleExecutor = getSingleThreadExecutor();
    
81.                 executeThread(singleExecutor);
    
82.                
    
83.         }
    
84.        
    
85.         // 只存在一個(gè)線程的線程池
    
86.         private static ExecutorService getSingleThreadExecutor() {
    
87.                 return Executors.newSingleThreadExecutor();
    
88.         }
    
89. 
    
90.         // 創(chuàng)建一個(gè)根據(jù)需要自動(dòng)調(diào)整大小的線程池
    
91.         private static ExecutorService getCashedThreadPool() {
    
92.                 return Executors.newCachedThreadPool();
    
93.         }
    
94. 
    
95.         // 創(chuàng)建一個(gè)可重用的固定線程數(shù)的線程池
    
96.         private static ExecutorService getFixedThreadPool() {
    
97.                 return Executors.newFixedThreadPool(2);
    
98.         }
    
99.        
    
100.         private static void executeThread(ExecutorService pool) {
     
101.                 Thread t1 = new MyThread();
     
102.                 Thread t2 = new MyThread();
     
103.                 Thread t3 = new MyThread();
     
104.                 Thread t4 = new MyThread();
     
105.                 // 將Tread放入線程池中執(zhí)行
     
106.                 // MyThread類繼承自Thread, 而Thread類實(shí)現(xiàn)了Runnable接口
     
107.                 pool.execute(t1);
     
108.                 pool.execute(t2);
     
109.                 pool.execute(t3);
     
110.                 pool.execute(t4);
     
111.                 // 關(guān)閉線程池
     
112.                 pool.shutdown();
     
113.         }
     
114.        
     
115.         private static final class MyThread extends Thread {
     
116.                 @Override
     
117.                 public void run() {
     
118.                         super.run();
     
119.                         System.out.println(Thread.currentThread().getName() + ": is running!");
     
120.                 }
     
121.         }
     
122. }

復(fù)制代碼

程序的輸出為:

1. 
   
2. ----------------FixedThreadPool---------------------
   
3. 
   
4. pool-1-thread-1: is running!
   
5. 
   
6. pool-1-thread-2: is running!
   
7. 
   
8. pool-1-thread-2: is running!
   
9. 
   
10. pool-1-thread-1: is running!
    
11. 
    
12. ----------------CashedThreadPool---------------------
    
13. 
    
14. pool-2-thread-1: is running!
    
15. 
    
16. pool-2-thread-2: is running!
    
17. 
    
18. pool-2-thread-4: is running!
    
19. 
    
20. pool-2-thread-3: is running!
    
21. 
    
22. ----------------SingleThreadExecutor---------------------
    
23. 
    
24. pool-3-thread-1: is running!
    
25. 
    
26. pool-3-thread-1: is running!
    
27. 
    
28. pool-3-thread-1: is running!
    
29. 
    
30. pool-3-thread-1: is running!

復(fù)制代碼

|--Future<T> submit(Callable<T> task): 使用池中已存在的線程或新建一個(gè)線程執(zhí)行task, 與execute()方法不同的是, 該方法會(huì)返回線程的執(zhí)行結(jié)果. submit方法接受一個(gè)Callable<T>對(duì)象, Callable<T>接口是一個(gè)泛型接口, 實(shí)現(xiàn)Callable<T>接口需要重寫(xiě)其中的call()方法, call()方法將返回一個(gè)T對(duì)象. submit方法的返回值是Future<T>對(duì)象, 調(diào)用該對(duì)象的get()可以獲得call()方法的返回值.



Java代碼

1. 1.public class FutureDemo {   
   
2. 2.    public static void main(String[] args) throws Exception {   
   
3. 3.        ExecutorService pool = Executors.newFixedThreadPool(2);   
   
4. 4.           
   
5. 5.        Future<Integer> intFuture = pool.submit(new IntegerCallable());   
   
6. 6.                // get()方法將阻塞主線程, 直到IntegerCallable線程的call()運(yùn)行結(jié)束并返回結(jié)果時(shí)為止.   
   
7. 7.        Integer returnInt = intFuture.get();   
   
8. 8.        System.out.println("返回值為" + returnInt);   
   
9. 9.           
   
10. 10.        Future<Boolean> boolFuture = pool.submit(new BooleanCallable());   
    
11. 11.        Boolean returnBool = boolFuture.get();   
    
12. 12.        System.out.println("返回值為" + returnBool);   
    
13. 13.           
    
14. 14.        pool.shutdown();   
    
15. 15.    }   
    
16. 16.      
    
17. 17.    private final static class IntegerCallable implements Callable<Integer> {   
    
18. 18.        // call()方法的返回值類型由泛型決定     
    
19. 19.        @Override  
    
20. 20.        public Integer call() throws Exception {   
    
21. 21.            return 2;   
    
22. 22.        }   
    
23. 23.    }   
    
24. 24.      
    
25. 25.    private final static class BooleanCallable implements Callable<Boolean> {   
    
26. 26.        @Override  
    
27. 27.        public Boolean call() throws Exception {   
    
28. 28.            return true;   
    
29. 29.        }   
    
30. 30.    }   
    
31. 31.}  
    
32. public class FutureDemo {
    
33.         public static void main(String[] args) throws Exception {
    
34.                 ExecutorService pool = Executors.newFixedThreadPool(2);
    
35.                
    
36.                 Future<Integer> intFuture = pool.submit(new IntegerCallable());
    
37.                 // get()方法將阻塞主線程, 直到IntegerCallable線程的call()運(yùn)行結(jié)束并返回結(jié)果時(shí)為止.
    
38.                 Integer returnInt = intFuture.get();
    
39.                 System.out.println("返回值為" + returnInt);
    
40.                
    
41.                 Future<Boolean> boolFuture = pool.submit(new BooleanCallable());
    
42.                 Boolean returnBool = boolFuture.get();
    
43.                 System.out.println("返回值為" + returnBool);
    
44.                
    
45.                 pool.shutdown();
    
46.         }
    
47.        
    
48.         private final static class IntegerCallable implements Callable<Integer> {
    
49.                 // call()方法的返回值類型由泛型決定  
    
50.                 @Override
    
51.                 public Integer call() throws Exception {
    
52.                         return 2;
    
53.                 }
    
54.         }
    
55.        
    
56.         private final static class BooleanCallable implements Callable<Boolean> {
    
57.                 @Override
    
58.                 public Boolean call() throws Exception {
    
59.                         return true;
    
60.                 }
    
61.         }
    
62. }  

復(fù)制代碼

程序的輸出結(jié)果為:

返回值為2

返回值為true

|--List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks): 批量執(zhí)行多個(gè)任務(wù).



Future類. 如果需要獲取線程的執(zhí)行結(jié)果, 那么就會(huì)使用到Future. Future對(duì)象是一個(gè)指向異步執(zhí)行結(jié)果的引用, 由于線程的異步特性, Future對(duì)象在其創(chuàng)建之初可能并不可用, 比如線程的call()方法尚未完成時(shí). 可以調(diào)用Future對(duì)象的isDone()方法判斷線程結(jié)果是否已經(jīng)可用, 在線程結(jié)果返回之前調(diào)用Future對(duì)象的get()方法, 將導(dǎo)致阻塞.



關(guān)閉線程池. 使用完線程池后需要關(guān)閉它, 否則程序可能一直處于運(yùn)行狀態(tài). ExecutorService提供了2個(gè)方法用于關(guān)閉線程池:

|--void shutdown(): 關(guān)閉線程池, 不再接受新任務(wù). 如果存在正在執(zhí)行的任務(wù), 則等待任務(wù)執(zhí)行完成.

|--List<Runnable> shutdownNow(): 關(guān)閉線程池, 不再接受新任務(wù). 盡力嘗試停止正在執(zhí)行的任務(wù), 并返回正在等待的任務(wù)列表.

|--boolean isShutdown(): 判斷線程池是否已經(jīng)關(guān)閉.

心如止水_00

IT168文庫(kù)精選文檔推薦