Java 四种线程池

资料

介绍new Thread的弊端及Java四种线程池的使用，对Android同样适用。本文是基础篇，后面会分享下线程池一些高级功能。

1、new Thread的弊端
执行一个异步任务你还只是如下new Thread吗？

new Thread(new Runnable() {
 
@Override
public void run() {
// TODO Auto-generated method stub
}
}).start();

那你就out太多了，new Thread的弊端如下：

a. 每次new Thread新建对象性能差。
b. 线程缺乏统一管理，可能无限制新建线程，相互之间竞争，及可能占用过多系统资源导致死机或oom。
c. 缺乏更多功能，如定时执行、定期执行、线程中断。
相比new Thread，Java提供的四种线程池的好处在于：
a. 重用存在的线程，减少对象创建、消亡的开销，性能佳。
b. 可有效控制最大并发线程数，提高系统资源的使用率，同时避免过多资源竞争，避免堵塞。
c. 提供定时执行、定期执行、单线程、并发数控制等功能。

2、Java 线程池

Java通过Executors提供四种线程池，分别为：
newCachedThreadPool创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。
newFixedThreadPool 创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。
newScheduledThreadPool 创建一个定长线程池，支持定时及周期性任务执行。
newSingleThreadExecutor 创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。

(1). newCachedThreadPool
创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。示例代码如下：

ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
for (int i = 0; i < 10; i++) {
final int index = i;
try {
Thread.sleep(index * 1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
 
cachedThreadPool.execute(new Runnable() {
 
@Override
public void run() {
System.out.println(index);
}
});
}

线程池为无限大，当执行第二个任务时第一个任务已经完成，会复用执行第一个任务的线程，而不用每次新建线程。

(2). newFixedThreadPool
创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。示例代码如下：

ExecutorService fixedThreadPool = Executors.newFixedThreadPool(3);
for (int i = 0; i < 10; i++) {
final int index = i;
fixedThreadPool.execute(new Runnable() {
 
@Override
public void run() {
try {
System.out.println(index);
Thread.sleep(2000);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
});
}

因为线程池大小为3，每个任务输出index后sleep 2秒，所以每两秒打印3个数字。

定长线程池的大小最好根据系统资源进行设置。如Runtime.getRuntime().availableProcessors()。可参考PreloadDataCache。

(3) newScheduledThreadPool
创建一个定长线程池，支持定时及周期性任务执行。延迟执行示例代码如下：

package cn.iigrowing.threads.study.ThreadPool;

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

public class NewScheduledThreadPool {

 public static void main(String[] args) {
 ScheduledExecutorService scheduledThreadPool = Executors
 .newScheduledThreadPool(5);
 System.out.println("------" + new Date().toLocaleString());
 scheduledThreadPool.schedule(new Runnable() {

 // @Override
 public void run() {
 System.out.println("delay 3 seconds");
 System.out.println("-2----" + new Date().toLocaleString());
 }
 }, 3, TimeUnit.SECONDS);
 }
}

表示延迟3秒执行。

定期执行示例代码如下：

package cn.iigrowing.threads.study.ThreadPool;

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

public class NewScheduledThreadPool2 {

 public static void main(String[] args) {
 ScheduledExecutorService scheduledThreadPool = Executors
 .newScheduledThreadPool(5);
 System.out.println("------" + new Date().toLocaleString());
 scheduledThreadPool.scheduleAtFixedRate(new Runnable() {

 // @Override
 public void run() {
 System.out
 .println("delay 1 seconds, and excute every 3 seconds");
 System.out.println("-2----" + new Date().toLocaleString());
 }
 }, 1, 3, TimeUnit.SECONDS);
 }
}

表示延迟1秒后每3秒执行一次。

ScheduledExecutorService比Timer更安全，功能更强大，后面会有一篇单独进行对比。

(4)、newSingleThreadExecutor
创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。示例代码如下：

ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
for (int i = 0; i < 10; i++) {
final int index = i;
singleThreadExecutor.execute(new Runnable() {
 
@Override
public void run() {
try {
System.out.println(index);
Thread.sleep(2000);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
});
}

结果依次输出，相当于顺序执行各个任务。

现行大多数GUI程序都是单线程的。Android中单线程可用于数据库操作，文件操作，应用批量安装，应用批量删除等不适合并发但可能IO阻塞性及影响UI线程响应的操作。

线程池的作用：

线程池作用就是限制系统中执行线程的数量。
根 据系统的环境情况，可以自动或手动设置线程数量，达到运行的最佳效果；少了浪费了系统资源，多了造成系统拥挤效率不高。用线程池控制线程数量，其他线程排 队等候。一个任务执行完毕，再从队列的中取最前面的任务开始执行。若队列中没有等待进程，线程池的这一资源处于等待。当一个新任务需要运行时，如果线程池 中有等待的工作线程，就可以开始运行了；否则进入等待队列。

为什么要用线程池:

1.减少了创建和销毁线程的次数，每个工作线程都可以被重复利用，可执行多个任务。

2.可以根据系统的承受能力，调整线程池中工作线线程的数目，防止因为消耗过多的内存，而把服务器累趴下(每个线程需要大约1MB内存，线程开的越多，消耗的内存也就越大，最后死机)。

Java里面线程池的顶级接口是Executor，但是严格意义上讲Executor并不是一个线程池，而只是一个执行线程的工具。真正的线程池接口是ExecutorService。

比较重要的几个类：

ExecutorService真正的线程池接口。
ScheduledExecutorService能和Timer/TimerTask类似，解决那些需要任务重复执行的问题。
ThreadPoolExecutorExecutorService的默认实现。
ScheduledThreadPoolExecutor继承ThreadPoolExecutor的ScheduledExecutorService接口实现，周期性任务调度的类实现。

要配置一个线程池是比较复杂的，尤其是对于线程池的原理不是很清楚的情况下，很有可能配置的线程池不是较优的，因此在Executors类里面提供了一些静态工厂，生成一些常用的线程池。

1. newSingleThreadExecutor

创建一个单线程的线程池。这个线程池只有一个线程在工作，也就是相当于单线程串行执行所有任务。如果这个唯一的线程因为异常结束，那么会有一个新的线程来替代它。此线程池保证所有任务的执行顺序按照任务的提交顺序执行。

2.newFixedThreadPool

创建固定大小的线程池。每次提交一个任务就创建一个线程，直到线程达到线程池的最大大小。线程池的大小一旦达到最大值就会保持不变，如果某个线程因为执行异常而结束，那么线程池会补充一个新线程。

3. newCachedThreadPool

创建一个可缓存的线程池。如果线程池的大小超过了处理任务所需要的线程，

那么就会回收部分空闲（60秒不执行任务）的线程，当任务数增加时，此线程池又可以智能的添加新线程来处理任务。此线程池不会对线程池大小做限制，线程池大小完全依赖于操作系统（或者说JVM）能够创建的最大线程大小。

4.newScheduledThreadPool

创建一个大小无限的线程池。此线程池支持定时以及周期性执行任务的需求。

实例

1：newSingleThreadExecutor·

MyThread.java

public class MyThread extends Thread {

    @Override

    public void run() {

        System.out.println(Thread.currentThread().getName() + "正在执行。。。");

    }

}

TestSingleThreadExecutor.java

public class TestSingleThreadExecutor {

    public static void main(String[] args) {

        //创建一个可重用固定线程数的线程池

        ExecutorService pool = Executors. newSingleThreadExecutor();

        //创建实现了Runnable接口对象，Thread对象当然也实现了Runnable接口

        Thread t1 = new MyThread();

        Thread t2 = new MyThread();

        Thread t3 = new MyThread();

        Thread t4 = new MyThread();

        Thread t5 = new MyThread();

        //将线程放入池中进行执行

        pool.execute(t1);

        pool.execute(t2);

        pool.execute(t3);

        pool.execute(t4);

        pool.execute(t5);

        //关闭线程池

        pool.shutdown();

    }

}

输出结果

pool-1-thread-1正在执行。。。

pool-1-thread-1正在执行。。。

pool-1-thread-1正在执行。。。

pool-1-thread-1正在执行。。。

pool-1-thread-1正在执行。。。
复制代码

2：newFixedThreadPool

TestFixedThreadPool.Java

publicclass TestFixedThreadPool {

    publicstaticvoid main(String[] args) {

        //创建一个可重用固定线程数的线程池

        ExecutorService pool = Executors.newFixedThreadPool(2);

        //创建实现了Runnable接口对象，Thread对象当然也实现了Runnable接口

        Thread t1 = new MyThread();

        Thread t2 = new MyThread();

        Thread t3 = new MyThread();

        Thread t4 = new MyThread();

        Thread t5 = new MyThread();

        //将线程放入池中进行执行

        pool.execute(t1);

        pool.execute(t2);

        pool.execute(t3);

        pool.execute(t4);

        pool.execute(t5);

        //关闭线程池

        pool.shutdown();

    }

}

输出结果

pool-1-thread-1正在执行。。。

pool-1-thread-2正在执行。。。

pool-1-thread-1正在执行。。。

pool-1-thread-2正在执行。。。

pool-1-thread-1正在执行。。。

3：newCachedThreadPool

TestCachedThreadPool.java

public class TestCachedThreadPool {

    public static void main(String[] args) {

        //创建一个可重用固定线程数的线程池

        ExecutorService pool = Executors.newCachedThreadPool();

        //创建实现了Runnable接口对象，Thread对象当然也实现了Runnable接口

        Thread t1 = new MyThread();

        Thread t2 = new MyThread();

        Thread t3 = new MyThread();

        Thread t4 = new MyThread();

        Thread t5 = new MyThread();

        //将线程放入池中进行执行

        pool.execute(t1);

        pool.execute(t2);

        pool.execute(t3);

        pool.execute(t4);

        pool.execute(t5);

        //关闭线程池

        pool.shutdown();

    }

}

输出结果：

pool-1-thread-2正在执行。。。

pool-1-thread-4正在执行。。。

pool-1-thread-3正在执行。。。

pool-1-thread-1正在执行。。。

pool-1-thread-5正在执行。。。

4：newScheduledThreadPool

TestScheduledThreadPoolExecutor.java

public class TestScheduledThreadPoolExecutor {

    public static void main(String[] args) {

        ScheduledThreadPoolExecutor exec = new ScheduledThreadPoolExecutor(1);

        exec.scheduleAtFixedRate(new Runnable() {//每隔一段时间就触发异常

                      @Override

                      publicvoid run() {

                           //throw new RuntimeException();

                           System.out.println("================");

                      }

                  }, 1000, 5000, TimeUnit.MILLISECONDS);

        exec.scheduleAtFixedRate(new Runnable() {//每隔一段时间打印系统时间，证明两者是互不影响的

                      @Override

                      publicvoid run() {

                           System.out.println(System.nanoTime());

                      }

                  }, 1000, 2000, TimeUnit.MILLISECONDS);

    }

}

输出结果

================

8384644549516

8386643829034

8388643830710

================

8390643851383

8392643879319

8400643939383
来源：https://www.cnblogs.com/zhujiabin/p/5404771.html

 

一、Excutors的newScheduleThreadPool程序结构

1、创建延时任务线程池的Excutors源码，ScheduledThreadPoolExecutor实现了ScheduleExecutorService接口
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {

return new ScheduledThreadPoolExecutor(corePoolSize);

}

2、ScheduledThreadPoolExecutor源码，ScheduledThreadPoolExecutor继承自ThreadPoolExcute类，所以执行super构造方法调用的是ThreadPoolExcute构造函数。

public ScheduledThreadPoolExecutor(int corePoolSize) {

super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,

new DelayedWorkQueue());

}

3、我们在构造ThreadPoolExcute时，Queue队列使用了DelayedWorkQueue，这是一个可延时执行阻塞任务的队列，源码如下

private static class DelayedWorkQueue

extends AbstractCollection<Runnable>

implements BlockingQueue<Runnable> {

private final DelayQueue<RunnableScheduledFuture> dq = new DelayQueue<RunnableScheduledFuture>();}

我们看到DelayedWordkQueue继承了AbstractCollection接口，实现了BlockingQueue，所以和ArrayBlockingQueue以及LinkedBlockingQueue是兄弟关系。

DelayedWorkQueue定义了一个DelayQueue<RunnableScheduledFuture> dq，所以DelayedWorkQueue的实现是依赖DelayQueue的，

二、关于DelayQueue

JDK是这样定义的：Delayed元素的一个无界阻塞队列，只有在延迟期满时才能从中提取元素。该队列的头部是延迟期满后保存时间最长的Delayed元素。如果延迟都还没有期满，则队列没有头部，并且poll将返回null。当一个元素的 getDelay(TimeUnit.NANOSECONDS)方法返回一个小于等于0的值时，将发生到期。即使无法使用take或poll移除未到期的元素，也不会将这些元素作为正常元素对待。例如，size方法同时返回到期和未到期元素的计数。此队列不允许使用null元素。

可以简单理解他就是一个使用时间作为比较条件的优先级阻塞队列。

1、来看下DelayQUeue源码
public class DelayQueue<E extends Delayed> extends AbstractQueue<E>

implements BlockingQueue<E> {

private transient final ReentrantLock lock = new ReentrantLock();

private transient final Condition available = lock.newCondition();

private final PriorityQueue<E> q = new PriorityQueue<E>();}

2、再来看下他的pool方法和take方法，poll和take都是取元素，并且删除头部元素，区别在于poll()是非阻塞的，如果没有到期的元素则返回null，take()是一直阻塞到返回到期的头部元素。peek()：获取但不移除此队列的头部；如果此队列为空，则返回 null。与poll不同，如果队列中没有到期元素可用，则此方法返回下一个将到期的元素（如果存在一个这样的元素）。
/**

* Retrieves and removes the head of this queue, or returns <tt>null</tt>

* if this queue has no elements with an expired delay.

*

* @return the head of this queue, or <tt>null</tt> if this

* queue has no elements with an expired delay

*/

public E poll() {

final ReentrantLock lock = this.lock;

lock.lock();

try {

E first = q.peek();

if (first == null || first.getDelay(TimeUnit.NANOSECONDS) > 0)

return null;

else {

E x = q.poll();

assert x != null;

if (q.size() != 0)

available.signalAll();

return x;

}

} finally {

lock.unlock();

}

}

/**

* Retrieves and removes the head of this queue, waiting if necessary

* until an element with an expired delay is available on this queue.

*

* @return the head of this queue

* @throws InterruptedException {@inheritDoc}

*/

public E take() throws InterruptedException {

final ReentrantLock lock = this.lock;

lock.lockInterruptibly();

try {

for (;;) {

E first = q.peek();

if (first == null) {

available.await();

} else {

long delay = first.getDelay(TimeUnit.NANOSECONDS);

if (delay > 0) {

long tl = available.awaitNanos(delay);

} else {

E x = q.poll();

assert x != null;

if (q.size() != 0)

available.signalAll(); // wake up other takers

return x;

}

}

}

} finally {

lock.unlock();

}

}

3、DelayQueue是怎么实现的：我们知道在使用scheduledThreadPool时，我们给传递了TimeOut和TimeUnit参数作为延时执行时间。那么DelayQueue是怎么实现的？我们看源码
/**

* Retrieves and removes the head of this queue, waiting if necessary

* until an element with an expired delay is available on this queue,

* or the specified wait time expires.

*

* @return the head of this queue, or <tt>null</tt> if the

* specified waiting time elapses before an element with

* an expired delay becomes available

* @throws InterruptedException {@inheritDoc}

*/

public E poll(long timeout, TimeUnit unit) throws InterruptedException {

long nanos = unit.toNanos(timeout);

final ReentrantLock lock = this.lock;

lock.lockInterruptibly();

try {

for (;;) {

E first = q.peek();

if (first == null) {

if (nanos <= 0)

return null;

else

nanos = available.awaitNanos(nanos);

} else {

long delay = first.getDelay(TimeUnit.NANOSECONDS);

if (delay > 0) {

if (nanos <= 0)

return null;

if (delay > nanos)

delay = nanos;

long timeLeft = available.awaitNanos(delay);

nanos -= delay - timeLeft;

} else {

E x = q.poll();

assert x != null;

if (q.size() != 0)

available.signalAll();

return x;

}

}

}

} finally {

lock.unlock();

}

}

unit.toNanos(timeout),将时间单位和timeout结合转换成一个纳秒，首先判断peek取出的元素是否为null，如果为null则比较nanos大小，nanos小于0，表示不需要等待，则返回null，如果大于0，则等待nanos单位的时间。如果peek取出的元素不为null，说明头部不为空，再取剩余的延时单位，如果没有延时了，则直接返回q.poll()并且激活其他线程，否则等待nanos单位的时间。

4、有关lock.condition和await()、awaitNanos(long timeout)方法

condition的定义Condition available = lock.newCondition();(newCondition方法是FairSync和NoFairSync类实现的)

awaitNanos造成当前线程在接到信号、被中断或到达指定等待时间之前一直处于等待状态。
与此条件相关的锁以原子方式释放，并且出于线程调度的目的，将禁用当前线程，且在发生以下五种情况之一 以前，当前线程将一直处于休眠状态：

其他某个线程调用此 condition 的 signal() 方法，并且碰巧将当前线程选为被唤醒的线程；或者
其他某个线程调用此 condition 的 signalall() 方法；或者
其他某个线程中断当前线程，且支持中断线程的挂起；或者
已超过指定的等待时间；

await则纯粹是一种和wait差不多的阻塞方法，阻塞当前线程，释放原子锁。

 

关于ThreadFactory的支持
ThreadFactory defaultThreadFactory()

返回用于创建新线程的默认线程工厂。

   public static ThreadFactory defaultThreadFactory() {
        return new DefaultThreadFactory();
    }
ThreadFactory privilegedThreadFactory()

返回用于创建新线程的线程工厂，这些新线程与当前线程具有相同的权限。

    public static ThreadFactory privilegedThreadFactory() {
        return new PrivilegedThreadFactory();
    }

归纳

ExecutorServiceScheduledExecutorServiceThreadFactoryCallable
newFixedThreadPoolnewScheduledThreadPoolDefaultThreadFactorycallable
newScheduledThreadPoolnewSingleThreadScheduledExecutorPrivilegedThreadFactoryprivilegedCallable
newSingleThreadExecutorunconfigurableScheduledExecutorServiceprivilegedCallableUsingCurrentClassLoader
newCachedThreadPool

ThreadFactory

public interface ThreadFactory

  1. 根据需要创建新线程的对象。使用线程工厂就无需再手工编写对 new Thread 的调用了，从而允许应用程序使用特殊的线程子类、属性等等。
  2. Executors对其提供支持：DefaultThreadFactory和PrivilegedThreadFactory。
public interface ThreadFactory {
    /**
     * 构造一个新 Thread。
     */
    Thread newThread(Runnable r);
}
Executors.DefaultThreadFactory
    static class DefaultThreadFactory implements ThreadFactory {
        static final AtomicInteger poolNumber = new AtomicInteger(1);
        final ThreadGroup group;
        final AtomicInteger threadNumber = new AtomicInteger(1);
        final String namePrefix;

        DefaultThreadFactory() {
            SecurityManager s = System.getSecurityManager();
            group = (s != null)? s.getThreadGroup() :
                                 Thread.currentThread().getThreadGroup();
            namePrefix = "pool-" +
                          poolNumber.getAndIncrement() +
                         "-thread-";
        }

        public Thread newThread(Runnable r) {
            Thread t = new Thread(group, r,
                                  namePrefix + threadNumber.getAndIncrement(),
                                  0);
            if (t.isDaemon())
                t.setDaemon(false);
            if (t.getPriority() != Thread.NORM_PRIORITY)
                t.setPriority(Thread.NORM_PRIORITY);
            return t;
        }
    }

从源码看出DefaultThreadFactory就是创建一个普通的线程，非守护线程，优先级为5。
新线程具有可通过 pool-N-thread-M 的 Thread.getName() 来访问的名称，其中 N 是此工厂的序列号， M 是此工厂所创建线程的序列号。

Executors.PrivilegedThreadFactory
    static class PrivilegedThreadFactory extends DefaultThreadFactory {
        private final ClassLoader ccl;
        private final AccessControlContext acc;

        PrivilegedThreadFactory() {
            super();
            this.ccl = Thread.currentThread().getContextClassLoader();
            this.acc = AccessController.getContext();
            acc.checkPermission(new RuntimePermission("setContextClassLoader"));
        }

        public Thread newThread(final Runnable r) {
            return super.newThread(new Runnable() {
                public void run() {
                    AccessController.doPrivileged(new PrivilegedAction<Object>() {
                        public Object run() {
                            Thread.currentThread().setContextClassLoader(ccl);
                            r.run();
                            return null;
                        }
                    }, acc);
                }
            });
        }

    }

从源码看出，PrivilegedThreadFactory extends DefaultThreadFactory从而具有与 defaultThreadFactory() 相同设置的线程。但增加了两个特性：ClassLoader和AccessControlContext，从而使运行在此类线程中的任务具有与当前线程相同的访问控制和类加载器。