在项目开发的时候，复杂业务场景中经常用到并发编程。有时候需要获取并发执行结果，或者捕获多线程中的异常，这个时候用Runnable任务就不行了，而需要用到Callable。本文通过一个简单的实例来探讨一下Callable结合Executor框架的用法。

需求

假设系统中有一个模块，需要从另一个模块调用REST API接口查询用户（Person）的信息（通过用户ID查询），根据业务需要，会遇到一次性查询成千上万个用户信息的场景，需要进行并发查询，并获取查询结果。

准备工作

• Person model数据类结构
  为了简单起见，假设用户信息类只有两个字段：id和name，都为String类型。

package com.executor.model;

/**
 * 用户信息数据类
 * @author Administrator
 *
 */
public class Person {
    private String id;
    private String name;
    
    public Person(){
        
    }
    
    public Person(String id, String name) {
        this.id = id;
        this.name = name;
    }
    public String getId() {
        return id;
    }
    public void setId(String id) {
        this.id = id;
    }
    public String getName() {
        return name;
    }
    public void setName(String name) {
        this.name = name;
    }

    @Override
    public String toString() {
        return "Person [id=" + id + ", name=" + name + "]";
    }
}

• 查询用户信息的REST API接口封装的工具类PersonRestAPIUtil
  queryPersonFromAPI方法模拟调用REST接口查询用户信息，每次查询需要耗时300毫秒。

package com.executor.util;

import com.executor.model.Person;

/**
 * RestAPI工具类
 * 
 * @author Administrator
 *
 */
public class PersonRestAPIUtil {
    /**
     * 模拟调Rest API查询用户信息
     * 
     * @param id
     * @return
     */
    public static Person queryPersonFromAPI(String id) {
        Person p = new Person();
        
        // 模拟调接口耗时，300毫秒
        try {
            Thread.sleep(300);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        
        // 模拟查询到的用户信息，并返回
        p.setId(id);
        p.setName("person#" + id);
        return p;
    }
}

总体步骤梳理

主要是四步曲：

• 创建Callable任务类
• 构建Callable任务列表
• 用Executor线程池并发执行多个任务
• 获取任务执行结果

创建Callable任务类

在执行并发任务之前，需要通过创建一个实现Callable接口的任务类来定义每个任务具体的执行业务逻辑。对于本文的需求，名为QueryPersonTask的任务类来查询单个用户的信息，QueryPersonTask类继承自Callable<Person>。

package com.executor.task;

import java.util.concurrent.Callable;

import com.executor.model.Person;
import com.executor.util.PersonRestAPIUtil;

/**
 * 查询单个用户信息任务类
 * 
 * @author Administrator
 *
 */
public class QueryPersonTask implements Callable<Person> {

    private String id;

    public QueryPersonTask(String id) {
        this.id = id;
    }

    /**
     * call方法是任务执行的主体，
     */
    @Override
    public Person call() throws Exception {
        return PersonRestAPIUtil.queryPersonFromAPI(this.id);
    }

}

构建Callable任务列表

假设有一批用户的ID是知道的，存在一个名为ids的List<String>列表中，下面构建查询这多个用户信息的Callable任务列表：

// 构建查询多个用户信息的Callable任务列表
        List<Callable<Person>> tasks = new ArrayList<Callable<Person>>();
        for (String id : ids) {
            QueryPersonTask task = new QueryPersonTask(id);
            tasks.add(task);
        }

用Executor线程池并发执行多个任务

上一步已经构建好了任务列表，下面在线程池中并发执行这些任务。

• 创建线程池

ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors
                .newFixedThreadPool(100);

• 执行任务

