0041 Java学习笔记-多线程-线程池、ForkJoinPool、ThreadLocal

什么是线程池

• 创建线程，因为涉及到跟操作系统交互，比较耗费资源。如果要创建大量的线程，而每个线程的生存期又很短，这时候就应该使用线程池了，就像数据库的连接池一样，预先开启一定数量的线程，有任务了就将任务传递进去，任务执行完毕不终止线程，等待下一个任务

线程池的种类

• ExecutorService：
  • 这是个接口，代表尽快执行的线程池，只要有空闲进程，就立即执行
  • Future<?> submit(Runnable task)
    • 将Runnable对象提交给线程池，线程池有空闲线程时执行任务，返回的Future对象，因为run()没有返回值，因此实际是null，但可以调用isDone()和isCancelled()方法
  • <T> Future <T> submit(Runnable task,T result)
    • result是线程执行结束后的返回值
  • <T> Future <T> submit(Callable<T> task)
    • Future代表Callable的call()方法的返回值
  • void shutdown()
    • 不再接收新任务，已接收的任务执行完成，然后关闭线程池
  • List<Runnable> shutdownNow()
    • 停止所有线程任务，并返回等待处理的任务列表
  • boolean isShutdown()
  • boolean isTerminated()
    • shutdown()之后，所有任务都执行完毕，则返回true
• ScheduledExecutorService
  • 这也是个接口，代表在指定延迟或者周期性的执行任务的线程池
  • ScheduledFuture<V> schedule(Callable<V> callable,long delay,TimeUnit unit)
    • Callable任务在delay后执行
  • ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit)
    • Runnable任务在delay后执行
  • ScheduledFuture<?> scheduleAtFixedTate(Runnable command, long initialDelay, long period, TimeUnit unit)
    • 在delay延迟后开始执行，之后周期性(period)执行
    • 这里的周期是从上一个任务的开始时间开始计算
    • 比如：第一次执行任务在1秒处，执行了3秒，第4秒结束，如果设置的period是5秒，那就在第6秒第二次执行该任务
    • TimeUnit是个枚举类：可以是天、小时、分钟、秒、毫秒、毫微秒
  • ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit)
    • initialDelay后开始执行，执行完毕后，间隔delay后再次执行
    • 执行中遇到异常，会终止执行，否则会一直执行，应设立条件终止任务
    • 跟上一个方法不同的是，这个方法，第二个周期执行的起算点是第一个周期结束时间点

创建线程池

• 通过Executors的静态工厂方法创建线程池
• 创建ExecutorService线程池
  • newCachedThreadPool()
    • 具有缓存功能的线程池，系统根据需要创建线程，缓存与线程池中
  • newFixedThreadPool(int nThreads)
    • 可重用的、具有固定线程数的线程池
  • newSingleThreadExecutor()
    • 单线程的线程池
  • newWorkStealingPool(int parallelism)
    • 创建以讹持有足够线程的线程池来支持给定的并行级别，并且使用多个队列来减少竞争
  • newWorkStealingPool()
    • 这个方法是上个方法的简化版，将cpu个数传给上一个方法就是下面这个方法
• 创建ScheduledExecutorService线程池
  • newScheduledThreadPool(int corePoolSize)
    • corePoolSize是线程数目
  • newSingleThreadScheduledExecutor()
    • 只有一个线程

示例一：

package testpack;

import java.util.Date;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class Test2  { 
    public static void main(String[] args) throws InterruptedException{ 
    	Runnable task=()->{                        //创建一个线程任务
    		for (int i=0;i<1;i++) {
    			System.out.println(new Date());
    			try{
    				Thread.sleep(3000);
    			}catch(InterruptedException ex){
    				ex.printStackTrace();
    			}
    			System.out.println(new Date());
    		}
    	};
    	ScheduledExecutorService ses=Executors.newScheduledThreadPool(1);  //创建一个单线程延迟处理处理线程池，
    	ses.scheduleAtFixedRate(task,1,5,TimeUnit.SECONDS);     //标记㈠。延迟1秒开始处理，从开始处理的时间点开始算，5秒后执行第二个周期
    }
}

输出：

Fri Dec 09 16:22:07 CST 2016 //07秒开始处理
Fri Dec 09 16:22:10 CST 2016 //run中暂停3秒
Fri Dec 09 16:22:12 CST 2016 //07+5秒开始第二个周期
Fri Dec 09 16:22:15 CST 2016 //run中暂停3秒
Fri Dec 09 16:22:17 CST 2016 //07+5+5开始第三个周期

