介绍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
创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。示例代码如下：
因为线程池大小为3，每个任务输出index后sleep 2秒，所以每两秒打印3个数字。
定长线程池的大小最好根据系统资源进行设置。如Runtime.getRuntime().availableProcessors()。可参考PreloadDataCache。
(3) newScheduledThreadPool
创建一个定长线程池，支持定时及周期性任务执行。延迟执行示例代码如下：
表示延迟3秒执行。
定期执行示例代码如下：
表示延迟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线程响应的操作。
***********************************************************************************************************
CountDownLatch，api 文档：http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/CountDownLatch.html
A synchronization aid that allows one or more threads to wait until a set of operations being performed in other threads completes.
假设我们要打印1-100，最后再输出“Ok“。1-100的打印顺序不要求统一，只需保证“Ok“是在最后出现即可。
解决方案：我们定义一个CountDownLatch，然后开10个线程分别打印（n-1）*10+1至（n-1）*10+10。主线程中调用await 方法等待所有线程的执行完毕，每个线程执行完毕后都调用countDown方法。最后再await返回后打印“Ok”。
package thread;
import java.util.concurrent.CountDownLatch;
public class TestCountDownLatch {
private static final int N = 10;
public static void main(String[] args) throws InterruptedException {
CountDownLatch doneSignal = new CountDownLatch(N);
CountDownLatch startSignal = new CountDownLatch(1);// 开始执行信号
for (int i = 1; i <= N; i++) {
new Thread(new Worker(i, doneSignal, startSignal)).start();// 线程启动了
}
System.out.println("begin------------");
startSignal.countDown();// 开始执行啦
doneSignal.await();// 等待所有的线程执行完毕
System.out.println("Ok");
}
static class Worker implements Runnable {
private final CountDownLatch doneSignal;
private final CountDownLatch startSignal;
private int beginIndex;
Worker(int beginIndex, CountDownLatch doneSignal,
CountDownLatch startSignal) {
this.startSignal = startSignal;
this.beginIndex = beginIndex;
this.doneSignal = doneSignal;
}
public void run() {
try {
startSignal.await(); // 等待开始执行信号的发布
beginIndex = (beginIndex - 1) * 10 + 1;
for (int i = beginIndex; i < beginIndex + 10; i++) {
System.out.println(i);
}
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
doneSignal.countDown();
}
}
}
}
CyclicBarrier，api 文档：http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/CyclicBarrier.html
A synchronization aid that allows a set of threads to all wait for each other to reach a common barrier point. CyclicBarriers are useful in programs involving a fixed sized party of threads that must occasionally wait for each other. The barrier is called cyclic because it can be re-used after the waiting threads are released.
举一个很简单的例子，今天晚上我们哥们4个去Happy。就互相通知了一下：晚上八点准时到xx酒吧门前集合，不见不散！。有个哥们住的近，早早就到了。有的事务繁忙，刚好踩点到了。无论怎样，先来的都不能独自行动，只能等待所有人
public class TestCyclicBarrier {
public static void main(String[] args) {
//new 一个线程池
ExecutorService exec = Executors.newCachedThreadPool();
final Random random = new Random();
final CyclicBarrier barrier = new CyclicBarrier(4, new Runnable() {
@Override
public void run() {
System.out.println("大家都到齐了，开始happy去");
}
});
for (int i = 0; i < 4; i++) {
exec.execute(new Runnable() {
@Override
public void run() {
try {
Thread.sleep(random.nextInt(10000));
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName()
+ "到了，其他哥们呢");
try {
barrier.await();// 等待其他哥们
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
}
});
}
exec.shutdown();
}
}
Semaphore api：http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Semaphore.html
A counting semaphore. Conceptually, a semaphore maintains a set of permits. Each acquire() blocks if necessary until a permit is available, and then takes it. Each release() adds a permit, potentially releasing a blocking acquirer. However, no actual permit objects are used; the Semaphore just keeps a count of the number available and acts accordingly.
例如：对于某个容器，我们规定，最多只能容纳n个线程同时操作 使用信号量来模拟实现
public class TestSemaphore {
public static void main(String[] args) {
ExecutorService exec = Executors.newCachedThreadPool();
TestSemaphore t = new TestSemaphore();
final BoundedHashSet<String> set = t.getSet();
for (int i = 0; i < 3; i++) {// 三个线程同时操作add
exec.execute(new Runnable() {
public void run() {
try {
set.add(Thread.currentThread().getName());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
}
for (int j = 0; j < 3; j++) {// 三个线程同时操作remove
exec.execute(new Runnable() {
public void run() {
set.remove(Thread.currentThread().getName());
}
});
}
exec.shutdown();
}
public BoundedHashSet<String> getSet() {
return new BoundedHashSet<String>(2);// 定义一个边界约束为2的线程
}
class BoundedHashSet<T> {
private final Set<T> set;
private final Semaphore semaphore;
public BoundedHashSet(int bound) {
this.set = Collections.synchronizedSet(new HashSet<T>());//①
this.semaphore = new Semaphore(bound, true);
}
public void add(T o) throws InterruptedException {
semaphore.acquire();// 信号量控制可访问的线程数目
set.add(o);
System.out.printf("add:%s%n", o);
}
public void remove(T o) {
if (set.remove(o))
semaphore.release();// 释放掉信号量
System.out.printf("remove:%s%n", o);
}
}
}
解释① ：Collection类中提供了多个synchronizedXxx方法，该方法返回指定集合对象对应的同步对象，从而解决多线程并发访问集合时线程的安全问题。java中常用的HashSet、ArrayList、HashMap都是线程不安全的，如果多条线程访问他们，而且多于一条的线程试图修改它们，则可能出错。以下方法直接将新建的集合传给了Collections的synchronizedXxx方法，这样就直接获取它们的线程安全实现版本。
Collection c = Collections.synchronizedCollection(new ArrayList());
List l = Collections.synchronizedList(new ArrayList());
Set s = Collections.synchronizedSet(new HashSet());
Map m = Collections.synchronizedMap(new HashMap());
FutureTask api：http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/FutureTask.html
A cancellable asynchronous computation. This class provides a base implementation of Future, with methods to start and cancel a computation, query to see if the computation is complete, and retrieve the result of the computation. The result can only be retrieved when the computation has completed; the get methods will block if the computation has not yet completed. Once the computation has completed, the computation cannot be restarted or cancelled (unless the computation is invoked using runAndReset()).
应用举例：我们的算法中有一个很耗时的操作，在编程的是，我们希望将它独立成一个模块，调用的时候当做它是立刻返回的，并且可以随时取消的
public class TestFutureTask {
public static void main(String[] args) {
ExecutorService exec = Executors.newCachedThreadPool();
FutureTask<String> task = new FutureTask<String>(
new Callable<String>() {// FutrueTask的构造参数是一个Callable接口
@Override
public String call() throws Exception {
return Thread.currentThread().getName();// 这里可以是一个异步操作
}
});
try {
exec.execute(task);// FutureTask实际上也是一个线程
String result = task.get();// 取得异步计算的结果，如果没有返回，就会一直阻塞等待
System.out.printf("get:%s%n", result);
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
}
总结：FutureTask其实就是新建了一个线程单独执行，使得线程有一个返回值，方便程序的编写
Exchanger api：http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Exchanger.html
package thread;
import java.util.ArrayList;
import java.util.concurrent.Exchanger;
/**
* 1. Exchanger用于在2个线程中交换对象。
2. return_object = exchanger.exchange(exch_object)
3. 例子中Producer向ArrayList中缓慢填充随机整数，Consumer从另一个ArrayList中缓慢取出整数并输出。
4. 当Producer的ArrayList填满，并且Consumer的ArrayList为空时，2个线程才交换ArrayList。
* @author Administrator
*/
public class ExchangerTest {
private static Exchanger<ArrayList<Integer>> exchanger = null;
private static ArrayList<Integer> buffer1 = null;
private static ArrayList<Integer> buffer2 = null;
public static void main(String[] args) throws Exception {
exchanger = new Exchanger<ArrayList<Integer>>();
buffer1 = new ArrayList<Integer>(10);
buffer2 = new ArrayList<Integer>(10);
Thread pth = new ProducerThread();
Thread cth = new ConsumerThread();
pth.start();
cth.start();
Thread.sleep(60 * 1000);
System.out.println("main: interrupting threads.");
pth.interrupt();
cth.interrupt();
pth.join();
cth.join();
System.out.println("main: end.");
}
private static class ProducerThread extends Thread {
@Override
public void run() {
ArrayList<Integer> buff = buffer1;
try {
while (true) {
if (buff.size() >= 10) {
// 与consumer交换buffer.
System.out.println("producer: exchanging.");
buff = exchanger.exchange(buff);
buff.clear();
}
// 随机产生一个0-100的整数。
int x = (int) (Math.random() * 100);
buff.add(x);
System.out.println("producer: " + x);
// 随机等待0-3秒 。
int t = (int) (Math.random() * 3);
Thread.sleep(t * 1000);
}
} catch (InterruptedException e) {
System.out.println("producer: interrupted.");
}
}
}
private static class ConsumerThread extends Thread {
@Override
public void run() {
ArrayList<Integer> buff = buffer2;
try {
while (true) {
for (Integer x : buff) {
System.out.println("consumer: " + x);
// 随机等待0-3秒 。
int t = (int) (Math.random() * 3);
Thread.sleep(t * 1000);
}
// 与producer交换buffer。
System.out.println("consumer: exchanging.");
buff = exchanger.exchange(buff);
}
} catch (InterruptedException e) {
System.out.println("consumer: interrupted.");
}
}
}
}
在JDK1.5之前，我们关于定时/周期操作都是通过Timer来实现的。但是Timer有以下几种危险[JCIP]
a. Timer是基于绝对时间的。容易受系统时钟的影响。
b. Timer只新建了一个线程来执行所有的TimeTask。所有TimeTask可能会相关影响
c. Timer不会捕获TimerTask的异常，只是简单地停止。这样势必会影响其他TimeTask的执行。
如果你是使用JDK1.5以上版本，建议用ScheduledThreadPoolExecutor代替Timer。它基本上解决了上述问题。它采用相对时间，用线程池来执行TimerTask，会出来TimerTask异常。
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class TestScheduledThreadPoolExecutor {
public static void main(String[] args) {
ScheduledThreadPoolExecutor exec = new ScheduledThreadPoolExecutor(1);
exec.scheduleAtFixedRate(new Runnable() {// 每隔一段时间就触发异常
@Override
public void run() {
throw new RuntimeException();
}
}, 1000, 5000, TimeUnit.MILLISECONDS);
exec.scheduleAtFixedRate(new Runnable() {// 每隔一段时间打印系统时间，证明两者是互不影响的
@Override
public void run() {
System.out.println(System.nanoTime());
}
}, 1000, 2000, TimeUnit.MILLISECONDS);
}
}
BlockingQueue API文档：http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/BlockingQueue.html
A Queue that additionally supports operations that wait for the queue to become non-empty when retrieving an element, and wait for space to become available in the queue when storing an element.
BlockingQueue的经典用途是 生产者-消费者模式
package thread;
import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
public class BlockingQueueTest {
public static void main(String[] args) {
final BlockingQueue<Integer> queue = new LinkedBlockingQueue<Integer>(3);
final Random random = new Random();
class Producer implements Runnable {
@Override
public void run() {
while (true) {
try {
int i = random.nextInt(100);
queue.put(i);// 当队列达到容量时候，会自动阻塞的
if (queue.size() == 3) {
System.out.println("full");
Thread.sleep(1000);
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
class Consumer implements Runnable {
@Override
public void run() {
while (true) {
try {
queue.take();// 当队列为空时，也会自动阻塞
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
new Thread(new Producer()).start();
new Thread(new Consumer()).start();
}
}
DelayQueue
在现实生活中，很多DelayQueue的例子。就拿上海的SB会来说明，很多国家地区的开馆时间不同。你很早就来到园区，然后急急忙忙地跑到一些心仪的馆区，发现有些还没开，你吃了闭门羹。
仔细研究DelayQueue，你会发现它其实就是一个PriorityQueue的封装（按照delay时间排序），里面的元素都实现了Delayed接口，相关操作需要判断延时时间是否到了。
在实际应用中，有人拿它来管理跟实际相关的缓存、session等
下面我就通过 “上海SB会的例子来阐述DelayQueue的用法”
package thread;
import java.util.Random;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
public class TestDelayQueue {
private class Stadium implements Delayed {
long trigger;
public Stadium(long i) {
trigger = System.currentTimeMillis() + i;
}
@Override
public long getDelay(TimeUnit arg0) {
long n = trigger - System.currentTimeMillis();
return n;
}
@Override
public int compareTo(Delayed arg0) {
return (int) (this.getDelay(TimeUnit.MILLISECONDS) - arg0
.getDelay(TimeUnit.MILLISECONDS));
}
public long getTriggerTime() {
return trigger;
}
}
public static void main(String[] args) throws Exception {
Random random = new Random();
DelayQueue<Stadium> queue = new DelayQueue<Stadium>();
TestDelayQueue t = new TestDelayQueue();
for (int i = 0; i < 5; i++) {
queue.add(t.new Stadium(random.nextInt(30000)));
}
Thread.sleep(2000);
while (true) {
Stadium s = queue.take();// 延时时间未到就一直等待
if (s != null) {
System.out.println(System.currentTimeMillis()
- s.getTriggerTime());// 基本上是等于0
}
if (queue.size() == 0)
break;
}
}
}
总结：适用于需要延时操作的队列管理
SynchronousQueue API：http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/SynchronousQueue.html
这个队列其实是BlockingQueue的一种实现。每个插入操作必须等待另一个线程的对应移除操作，反之亦然。它给我们提供了在线程之间交换单一元素的极轻量级方法
应用举例：我们要在多个线程中传递一个变量。
package thread;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.SynchronousQueue;
public class TestSynchronousQueue {
class Producer implements Runnable {
private BlockingQueue<String> queue;
List<String> objects = Arrays.asList("one", "two", "three");
public Producer(BlockingQueue<String> q) {
this.queue = q;
}
@Override
public void run() {
try {
for (String s : objects) {
queue.put(s);// 产生数据放入队列中
System.out.printf("put:%s%n", s);
}
queue.put("Done");// 已完成的标志
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
class Consumer implements Runnable {
private BlockingQueue<String> queue;
public Consumer(BlockingQueue<String> q) {
this.queue = q;
}
@Override
public void run() {
String obj = null;
try {
while (!((obj = queue.take()).equals("Done"))) {
System.out.println(obj);// 从队列中读取对象
Thread.sleep(3000); // 故意sleep，证明Producer是put不进去的
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
public static void main(String[] args) {
BlockingQueue<String> q = new SynchronousQueue<String>();
TestSynchronousQueue t = new TestSynchronousQueue();
new Thread(t.new Producer(q)).start();
new Thread(t.new Consumer(q)).start();
}
}
总结：SynchronousQueue主要用于单个元素在多线程之间的传递
