java 多線程框架

2019-11-06 07:42:35
字體：大中小
來源：轉載
供稿：網友
http://blog.csdn.net/ghsau/article/category/1707779
http://blog.csdn.net/lufeng20/article/details/24314381
1、繼承Thread類、繼承Runnable接口、閉包開辟線程：
package com.busymonkey.concurrent;public class ThreadTest {  	      public static void main(String[] args) {      	ThreadFun1 thread = new ThreadFun1();      	thread.start();    }  }    class ThreadFun1 extends Thread {  	@Override    public void run() {          while(true) {              try {                  Thread.sleep(1000);              } catch (InterruptedException e) {                  e.PRintStackTrace();              }              System.out.println("Hello!");          }      }  }package com.busymonkey.concurrent;public class RunnableTest {  	      public static void main(String[] args) {       	ThreadFun2 thread = new ThreadFun2();          Thread t = new Thread(thread);        t.start();      }  }    class ThreadFun2 implements Runnable {  	@Override    public void run() {          while(true) {              try {                  Thread.sleep(1000);              } catch (InterruptedException e) {                   e.printStackTrace();              }              System.out.println("Hello!");          }      }  }package com.busymonkey.concurrent;public class ThreadTestClosure {	public static void main(String[] args) {		for (int i = 1; i < 5; i ++) {			final int taskId = i;			new Thread(new Runnable() {				public void run() {					for (int i = 1; i < 5; i ++) {						try {							Thread.sleep(20);						} catch (InterruptedException e) {							e.printStackTrace();						}						System.out.println("Task : " + taskId + "; run time :" + i);					}				}			}).start();		}	}}package com.busymonkey.concurrent;public class ThreadTestClosure {	public static void main(String[] args) {		for (int i = 1; i < 5; i ++) {			final int taskId = i;			new Thread() {				public void run() {					for (int i = 1; i < 5; i ++) {						try {							Thread.sleep(20);						} catch (InterruptedException e) {							e.printStackTrace();						}						System.out.println("Task : " + taskId + "; run time :" + i);					}				}			}.start();		}	}}
2、ExecutorService 線程池開辟（開辟固定數量、動態開辟、開辟單個線程）測試發現開辟固定數量1個線程和開辟單個線程方法是一樣效果：
package com.busymonkey.concurrent;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;public class ThreadPoolTest {	public static void main(String[] args) {		// 創建可以容納3個線程的線程池		//ExecutorService threadPool = Executors.newFixedThreadPool(1);		// 線程池的大小會根據執行的任務數動態分配 		//ExecutorService threadPool = Executors.newCachedThreadPool();		// 創建單個線程的線程池，如果當前線程在執行任務時突然中斷，則會創建一個新的線程替代它繼續執行任務		ExecutorService threadPool = Executors.newSingleThreadExecutor();		for (int i = 1; i < 5; i++) {			final int taskID = i;			threadPool.execute(new Runnable() {				public void run() {					for (int i = 1; i < 5; i++) {						try {							Thread.sleep(20);// 為了測試出效果，讓每次任務執行都需要一定時間						} catch (InterruptedException e) {							e.printStackTrace();						}						System.out.println("第" + taskID + "次任務的第" + i + "次執行");					}				}			});		}		threadPool.shutdown();// 任務執行完畢，關閉線程池	}}
3、ExecutorService 線程池開辟定時線程：
package com.busymonkey.concurrent;import java.util.concurrent.Executors;import java.util.concurrent.ScheduledExecutorService;import java.util.concurrent.TimeUnit;public class ThreadPoolTestTimer {	public static void main(String[] args) {		ScheduledExecutorService schedulePool = Executors.newScheduledThreadPool(1);		// 5秒后執行任務		schedulePool.schedule(new Runnable() {			public void run() {				System.out.println("爆炸");			}		}, 5, TimeUnit.SECONDS);		// 5秒后執行任務，以后每2秒執行一次		schedulePool.scheduleAtFixedRate(new Runnable() {			@Override			public void run() {				System.out.println("爆炸");			}		}, 5, 2, TimeUnit.SECONDS);	}}4、ThreadLocal 線程本地變量：
package com.busymonkey.concurrent;public class ThreadLocalTest {      //通過匿名內部類覆蓋ThreadLocal的initialValue()方法，指定初始值      private static ThreadLocal<Integer> seqNum = new ThreadLocal<Integer>() {          public Integer initialValue() {              return 0;          }      };        // ②獲取下一個序列值      public int getNextNum() {          seqNum.set(seqNum.get() + 1);          return seqNum.get();      }        public static void main(String[] args) {      	ThreadLocalTest sn = new ThreadLocalTest();          //3個線程共享sn，各自產生序列號          TestClient t1 = new TestClient(sn);          TestClient t2 = new TestClient(sn);          TestClient t3 = new TestClient(sn);          t1.start();          t2.start();          t3.start();      }        private static class TestClient extends Thread {          private ThreadLocalTest sn;            public TestClient(ThreadLocalTest sn) {              this.sn = sn;          }            public void run() {              for (int i = 0; i < 3; i++) {                  //每個線程打出3個序列值                  System.out.println("thread[" + Thread.currentThread().getName() + "] --> sn["                           + sn.getNextNum() + "]");              }          }      }  }  
Thread同步機制的比較
　　ThreadLocal和線程同步機制相比有什么優勢呢？ThreadLocal和線程同步機制都是為了解決多線程中相同變量的訪問沖突問題。
　　在同步機制中，通過對象的鎖機制保證同一時間只有一個線程訪問變量。這時該變量是多個線程共享的，使用同步機制要求程序慎密地分析什么時候對變量進行讀寫，什么時候需要鎖定某個對象，什么時候釋放對象鎖等繁雜的問題，程序設計和編寫難度相對較大。
　　而ThreadLocal則從另一個角度來解決多線程的并發訪問。ThreadLocal會為每一個線程提供一個獨立的變量副本，從而隔離了多個線程對數據的訪問沖突。因為每一個線程都擁有自己的變量副本，從而也就沒有必要對該變量進行同步了。ThreadLocal提供了線程安全的共享對象，在編寫多線程代碼時，可以把不安全的變量封裝進ThreadLocal。
　　由于ThreadLocal中可以持有任何類型的對象，低版本JDK所提供的get()返回的是Object對象，需要強制類型轉換。但JDK 5.0通過泛型很好的解決了這個問題，在一定程度地簡化ThreadLocal的使用，代碼清單 9 2就使用了JDK 5.0新的ThreadLocal<T>版本。
　　概括起來說，對于多線程資源共享的問題，同步機制采用了“以時間換空間”的方式，而ThreadLocal采用了“以空間換時間”的方式。前者僅提供一份變量，讓不同的線程排隊訪問，而后者為每一個線程都提供了一份變量，因此可以同時訪問而互不影響。
　　spring使用ThreadLocal解決線程安全問題我們知道在一般情況下，只有無狀態的Bean才可以在多線程環境下共享，在Spring中，絕大部分Bean都可以聲明為singleton作用域。就是因為Spring對一些Bean（如RequestContextHolder、TransactionSynchronizationManager、LocaleContextHolder等）中非線程安全狀態采用ThreadLocal進行處理，讓它們也成為線程安全的狀態，因為有狀態的Bean就可以在多線程中共享了。
　　一般的Web應用劃分為展現層、服務層和持久層三個層次，在不同的層中編寫對應的邏輯，下層通過接口向上層開放功能調用。在一般情況下，從接收請求到返回響應所經過的所有程序調用都同屬于一個線程，如圖9?2所示：
　　同一線程貫通三層這樣你就可以根據需要，將一些非線程安全的變量以ThreadLocal存放，在同一次請求響應的調用線程中，所有關聯的對象引用到的都是同一個變量。
ThreadLocal類接口很簡單，只有4個方法，我們先來了解一下：
void set(Object value)設置當前線程的線程局部變量的值。public Object get()該方法返回當前線程所對應的線程局部變量。public void remove()將當前線程局部變量的值刪除，目的是為了減少內存的占用，該方法是JDK 5.0新增的方法。需要指出的是，當線程結束后，對應該線程的局部變量將自動被垃圾回收，所以顯式調用該方法清除線程的局部變量并不是必須的操作，但它可以加快內存回收的速度。protected Object initialValue()返回該線程局部變量的初始值，該方法是一個protected的方法，顯然是為了讓子類覆蓋而設計的。這個方法是一個延遲調用方法，在線程第1次調用get()或set(Object)時才執行，并且僅執行1次。ThreadLocal中的缺省實現直接返回一個null。　　值得一提的是，在JDK5.0中，ThreadLocal已經支持泛型，該類的類名已經變為ThreadLocal<T>。API方法也相應進行了調整，新版本的API方法分別是void set(T value)、T get()以及T initialValue()。
　　ThreadLocal是如何做到為每一個線程維護變量的副本的呢？其實實現的思路很簡單：在ThreadLocal類中有一個Map，用于存儲每一個線程的變量副本，Map中元素的鍵為線程對象，而值對應線程的變量副本。
5、Callable和Future：
Callable接口類似于Runnable，從名字就可以看出來了，但是Runnable不會返回結果，并且無法拋出返回結果的異常，而Callable功能更強大一些，被線程執行后，可以返回值，這個返回值可以被Future拿到，也就是說，Future可以拿到異步執行任務的返回值。
package com.busymonkey.concurrent;import java.util.Random;import java.util.concurrent.Callable;import java.util.concurrent.ExecutionException;import java.util.concurrent.FutureTask;public class CallableAndFuture {	public static void main(String[] args) {        Callable<Integer> callable = new Callable<Integer>() {            public Integer call() throws Exception {                return new Random().nextInt(100);            }        };        FutureTask<Integer> future = new FutureTask<Integer>(callable);        new Thread(future).start();        try {            Thread.sleep(5000);// 可能做一些事情            System.out.println(future.get());        } catch (InterruptedException e) {            e.printStackTrace();        } catch (ExecutionException e) {            e.printStackTrace();        }    }}public class CallableAndFuture {    public static void main(String[] args) {        ExecutorService threadPool = Executors.newSingleThreadExecutor();        Future<Integer> future = threadPool.submit(new Callable<Integer>() {            public Integer call() throws Exception {                return new Random().nextInt(100);            }        });        try {            Thread.sleep(5000);// 可能做一些事情            System.out.println(future.get());        } catch (InterruptedException e) {            e.printStackTrace();        } catch (ExecutionException e) {            e.printStackTrace();        }    }}public class CallableAndFuture {    public static void main(String[] args) {        ExecutorService threadPool = Executors.newCachedThreadPool();        CompletionService<Integer> cs = new ExecutorCompletionService<Integer>(threadPool);        for(int i = 1; i < 5; i++) {            final int taskID = i;            cs.submit(new Callable<Integer>() {                public Integer call() throws Exception {                    return taskID;                }            });        }        // 可能做一些事情        for(int i = 1; i < 5; i++) {            try {                System.out.println(cs.take().get());            } catch (InterruptedException e) {                e.printStackTrace();            } catch (ExecutionException e) {                e.printStackTrace();            }        }    }} 
6、ForkJoinPool：
ForkJoinPool：這個類實現了ExecutorService接口和工作竊取算法（Work-Stealing Algorithm）。它管理工作者線程，并提供任務的狀態信息，以及任務的執行信息。
RecursiveAction：用于任務沒有返回結果的場景。
RecursiveTask：用于任務有返回結果的場景。
package com.busymonkey.concurrent;import java.util.concurrent.ForkJoinPool;import java.util.concurrent.RecursiveAction;import java.util.concurrent.TimeUnit;public class ForkJoinPoolTest {	public static void main(String[] args) throws InterruptedException {		ForkJoinPool pool = new ForkJoinPool();		pool.submit(new PrintTask(1, 100));		pool.awaitTermination(2, TimeUnit.SECONDS);// 阻塞當前線程直到 ForkJoinPool													 //中所有的任務都執行結束		pool.shutdown();	}}class PrintTask extends RecursiveAction {	private static final long serialVersionUID = 8635119133774500468L;	private int start;	private int end;	private int num;	final int MAX = 50;	public PrintTask(int start, int end) {		this.start = start;		this.end = end;	}	protected void compute() {		if (end - start < 50) {			for (int i = start; i <= end; i++) {				num += i;			}			System.out.println("當前任務結果為： " + num);		} else {			int mid = (end + start) / 2;			PrintTask left = new PrintTask(start, mid);			PrintTask right = new PrintTask(mid + 1, end);			left.fork();			right.fork();		}	}}
package com.busymonkey.concurrent;import java.util.concurrent.ExecutionException;import java.util.concurrent.ForkJoinPool;import java.util.concurrent.Future;import java.util.concurrent.RecursiveTask;public class ForkJoinPoolTask {	public static void main(String[] args) throws InterruptedException {		Integer result = 0;		ForkJoinPool pool = new ForkJoinPool();		Future<Integer> future = pool.submit(new PrintTask1(1, 9999));		try {			result = future.get();		} catch (ExecutionException e) {			e.printStackTrace();		}		System.out.println("當前任務結果為： " + result);		pool.shutdownNow();		//pool.shutdown();	}}class PrintTask1 extends RecursiveTask<Integer> {	private static final long serialVersionUID = 8635119133774500468L;	private int start;	private int end;	private int num;	final int MAX = 50;	public PrintTask1(int start, int end) {		this.start = start;		this.end = end;	}	protected Integer compute() {		if (end - start < 20) {			for (int i = start; i <= end; i++) {				num += i;			}			System.out.println("當前任務結果為： " + num);			return num;		} else {			int mid = (end + start) / 2;			PrintTask1 left = new PrintTask1(start, mid);			PrintTask1 right = new PrintTask1(mid + 1, end);			left.fork();			right.fork();			return left.join() + right.join();		}	}}
線程的暫停有兩個方法 shutdown 與shutdownNow 兩個方法的調用都會阻止新任務的提交，區別是關于已經提交未完成任務的處理已經線程終端的處理，shutdown會繼續執行并且完成所有未執行的任務，shutdownNow 會清楚所有未執行的任務并且在運行線程上調用interrupt() 。