Java 线程池

1 线程池概念

1.1 现有问题

• 线程是宝贵的内存资源、单个线程约占1MB空间，过多分配易造成内存溢出。
• 频繁的创建及销毁线程会增加虚拟机回收频率、资源开销，造成程序性能下降。

1.2 线程池

• 线程容器，可设定线程分配的数量上限。
• 将预先创建的线程对象存入池中，并重用线程池中的线程对象。
• 避免频繁的创建和销毁。

2 线程池原理

将任务提交给线程池，由线程池分配线程、运行任务，并在当前任务结束后复用线程。

3 如何使用线程池

3.1 获取线程池

常用的线程池接口和类(所在包java.util.concurrent)：

• Executor：线程池的顶级接口。
  • execute(); 执行任务
• ExecutorService：线程池接口，可通过submit(Runnable task) 提交任务代码。
  • shutdown() 关闭线程池
  • isTerminated() 判断线程池中任务和线程是否已经执行完毕。
  • submit() 提交任务
• Executors工厂类：通过此类可以获得一个线程池。
  • newFixedThreadPool(int nThreads) 获取固定数量的线程池。
    参数：指定线程池中线程的数量。
  • newCachedThreadPool() 获得动态数量的线程池，如不够则创建新的，无上限。

eg:

//1 创建固定数量的线程池
ExecutorService es = Executors.newFixedThreadPool(4);
//2 创建动态数量的线程池
ExecutorService es = Executors.newCachedThreadPool();

public class TestExecutors {
    public static void main(String[] args) {
        //1.1 创建固定数量的线程池
        //ExecutorService es = Executors.newFixedThreadPool(4);
        //1.2 创建动态数量的线程池
        ExecutorService es = Executors.newCachedThreadPool();
        //2 添加任务
        Runnable ticket = new Runnable() {
            private int count = 1000;
            @Override
            public void run() {
                while (true) {
                    synchronized (this) {
                        if (count<=0) {
                            break;
                        }
                        System.out.println(
                                Thread.currentThread().getName()+"卖了第"+count+"张票"
                        );
                        count--;
                    }
                }
            }
        };
        for (int i = 0; i < 4; i++) {
            //提交任务
            es.submit(ticket);
        }
        //3 关闭线程池
        es.shutdown();
    }
}

eg:

//3 创建单线程线程池, 只有一个线程的线程池

public class TestSingThread {
    public static void main(String[] args) {
        //3 创建单线程线程池, 只有一个线程的线程池
        ExecutorService es = Executors.newSingleThreadExecutor();
        for (int i = 0; i < 10; i++) {
            es.submit(new Runnable() {
                @Override
                public void run() {
                    System.out.println(Thread.currentThread().getName()+"...");
                }
            });
        }
        es.shutdown();
    }
}

eg:

//4 创建调度线程池, 实现周期执行和延迟执行

public class TestScheduleThread {
    public static void main(String[] args) {
        //4 创建调度线程池, 实现周期执行和延迟执行
        //周期执行: 间隔指定时间执行一次, 不能关闭线程
        ScheduledExecutorService es = Executors.newScheduledThreadPool(1);
        es.scheduleAtFixedRate(new Runnable() {
            int num = 0;
            @Override
            public void run() {
                System.out.println(new Date()+"..."+num);
                num++;
                if (num == 10) {
                    es.shutdown();
                }
            }
        },0,1000, TimeUnit.MILLISECONDS);

        //延迟执行: 延迟一定时间执行
        es.schedule(new Runnable() {
            @Override
            public void run() {
                System.out.println(new Date());
            }
        },3000, TimeUnit.MILLISECONDS);
        es.shutdown();
    }
}

eg:

//5 工作窃取线程池

public class TestWorkStealingPool {
    public static void main(String[] args) {
        //5 工作窃取线程池
        //5.1 工作窃取线程池中的都属于后台线程
        //5.2 每一个线程都有自己的双端队列, 队头和队尾都可以加入和删除元素
        //5.3 自己线程的队列中任务执行完后, 会去别的线程队列的队尾窃取一个任务执行
        //默认使用CPU内核个数作为线程个数
        ExecutorService es = Executors.newWorkStealingPool();
        for (int i = 0; i < 10; i++) {
            es.submit(new Runnable() {
                @Override
                public void run() {
                    System.out.println(Thread.currentThread().getName()+"...");
                }
            });
        }
        es.shutdown();
        //阻止主线程结束
        while (!es.isTerminated()); //判断线程池中的任务和线程是否执行完毕
        System.out.println("执行完毕");
    }
}

4 创建线程的第四种方式

实现 Callable 接口, 重写 call() 方法

• JDK5加入，与Runnable接口类似，实现之后代表一个线程任务。
• Callable具有泛型返回值、可以声明异常。

语法:

public interface Callable<V>{
	public V call() throws Exception;
}

Future接口

• 概念：异步接收ExecutorService.submit()所返回的状态结果，当中包含了call()的返回值。
• 方法：V get()以阻塞形式等待Future中的异步处理结果（call()的返回值）

eg:

MyCallable:

public class MyCallable implements Callable<Integer> {
    @Override
    public Integer call() throws Exception {
        int sum = 0;
        System.out.println(Thread.currentThread().getName()+"开始计算...");
        for (int i = 1; i <= 100; i++) {
            sum+=i;
            Thread.sleep(100);
        }
        System.out.println(Thread.currentThread().getName()+"结束计算...");
        return sum;
    }
}

Test:

public class Test {
    public static void main(String[] args) throws Exception{
        //1 创建可调用对象
        MyCallable mc = new MyCallable();
        //2 创建线程池(单线程线程池)
        ExecutorService es = Executors.newSingleThreadExecutor();
        //3 提交任务
        //Future: 表示线程将要执行完毕的结果
        Future<Integer> future = es.submit(mc);
        //4 获取结果(阻塞方法, 直到线程池中的任务执行完毕才返回)
        System.out.println("结果是:"+future.get());
        //5 关闭线程池
        es.shutdown();
    }
}

课堂案例

eg:

需求：使用两个线程，并发计算1~50、51~100的和，再进行汇总统计。

public class Test {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        //创建动态线程池
        ExecutorService es = Executors.newCachedThreadPool();
        //添加任务1
        Future<Integer> f1 = es.submit(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                int sum = 0;
                for (int i = 1; i <= 50 ; i++) {
                    sum+=i;
                }
                return sum;
            }
        });
        //添加任务2
        Future<Integer> f2 = es.submit(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                int sum = 0;
                for (int i = 51; i <= 100 ; i++) {
                    sum+=i;
                }
                return sum;
            }
        });
        //汇总
        System.out.println("结果为:"+(f1.get() + f2.get()));
        //关闭线程池
        es.shutdown();
    }
}