实战Concurrent(1)

来自http://www.iteye.com/topic/363625

 

编写多线程的程序一直都是一件比较麻烦的事情，要考虑很多事情，处理不好还会出很多意想不到的麻烦。加上现在很多开发者接触到的项目都是打着企业级旗号的B/S项目，大多数人都很少涉及多线程，这又为本文的主角增加了一份神秘感。

 

讲到Java多线程，大多数人脑海中跳出来的是Thread、Runnable、synchronized……这些是最基本的东西，虽然已经足够强大，但想要用好还真不容易。从JDK 1.5开始，增加了java.util.concurrent包，它的引入大大简化了多线程程序的开发（要感谢一下大牛Doug Lee）。

 

java.util.concurrent包分成了三个部分，分别是java.util.concurrent、java.util.concurrent.atomic和java.util.concurrent.lock。内容涵盖了并发集合类、线程池机制、同步互斥机制、线程安全的变量更新工具类、锁等等常用工具。

 

为了便于理解，本文使用一个例子来做说明，交代一下它的场景：

假设要对一套10个节点组成的环境进行检查，这个环境有两个入口点，通过节点间的依赖关系可以遍历到整个环境。依赖关系可以构成一张有向图，可能存在环。为了提高检查的效率，考虑使用多线程。

 

1、Executors

通过这个类能够获得多种线程池的实例，例如可以调用newSingleThreadExecutor()获得单线程的ExecutorService，调用newFixedThreadPool()获得固定大小线程池的ExecutorService。拿到ExecutorService可以做的事情就比较多了，最简单的是用它来执行Runnable对象，也可以执行一些实现了Callable<T>的对象。用Thread的start()方法没有返回值，如果该线程执行的方法有返回值那用ExecutorService就再好不过了，可以选择submit()、invokeAll()或者invokeAny()，根据具体情况选择合适的方法即可。

package
 service;

import
 java.util.ArrayList;
import
 java.util.List;
import
 java.util.concurrent.ExecutionException;
import
 java.util.concurrent.ExecutorService;
import
 java.util.concurrent.Executors;
import
 java.util.concurrent.Future;
import
 java.util.concurrent.TimeUnit;

/**
 * 线程池服务类
 *
 * @author DigitalSonic
 */
public
 
class
 ThreadPoolService {
    /**
     * 默认线程池大小
     */
    public
 
static
 
final
 
int
  DEFAULT_POOL_SIZE    = 
5
;

    /**
     * 默认一个任务的超时时间，单位为毫秒
     */
    public
 
static
 
final
 
long
 DEFAULT_TASK_TIMEOUT = 
1000
;

    private
 
int
              poolSize             = DEFAULT_POOL_SIZE;
    private
 ExecutorService  executorService;

    /**
     * 根据给定大小创建线程池
     */
    public
 ThreadPoolService(
int
 poolSize) {
        setPoolSize(poolSize);
    }

    /**
     * 使用线程池中的线程来执行任务
     */
    public
 
void
 execute(Runnable task) {
        executorService.execute(task);
    }

    /**
     * 在线程池中执行所有给定的任务并取回运行结果，使用默认超时时间
     *
     * @see #invokeAll(List, long)
     */
    public
 List<Node> invokeAll(List<ValidationTask> tasks) {
        return
 invokeAll(tasks, DEFAULT_TASK_TIMEOUT * tasks.size());
    }

    /**
     * 在线程池中执行所有给定的任务并取回运行结果
     *
     * @param timeout 以毫秒为单位的超时时间，小于0表示不设定超时
     * @see java.util.concurrent.ExecutorService#invokeAll(java.util.Collection)
     */
    public
 List<Node> invokeAll(List<ValidationTask> tasks, 
long
 timeout) {
        List<Node> nodes = new
 ArrayList<Node>(tasks.size());
        try
 {
            List<Future<Node>> futures = null
;
            if
 (timeout < 
0
) {
                futures = executorService.invokeAll(tasks);
            } else
 {
                futures = executorService.invokeAll(tasks, timeout, TimeUnit.MILLISECONDS);
            }
            for
 (Future<Node> future : futures) {
                try
 {
                    nodes.add(future.get());
                } catch
 (ExecutionException e) {
                    e.printStackTrace();
                }
            }
        } catch
 (InterruptedException e) {
            e.printStackTrace();
        }
        return
 nodes;
    }

    /**
     * 关闭当前ExecutorService
     *
     * @param timeout 以毫秒为单位的超时时间
     */
    public
 
void
 destoryExecutorService(
long
 timeout) {
        if
 (executorService != 
null
 && !executorService.isShutdown()) {
            try
 {
                executorService.awaitTermination(timeout, TimeUnit.MILLISECONDS);
            } catch
 (InterruptedException e) {
                e.printStackTrace();
            }
            executorService.shutdown();
        }
    }

    /**
     * 关闭当前ExecutorService，随后根据poolSize创建新的ExecutorService
     */
    public
 
void
 createExecutorService() {
        destoryExecutorService(1000
);
        executorService = Executors.newFixedThreadPool(poolSize);
    }

    /**
     * 调整线程池大小
     * @see #createExecutorService()
     */
    public
 
void
 setPoolSize(
int
 poolSize) {
        this
.poolSize = poolSize;
        createExecutorService();
    }
}

 

这里要额外说明一下invokeAll()和invokeAny()方法。前者会执行给定的所有Callable<T>对象，等所有任务完成后返回一个包含了执行结果的List<Future<T>>，每个Future.isDone()都是true，可以用Future.get()拿到结果；后者只要完成了列表中的任意一个任务就立刻返回，返回值就是执行结果。

还有一个比较诡异的地方

本代码是在JDK 1.6下编译测试的，如果在JDK 1.5下测试，很可能在invokeAll和invokeAny的地方出错。明明ValidationTask实现了 Callable<Node>，可是它死活不认，类型不匹配，这时可以将参数声明由List<ValidationTask>改为 List<Callable<Node>>。
造成这个问题的主要原因是两个版本中invokeAll和invokeAny的方法签名不同，1.6里是invokeAll(Collection<? extends Callable<T>> tasks)，而1.5里是invokeAll(Collection<Callable<T>> tasks)。网上也有人遇到类似的问题(invokeAll() is not willing to acept a Collection<Callable<T>> )。

 

和其他资源一样，线程池在使用完毕后也需要释放，用shutdown()方法可以关闭线程池，如果当时池里还有没有被执行的任务，它会等待任务执行完毕，在等待期间试图进入线程池的任务将被拒绝。也可以用shutdownNow()来关闭线程池，它会立刻关闭线程池，没有执行的任务作为返回值返回。