Java并行程序基础（六）

• ThreadFactory

ThreadFactory是一个接口，它只有一个方法，用来创建线程:

Thread newThread(Runnable r);

自定义线程池，可以跟踪线程池究竟何时创建了多少线程，也可以自定义线程的名称，组以及优先级等信息，甚至可以任性的将所有的线程设置为守护线程。总之，使用自定义线程池可以让我们更加自由的设置池子中所有的线程状态

1package com.thread.t02;
2
3import java.util.concurrent.ExecutorService;
4import java.util.concurrent.SynchronousQueue;
5import java.util.concurrent.ThreadFactory;
6import java.util.concurrent.ThreadPoolExecutor;
7import java.util.concurrent.TimeUnit;
8
9public class Test04 {
10	public static class MyTask implements Runnable{
11		@Override
12		public void run() {
13			System.out.println(System.currentTimeMillis()+":Thread ID:"+Thread.currentThread().getId());
14			try {
15				Thread.sleep(100);
16			} catch (InterruptedException e) {
17				e.printStackTrace();
18			}
19		}
20	}
21	public static void main(String[] args) throws InterruptedException {
22		MyTask task = new MyTask();
23		ExecutorService es = new ThreadPoolExecutor(5, 5, 0L, 
24				TimeUnit.MILLISECONDS, 
25				new SynchronousQueue<Runnable>(), new ThreadFactory() {
26					@Override
27					public Thread newThread(Runnable r) {
28						Thread t = new Thread(r);
29						t.setDaemon(true);
30						System.out.println("create "+ t);
31						return t;
32					}
33				});
34		for(int i=0;i<5;i++){
35			es.submit(task);
36		}
37		Thread.sleep(2000);
38	}
39}

• 扩展线程池

ThreadPoolExecutor是一个可以扩展的线程池。它提供了beforeExecute(),afterExecute(),terminated()三个接口对线程池进行控制。

1package com.thread.t02;
2
3import java.util.concurrent.ExecutorService;
4import java.util.concurrent.LinkedBlockingQueue;
5import java.util.concurrent.ThreadPoolExecutor;
6import java.util.concurrent.TimeUnit;
7
8public class Test05 {
9
10	public static class MyTask implements Runnable{
11		public String name;
12		
13		public MyTask(String name) {
14			super();
15			this.name = name;
16		}
17
18		@Override
19		public void run() {
20			System.out.println("正在执行"+":Thread ID:"+Thread.currentThread().getId()+",Task Name="+name);
21			try {
22				Thread.sleep(100);
23			} catch (InterruptedException e) {
24				e.printStackTrace();
25			}
26		}
27		
28	}
29	
30	public static void main(String[] args) throws InterruptedException {
31		ExecutorService es = new ThreadPoolExecutor(5, 5, 0L, 
32				TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()){
33			@Override
34			protected void beforeExecute(Thread t, Runnable r) {
35				System.out.println("准备执行:"+((MyTask)r).name);
36			}
37			@Override
38					protected void afterExecute(Runnable r, Throwable t) {
39						// TODO Auto-generated method stub
40				System.out.println("准备执行:"+((MyTask)r).name);
41					}
42			@Override
43					protected void terminated() {
44						// TODO Auto-generated method stub
45						System.out.println("线程池退出");
46					}
47		};
48		for(int i=0;i<5;i++){
49			MyTask task = new MyTask("Task-GEMY-"+i);
50			es.execute(task);
51			Thread.sleep(10);
52		}
53		es.shutdown();
54	}
55	
56	
57	
58}

• 优化线程池线程数量

Ncpu=CPU的数量

Ucpu=目标CPU的使用率，0<=Ucpu<=1

W/C = 等待时间与计算时间的比率

为保持处理器达到期望的使用率，最优的池的大小等于:

Ntheads=Ncpu*Ucpu*(1+W/C)

在java中，可以通过

Runtime.getRuntime().availableProcessors()

