Java多线程编程简明教程
缘起
关于多线程编程的教程汗牛充栋了，比如阿里集团内部就有一粟、高铁等大牛的讲座，更不用说有清英同学专门创建了并发编程网站来专注于这件事情。专门讲Java并发开发的书籍也是相当丰富了。
我们举个例子，典型的一本Java多线程开发的教材是这样写的，上来就是介绍如何创建线程，然后再讲线程安全，线程之间如何做同步和通信，接着才是线程池和执行器，最后是线程安全的数据结构。
这样写当然从技术上讲是没问题的，不过问题在于，门槛太高了。假如读者的时间短，只看完创建线程这一章就开始照猫画虎写多线程的代码，于是，多线程竞态的问题在那里等着呢。假如他多再多看一些，学会了线程的同步互斥，而且也解决了竞态和死锁的问题，他仍然不会想到要使用线程池，在使用容器的时候也不免会遇到一些坑。
但是，真的要经过这么专业的训练才可以写一个简单的多任务的么？比如对于很多任务来说，只是需要静静地在后台去执行一个任务而己，在运行过程中，并不需要同步访问共享，最后只是返回一个结果就好了。对于一些更复杂的问题，可以采用分治法处理，规划得当的话并不会出现访问冲突。对于前面说的第一种情况，使用JDK 5引入的Future模式已经足够了，而对于后一种的情况，使用JDK 7中引入的Fork-Join框架就可以很好的解决。这两种模式都是线程安全的，根本不共享状态有什么不安全的呢。而学会了这两个模式之后，习惯成自然地，将来再学习线程也会习惯性地使用线程池，不会引入创建和销毁大量线程的消耗。
学会了这两个模式之后，我们进一步再学习如何避免共享状态，如何封闭状态。如何通过使用安全的数据结构进行通信去共享状态。
如果以上都实在解决不了问题，最后再老老实实地学习Java的内存模型，再去实现传统上一开始就讲的创建线程这一套机制。
按照传统的思路，讲到创建线程的时候，初学者对于Java内存模型，对于在多线程情况下的数据结构，对于线程池这些完全没有概念，一切都按照单任务的经验来写代码，于是造成了很多问题而不知。而现在是在不满足前面的常用模式的情况下才选择这条路，读者已经非常清楚他在做什么，要承担什么样的风险，以及如何降低这个风险。
最后，他山之石，可以攻玉。学习借鉴其他语言和模型的成功经验和教训，可能比只懂线程的同步和互斥有更广的思路。
很多书籍的编排已经透出了不少新意，比如方腾飞大牛的《Java并发编程的艺术》，在讲线程之前，先讲一章庞大的Java内存模型。这样在学习线程的时候，对于可见不可见，执行顺序等等已经有很清楚的概念了。稍嫌不足的是容器出场太晚，如果没看到这章就开始写的话，对于独立变量应该是没问题了，一旦跟容器打交道了，还容易吃亏，我就吃过这样的亏。而《Java Concurrency in Practice》这点做得就更好一些。
我的教程希望能够，不管在哪一部分看完之后暂时中断了，学到的体系也是相对完整的，力争避免会了线程，但是写出一堆问题的代码。哪怕学习了如何危险还不知道如何写，也比不知道危险要好，至少知道要学完才能用。
Future模式
Future模式5分钟教程
Future模式用于的场景是，不急于马上就拿到结果的任务，可以放到后台先做着，然后主线程继续忙别的去。等主线程需要用后台任务的结果了，再去从Future去拿结果。如果Future任务已经做完了，当然皆大欢喜，马上使用结果，代码逻辑继续往下跑；如果还没做完，至少等待的时间比一直干等着强。总之，稳赚不赔，何乐而不为。如果这时候那个结果已经不重要了，取消了也没问题。
这里面没有竞态，没有共享哪来的竞态。同样，不需要懂锁是什么东西。
使用Future模式需要4个步骤：
- 构造一个Callable接口的实现，在其中写在后台要实现的逻辑
- 构造一个线程池执行器，提交执行
- 主线程继续忙自己的
- 想要用这个Future的时候，调用FutureTask对象的get()方法去获取值
我们举个简单的例子来看一下：
public class AsyncTaskSimple {
public static class Result implements Callable<String>{
@Override
public String call() throws Exception {
return doRealLogic();
}
private String doRealLogic(){
//Here to do the background logic
return new String("Done");
}
}
public static void main(String[] args) {
FutureTask<String> future = new FutureTask<String>(new Result());
ExecutorService executor = Executors.newFixedThreadPool(1);
executor.submit(future);
someThingToDo();
try {
String s = future.get();
System.out.println("The result is:"+s);
}catch (InterruptedException e){
//Deal with InterruptedExcpeiotn
}catch(ExecutionException ee){
//Deal with ExecutionException
}
}
private static void someThingToDo(){
//Main thread logic
}
}
从上面的例子代码可以看到，都是填空题，逻辑很简单：
- 需要的返回值是什么类型，就实现Callable<类型>接口，然后在call()方法里实现后台逻辑。
- 用Callable接口的实现类去构造一个FutureTask对象。
- 构建ExecutorService对象，决定用哪种线程池。
- 提交任务。
- 让后台任务在空中飞一会儿，前台该干嘛干嘛。
- 前台的事儿忙完了，去读取后台的结果。
就这么简单，线程是什么？锁是什么？如何防止竞态和死锁，这些统统用不到。
线程池
关于线程池，我们只讲三种最基本的，其余的细节后面详细讲。
这三种是：只有一个线程，固定数目的，按需分配的.
分别对应了Executors类的三个静态方法：
- public static ExecutorService newSingleThreadExecutor ():就建一个，大家排队按顺序来
- public static ExecutorService newFixedThreadPool (int nThreads)：定义固定nThreads个大小的线程池。
- public static ExecutorService newCachedThreadPool ():最省事了，完全由系统自己管理。能重用就重用，不能重用就建新的线程。
好，Future模式就讲完了，大家可以上手写代码了。无锁、无竞态、使用了线程池。从创建线程开始学的同学们还要学习Java内存模型，如何同步，如何互斥，如何避免死锁等等，这些我们都暂时不用学，也不需要线程安全的数据结构，因为传进去的参数和返回的结果都是只读的，没有多线程去抢着写它们。
揭开Android AsyncTask的面纱
下面我们说说Android中叫做AsyncTask的东西，其实就是在Future上做的一个简要封装。
我们一起看下，有Android经验的同学正好借已有的经验来加深一下印象。
定义
public abstract class AsyncTask<Params, Progress, Result> {
这个类有三个泛型参数，Params, Progress, Result，这里先记住，我们后面会看到它们各起什么作用。
构造方法
我们把无关的代码省略掉，核心逻辑一共就两句话，一句是调用子类的doInBackground方法，一句是调用postResult方法将返回值返回。
288 public AsyncTask() {
289 mWorker = new WorkerRunnable<Params, Result>() {
290 public Result call() throws Exception {
...
295 Result result = doInBackground(mParams);
...
297 return postResult(result);
298 }
299 };
这个WorkerRunnable是实现了Callable<Result>的一个抽象类.
private static abstract class WorkerRunnable<Params, Result> implements Callable<Result> {
Params[] mParams;
}
通过这一步，完成了我们前面学习的Callable接口对象的创建。按照我们前面学习的，下一步该构造FutureTask对象了，AsyncTask的构造方法就是这么做的！用刚才生成的mWorker对象去构造mFuture对象。
这里与前面所讲的有一点不同的是，前面我们是处理完了由主线程主动查询Future是否结束，而AsyncTask是在任务结束之后，也就是重载done()方法，在这里面再去调用get()方法去获取Future的值，再将其主动发出去。
301 mFuture = new FutureTask<Result>(mWorker) {
302 @Override
303 protected void done() {
304 try {
305 postResultIfNotInvoked(get());
306 } catch (InterruptedException e) {
307 android.util.Log.w(LOG_TAG, e);
308 } catch (ExecutionException e) {
309 throw new RuntimeException("An error occurred while executing doInBackground()",
310 e.getCause());
311 } catch (CancellationException e) {
312 postResultIfNotInvoked(null);
313 }
314 }
315 };
316 }
上面的调用的get()方法一点新意也没有，就是mFuture.get()的简单封装，我们看代码：
497 public final Result get() throws InterruptedException, ExecutionException {
498 return mFuture.get();
499 }
只有到了将结果通知出来这个过程是跟Android相关的，用到了Android的消息队列。
318 private void postResultIfNotInvoked(Result result) {
319 final boolean wasTaskInvoked = mTaskInvoked.get();
320 if (!wasTaskInvoked) {
321 postResult(result);
322 }
323 }
324
325 private Result postResult(Result result) {
326 @SuppressWarnings("unchecked")
327 Message message = getHandler().obtainMessage(MESSAGE_POST_RESULT,
328 new AsyncTaskResult<Result>(this, result));
329 message.sendToTarget();
330 return result;
331 }
executeOnExecutor
我们再看看mFuture是何时被执行的，一句话，是在executeOnExecutor中。这个方法是在主线程中运行的，所以可以先安全地运行onPreExecute()，而onPostExecute()则要是在Handler里面处理了。
587 @MainThread
588 public final AsyncTask<Params, Progress, Result> executeOnExecutor(Executor exec,
589 Params... params) {
...
604 onPreExecute();
605
606 mWorker.mParams = params;
607 exec.execute(mFuture);
608
609 return this;
610 }
onPostExecute和通知进度
这个都是在Handler中实现的了，结束了就onPostExecute，还没完就通知下进度。
656 private static class InternalHandler extends Handler {
657 public InternalHandler() {
658 super(Looper.getMainLooper());
659 }
660
661 @SuppressWarnings({"unchecked", "RawUseOfParameterizedType"})
662 @Override
663 public void handleMessage(Message msg) {
664 AsyncTaskResult<?> result = (AsyncTaskResult<?>) msg.obj;
665 switch (msg.what) {
666 case MESSAGE_POST_RESULT:
667 // There is only one result
668 result.mTask.finish(result.mData[0]);
669 break;
670 case MESSAGE_POST_PROGRESS:
671 result.mTask.onProgressUpdate(result.mData);
672 break;
673 }
674 }
675 }
finish方法根据是否cancel决定调用哪一个，正常情况下是onPostExecute
647 private void finish(Result result) {
648 if (isCancelled()) {
649 onCancelled(result);
650 } else {
651 onPostExecute(result);
652 }
653 mStatus = Status.FINISHED;
654 }
Fork-Join模式
说起Fork-Join模式,我们不免联想起了Map-Reduce.它们的原理都是分治法,就是将一个大问题划分成若干个小问题,如果这些小问题之间互相不影响的话,就可以并发去执行. 最后,统一将各小问题的结果汇总起来,就是这个大问题的结果.
这个任务最适合处理像一棵树一样的问题.
ForkJoinPool
Fork-Join模式不再是只管一个后台作务,而是有多个任务并发执行. 这时我们前面学到的简单的线程池执行器的功能就显得不足了.这时候JDK 7开始为我们提供了ForkJoinPool.
ForkJoinPool不但自动计算开多大的线程池合适,而且提供了称为工作窃取算法的算法来管理这些任务. 如果有的线程空闲, ForkJoinPool会从其它线程的队列尾中窃取一个任务给空闲线程来运行.而正常的线程是从任务队列头中取任务,二者不会有冲突.
RecusiveTask
如同FutureTask一样,Fork-Join模式也有自己的Task类ForkJoinTask. 不过一般我们都是从ForkJoinTask的子类RecursiveTask来继承. 通过重载RecursiveTask类的compute方法,来实现Fork-Join的逻辑.
在compute方法里, 要实现两件事, 顾名思义, Fork-Join就是要先fork出RecursiveTask对象的子任务,然后将它们join在一起.
Fork-Join模式10分钟速成教程
我们先写个copy二叉树结构的简单任务学习一下如何利用Fork-Join框架来实现功能.
先实现一个最简单的二叉树节点,带左右孩子,一个字符串吧:
public class BinaryTree {
public static class Node{
public Node leftChild;
public Node rightChild;
public String content;
public Node(String ct){
content = ct;
}
}
然后实现一个RecursiveTask的子类,重载它的compute方法.
public static class NodeCopyTask extends RecursiveTask<Node>{
Node mNode;
public NodeCopyTask(Node node){
mNode = node;
}
@Override
protected Node compute() {
if(mNode==null)
return null;
下面我们开始实现分叉, 对于左右子树分别fork出一个子任务. 这两个子任务又会分叉出它的的子任务,直至结束.
NodeCopyTask taskLeft = new NodeCopyTask(mNode.leftChild);
taskLeft.fork();
NodeCopyTask taskRight = new NodeCopyTask(mNode.rightChild);
taskRight.fork();
fork之后, 任务就在后台开始运行了. 这时候我们开始构造我们的左右子树的父节点:
Node node = new Node(mNode.content);
实际问题中一般不会这么简单.主线任务完成了之后,就是等待子任务交活儿,将它们组装在一起:
node.leftChild = taskLeft.join();
node.rightChild = taskRight.join();
return node;
}
}
核心功能实现完了,下面我们写个主函数让它运行起来吧. 先构造一个被复制的对象.
public static void main(String[] args){
Node node = new Node("Hello,Fork-Join");
node.leftChild = new Node("Left");
node.rightChild = new Node("Right");
下面我们前面介绍的主角之一 - ForkJoinPool粉墨登场. 没什么复杂的设置,直接new一个就好:
ForkJoinPool forkJoinPool = new ForkJoinPool();
ForkJoinPool有了之后, 再创建一个我们的RecursiveTask的对象, 然后调用ForkJoinPool的submit方法将其提交, 这又是一个Future模式了. 最后我们通过这个FutureTask的get方法获取结果就一切OK了.
NodeCopyTask task = new NodeCopyTask(node);
Future<Node> future = forkJoinPool.submit(task);
try {
Node node2 = future.get();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
}
整理一下上面的步骤:
- 实现一个RecursiveTask的子类,重载compute方法实现fork-join逻辑
- 合理划分任务,调用递归的RecursiveTask子类,fork出每个子任务
- 通过join方法获取子任务的值,并将它们组合到一起
- 构造ForkJoinPool线程池
- 创建第一步的子类的对象,通过Future模式,提交到ForkJoinPool线程中运行
- 获取Future的值,即可得到Fork-Join的结果.
总结一下,把刚才拆散的代码整合在一起:
public class BinaryTree {
public static class Node{
public Node leftChild;
public Node rightChild;
public String content;
public Node(String ct){
content = ct;
}
}
public static class NodeCopyTask extends RecursiveTask<Node>{
Node mNode;
public NodeCopyTask(Node node){
mNode = node;
}
@Override
protected Node compute() {
if(mNode==null)
return null;
NodeCopyTask taskLeft = new NodeCopyTask(mNode.leftChild);
taskLeft.fork();
NodeCopyTask taskRight = new NodeCopyTask(mNode.rightChild);
taskRight.fork();
Node node = new Node(mNode.content);
node.leftChild = taskLeft.join();
node.rightChild = taskRight.join();
return node;
}
}
public static void main(String[] args){
//TODO: construct a real tree
Node node = new Node("Hello,Fork-Join");
node.leftChild = new Node("Left");
node.rightChild = new Node("Right");
ForkJoinPool forkJoinPool = new ForkJoinPool();
NodeCopyTask task = new NodeCopyTask(node);
Future<Node> future = forkJoinPool.submit(task);
try {
Node nodeNew = future.get();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
注意事项
- 现在这个阶段,暂时先不要共享内存,通过复制成不变的对象传递给子任务.返回值也创建新对象,当然可以使用对象池等技术.
- 暂时不要使用跨任务的容器,我们还没有经过相关的训练,时机还不成熟.
- 暂时不要使用其他的同步机制,我们的知识储备暂时还不够.
- 要注意任务中的异常会接收不到,一定在任务中处理好可能出现的异常. 否则发生了异常,在主任务中却收不到,会感到很奇怪.
- 注意I/O操作,建议目前阶段在Fork-Join之前将I/O操作提前做好.
尽管有一些限制,但是Fork-Join框架还是给我们带来了很大的便利. 按照Fork-Join设计好的代码,在将来计算核数增加时,会自动给我们的代码获得性能提高.
另外需要说明的一点是,如果用并行花得代价大的话, 可以先做一个判断, 在这样的情况下保持串行.
不变模式
在结束这个快餐教程之前,我们得再次强调一下内存共享的风险. 请初学的同学们一定要重视起来.目前我们还没有学习Java对象模型和容器的安全用法, 所以目前阶段最安全的就是不共享任何状态.
只读的对象是不会引起线程安全问题的.我们所有的跨任务的数据传递,暂时都只传递不变的对象.
这样的限制可能会带来一些不便和一些性能损失.但是,它是线程安全的,对于开发人员是种投入小见效快的好事情. 如果暂时还不能满意你的需求,我们会继续学习,从此开始,没有快餐式的速成教程了,我们要经过一段非常扎实的训练.
另一个例子
下面我们看下官方的例子, 通过这个例子我们想说明如果当前线程无事可做, 可以fork一部分, 在当前任务中执行另一部分. 如下面所示, 它将一部分fork在f1去执行, 另外一部分f2在当前任务中执行.
class Fibonacci extends RecursiveTask {
final int n;
Fibonacci(int n) { this.n = n;
Integer compute() {
if (n <= 1)
return n;
Fibonacci f1 = new Fibonacci(n - 1);
f1.fork();
Fibonacci f2 = new Fibonacci(n - 2);
return f2.compute() + f1.join();
}
}}
Android的特别注意事项
请大家注意,Java中的Fork-Join并没有办法处理Android的UI线程等问题, 如果需要运行在UI线程, 区分主线程和工作线程等, 还请参考上节我们分析AsyncTask中的做法, 该使用Handler的还是要用Handler. 后面我们还会详情说细节.
