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() 。