示例二：将上面标记㈠的方法改为：scheduleWithFixedDelay，输出如下：

Fri Dec 09 16:28:35 CST 2016 //35秒开始处理
Fri Dec 09 16:28:38 CST 2016 //run中暂停3秒
Fri Dec 09 16:28:43 CST 2016 //38+5秒开始第二个周期
Fri Dec 09 16:28:46 CST 2016 //run中暂停3秒
Fri Dec 09 16:28:51 CST 2016 //46+5秒开始第三个周期

ForkJoinPool

• ForkJoinPool与多CPU、多核CPU计算有关
• 是ExecutorService的实现类，也是一种线程池
• 配合ForkJoinTask完成对一个大任务进行递归拆解成多个小任务并行计算
• 构造器与方法：
  • ForkJoinPool(int parallelism)
    • 创建一个包含指定个数并行线程的线程池
  • ForkJoinPool()
    • 创建一个Runtime.availableProcessors()返回值个数的并行线程的线程池
  • static ForkJoinPool commenPool()
    • 返回一个通用线程池，其运行状态不受shutdown()和shutdownNow()的影响，除非调用System.exit()退出虚拟机
  • static int getCommenPoolparallelism()
    • 返回通用池的并行级别
  • submit(ForkJoinTask task)
    • 执行指定任务
  • invoke(ForkJoinTask task)
• ForkJoinTask
  • 这是个抽象类，有RecursiveAction()和RecursiveTask()两个抽象子类，前者无返回值，后者有返回值
  • 创建这两类任务时，要继承这两个类，然后重写compute()方法
• 进一步参考：
  • 并发编程网
  • 《Java并发编程的艺术》第6.4节
  • 《Java 7 并发编程实战手册》第5章
• 示例：

package testpack;

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveAction;
import java.util.concurrent.TimeUnit;

public class Test2  { 
    public static void main(String[] args) throws InterruptedException{ 
    	ForkJoinPool pool=new ForkJoinPool();
    	pool.submit(new Task(0,327));
    	pool.awaitTermination(2,TimeUnit.SECONDS);
    	pool.shutdown();
    }
}

class Task extends RecursiveAction{
	private static final int THRESHOLD=50;
	private int start;
	private int end;
	public Task(int start,int end){
		this.start=start;
		this.end=end;
	}
	protected void compute(){
		if (end-start<THRESHOLD) {
			for (int i=start;i<end;i++){
				System.out.println(Thread.currentThread().getName()+"输出： "+i);
			}
		}else{
			int middle = (start+end)/2;
			Task left=new Task(start,middle);
			Task right=new Task(middle,end);
			left.fork();                         
			right.fork();
		}
	}
}

ThreadLocal类

• ThreadLocal可以把一个多个线程一起操作的变量包装成一个局部变量，每个线程都拥有一个该变量的副本，各线程对该变量的操作互补影响
• 只有如下三个方法：
  • void set(T value)
  • T get();
  • void remove()
• 各个线程的该局部变量的初始值是null
• 示例

package testpack;
public class Test1  { 
    public static void main(String[] args){ 
    	Account a=new Account("初始名字");
    	new MyThread(a).start();
    	new MyThread(a).start();
    }
}
class MyThread extends Thread{
	private Account account;
	public MyThread(Account account){
		this.account=account;
	}
	public void run(){
		for (int i=0;i<5;i++) {
			if (i==3){
				account.setName(Thread.currentThread().getName());
			}
			System.out.println("线程名称："+getName()+" ，账户名："+account.getName()+" 输出："+i);
		}
	}
}
class Account {
	private ThreadLocal<String> name=new ThreadLocal<>();
	public Account(String str){
		this.name.set(str);
		System.out.println("构造时的账户名："+this.name.get());
	}
	public String getName(){
		return name.get();
	}
	public void setName(String str){
		this.name.set(str);
	}
} 

输出：

构造时的账户名：初始名字
线程名称：Thread-0 ，账户名：null 输出：0
线程名称：Thread-1 ，账户名：null 输出：0
线程名称：Thread-0 ，账户名：null 输出：1
线程名称：Thread-1 ，账户名：null 输出：1
线程名称：Thread-0 ，账户名：null 输出：2
线程名称：Thread-1 ，账户名：null 输出：2 //初始值为null
线程名称：Thread-0 ，账户名：Thread-0 输出：3
线程名称：Thread-0 ，账户名：Thread-0 输出：4
线程名称：Thread-1 ，账户名：Thread-1 输出：3
线程名称：Thread-1 ，账户名：Thread-1 输出：4