        List<Future<Person>> futureResults = new ArrayList<Future<Person>>();
        try {
            futureResults = executor.invokeAll(tasks);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

可以看到任务执行结果存储在了一个List<Future<Person>>类型的futureResults对象中。

获取任务执行结果

        // 从futureResults中获取并解析出Person列表
        List<Person> persons = new ArrayList<Person>();
        for (Future<Person> ret : futureResults) {
            Person p;
            try {
                p = ret.get(); // get()方法会阻塞等到，直到获取到结果为止
                if (null != p) {
                    persons.add(p);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
                throw new Exception("InterruptedException occurs.");
            } catch (ExecutionException e) {
                e.printStackTrace();
                throw new Exception("ExecutionException occurs.");
            }
        }

最终任务执行结果被解析为了List<Person>，这也就是我们最终想要的数据。

性能对比：并发与单线程

下面写一个PersonBusiness业务类，来测试一下单线程与并发的性能对比：

package com.executor.business;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;

import com.executor.model.Person;
import com.executor.task.QueryPersonTask;
import com.executor.util.PersonRestAPIUtil;

public class PersonBusiness {
    public static void main(String[] args) throws Exception {
        // 构建20个用户的ID列表
        List<String> ids = new ArrayList<String>();
        for (int i = 1; i <= 20; i++) {
            ids.add(String.valueOf(i));
        }
        
        // 单线程查询
        singleThreadQueryPersons(ids);
        // 并发查询
        concurrentQueryPersons(ids);
    }

    /**
     * 单线程查询多个用户信息
     * 
     * @param ids
     * @return
     */
    private static List<Person> singleThreadQueryPersons(List<String> ids) {
        // 计时开始
        long start = System.currentTimeMillis();

        List<Person> persons = new ArrayList<Person>();

        // 遍历每个用户ID，依次查询用户信息
        for (String id : ids) {
            Person p = PersonRestAPIUtil.queryPersonFromAPI(id);
            persons.add(p);
        }

        // 计时结束
        long end = System.currentTimeMillis();
        System.out.println("查询" + ids.size()
                + "个用户，singleThreadQueryPersons方法共耗时：" + (end - start) + "毫秒");

        return persons;
    }

    private static List<Person> concurrentQueryPersons(List<String> ids)
            throws Exception {
        // 计时开始
        long start = System.currentTimeMillis();

        // 1. 构建查询多个用户信息的Callable任务列表
        List<Callable<Person>> tasks = new ArrayList<Callable<Person>>();
        for (String id : ids) {
            QueryPersonTask task = new QueryPersonTask(id);
            tasks.add(task);
        }

        // 2. 并发执行多个任务，并获取并发执行结果
        // 2.1 获取线程池
        ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors
                .newFixedThreadPool(100);

        // 2.2 执行任务，并获取任务执行结果
        List<Future<Person>> futureResults = new ArrayList<Future<Person>>();
        try {
            futureResults = executor.invokeAll(tasks);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        // 2.3 从futureResults中获取并解析出Person列表
        List<Person> persons = new ArrayList<Person>();
        for (Future<Person> ret : futureResults) {
            Person p;
            try {
                p = ret.get(); // get()方法会阻塞等到，直到获取到结果为止
                if (null != p) {
                    persons.add(p);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
                throw new Exception("InterruptedException occurs.");
            } catch (ExecutionException e) {
                e.printStackTrace();
                throw new Exception("ExecutionException occurs.");
            }
        }

        // 计时结束
        long end = System.currentTimeMillis();
        System.out.println("查询" + persons.size()
                + "个用户，concurrentQueryPersons方法共耗时：" + (end - start) + "毫秒");
        return persons;
    }
}

运行上面代码，输出结果为：

查询20个用户，singleThreadQueryPersons方法共耗时：6011毫秒
查询20个用户，concurrentQueryPersons方法共耗时：301毫秒

可见并发执行效果明显，最终总耗时约等于查询单个用户的耗时。

优化

封装并发执行泛型工具类

可以看到在PersonBusiness.concurrentQueryPersons方法中，先构建了Callable任务列表，然后创建了线程池，然后执行任务，最后获取任务执行结果。这里是查询Person信息，那如果后面再遇到需求变动了，需要查询Dog、Cat等信息，难道再重复写一遍这些代码吗？

为了代码重用，把上述的执行并发任务和获取执行结果两步骤的代码封装成工具类，便于以后重用：

package com.executor.util;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;

import com.executor.model.Person;

/**
 * Executor Framework并发任务处理类
 * 
 * @author Administrator
 *
 */
public class ExecutorUtil {

    /**
     * 并发执行Callable任务方法，支持泛型参数
     * 
     * @param tasks
     * @return
     * @throws Exception
     */
    public static <T> List<T> concurrentExecute(List<Callable<T>> tasks)
            throws Exception {
        // 1. 获取线程池
        ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors
                .newFixedThreadPool(100);

        // 2. 并发执行任务，并获取返回结果
        List<Future<T>> futureResults = new ArrayList<Future<T>>();
        try {
            futureResults = executor.invokeAll(tasks);
        } catch (InterruptedException e) {
            e.printStackTrace();
            throw new Exception("InterruptedException occurs.");
        }

        // 3. 取回并解析返回结果
        List<T> results = getFromFutureResults(futureResults);

        return results;
    }

    /**
     * 从并发Future结果中取回并解析结果，支持泛型参数
     * 
     * @param futureResults
     * @return
     * @throws Exception
     */
    private static <T> List<T> getFromFutureResults(
            List<Future<T>> futureResults) throws Exception {

        List<T> results = new ArrayList<T>();
        for (Future<T> ret : futureResults) {
            try {
                T r = ret.get(); // get()方法会阻塞等到，直到获取到结果为止
                if (null != r) {
                    results.add(r);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
                throw new Exception("InterruptedException occurs.");
            } catch (ExecutionException e) {
                e.printStackTrace();
                throw new Exception("ExecutionException occurs.");
            }
        }

        return results;
    }

}

线程池对象单例化

如果每次执行并发任务都创建一个线程池，将会造成资源浪费，那么可以考虑进一步优化，对ExecutorUtil.concurrentExecute方法中创建线程池封装成一个单例模式的类，比如叫MyThreadPool，不用每次调用都创建新的线程池。

总结

最终的代码结构：

可以看到Callable任务加上Executor线程池的调度，可以让我们轻松写出可以获取执行结果的并发代码，而且执行效率很高。此外，如果想要获取并发执行过程中的异常，可以通过改造Callable任务类的泛型为Exception，并在call方法中捕获并返回异常即可，这里不再赘述。

代码已经push到：我的GitHub