获得可用的CPU的数量。

• 在线程池中寻找堆栈

  package com.thread.t02;

  import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; import java.util.concurrent.SynchronousQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit;

  public class Test06 {

  1public static class DivTask implements Runnable{
  2	int a,b;
  3
  4	public DivTask(int a, int b) {
  5		super();
  6		this.a = a;
  7		this.b = b;
  8	}
  9
  10	@Override
  11	public void run() {
  12		double re = a/b;
  13		System.out.println(re);
  14	}
  15	
  16}
  17
  18public static void main(String[] args) throws InterruptedException, ExecutionException {
  19	ThreadPoolExecutor pools = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 
  20			0L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());
  21	for(int i=0;i<5;i++){
  

  // pools.submit(new DivTask(100,i));//用execute代替submit，可以得到堆栈信息，submit不会打印堆栈信息 pools.execute(new DivTask(100,i)); } } }

  Exception in thread "pool-1-thread-1" java.lang.ArithmeticException: / by zero at com.thread.t02.Test06$DivTask.run(Test06.java:22) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) at java.lang.Thread.run(Thread.java:745) 100.0 50.0 25.0 33.0

我们可以从这个异常堆栈中只能知道异常时在哪里抛出的。但是我们希望得到另外一个更重要的信息，那就是这个任务到底在哪里提交的?

1package com.thread.t02;
2
3import java.util.concurrent.BlockingQueue;
4import java.util.concurrent.SynchronousQueue;
5import java.util.concurrent.ThreadPoolExecutor;
6import java.util.concurrent.TimeUnit;
7
8public class TraceThreadPoolExecutor extends ThreadPoolExecutor{
9	public static class DivTask implements Runnable{
10		int a,b;
11
12		public DivTask(int a, int b) {
13			super();
14			this.a = a;
15			this.b = b;
16		}
17
18		@Override
19		public void run() {
20			double re = a/b;
21			System.out.println(re);
22		}
23		
24	}
25	public TraceThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
26			BlockingQueue<Runnable> workQueue) {
27		super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
28	}
29	
30	private Runnable wrap(final Runnable task ,final Exception clientStack,String clientThreadName){
31		return new Runnable() {
32			@Override
33			public void run() {
34				try{
35					task.run();
36				}catch (Exception e) {
37					clientStack.printStackTrace();
38					throw e;
39				}
40			}
41		};
42	}
43	
44	@Override
45	public void execute(Runnable task) {
46		super.execute(wrap(task, clientTrace(), Thread.currentThread().getName()));
47	}
48	
49	private Exception clientTrace(){
50		return new Exception("Client stack trace");
51	}
52	
53	public static void main(String[] args) {
54		ThreadPoolExecutor pools = new TraceThreadPoolExecutor(0, Integer.MAX_VALUE, 
55				0, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());
56		
57		for(int i=0;i<5;i++){
58			pools.execute(new DivTask(100, i));
59		}
60	}
61}

现在，我们不仅可以得到异常发生的Runnable实现内的信息，也知道这个任务是在哪里提价的。

1java.lang.Exception: Client stack trace100.0
233.0
3
4	at com.thread.t02.TraceThreadPoolExecutor.clientTrace(TraceThreadPoolExecutor.java:50)
5	at com.thread.t02.TraceThreadPoolExecutor.execute(TraceThreadPoolExecutor.java:46)
6	at com.thread.t02.TraceThreadPoolExecutor.main(TraceThreadPoolExecutor.java:58)
750.0
8Exception in thread "pool-1-thread-1" 25.0java.lang.ArithmeticException: / by zero
9
10	at com.thread.t02.TraceThreadPoolExecutor$DivTask.run(TraceThreadPoolExecutor.java:20)
11	at com.thread.t02.TraceThreadPoolExecutor$1.run(TraceThreadPoolExecutor.java:35)
12	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
13	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
14	at java.lang.Thread.run(Thread.java:745)