java并发（二十四）多线程结果组装

本文主要介绍多线程的结果组装。其中可以忽略2处代码创建线程池的区别，请关注其他的业务逻辑代码。全部代码已经在附件中上传。如有疑问，请跟帖留言，笔者会予以答复。信号量相比自旋锁的优点很多，性能、代码简单。自旋锁不停得sleep并唤醒，而信号量的底层采用了wait进行编程，只唤醒一次即可。因此性能优越许多。


【自旋锁】
通常，我们会使用自旋锁进行多线程结果组装。这样的性能非常差。比如，大数据库表的多线程查询，无状态服务器的任务分发后的汇总等。
示例代码如下

package com.chinaso.search.spinlocks.service;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.ThreadPoolExecutor.AbortPolicy;
import java.util.concurrent.TimeUnit;

import com.chinaso.search.ServerSemaphore;
import com.chinaso.search.dao.DataDao;

/**
 * piaohailin 2014-3-21
 */
public class DataService {
    private final DataDao dataDao = new DataDao();

    /**
     * 多线程查询数据库
     * @param userId
     * @return
     * @throws Exception
     */
    public List<String> getAllData(final String userId) throws Exception {
        /**
         * 其中 第一个参数为初始空闲 
         * 第二个参数为最大线程 
         * 第三个参数为超时时间 
         * 第四个参数是超时时间的单位
         * 第五个参数是当超过最大线程数以后，可以放在队列中的线程 
         * 第六个参数 
         * 第七个参数是线程池塞满时候的策略
         */
        int corePoolSize = 2;
        int maximumPoolSize = 3;
        long keepAliveTime = 0;
        TimeUnit unit = TimeUnit.NANOSECONDS;
        BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<Runnable>(5);
        ThreadFactory threadFactory = Executors.defaultThreadFactory();
        /**
         * AbortPolicy 如果总线成熟超过maximumPoolSize + workQueue
         * ，则跑异常java.util.concurrent.RejectedExecutionException
         */
        RejectedExecutionHandler handler = new AbortPolicy();

        // ExecutorService 为线程池的接口
        ThreadPoolExecutor executor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
        List<Future<List<String>>> futures = new ArrayList<Future<List<String>>>();

        final int count = dataDao.getCount(userId); // 总记录数
        System.out.println("count=" + count);
        // 如果总记录数小于设置的阈值，就直接单线程查询
        int threadCount = ServerSemaphore.threadCount;
        if (count < ServerSemaphore.hold) {
            threadCount = 1;
        }
        int section = count / threadCount; // 区间大小
        // 创建线程
        for (int i = 0; i < threadCount; i++) {
            final int begin = i * section;
            final int end;
            // 最后一个区间判断
            if ((i + 1) == threadCount) {
                end = count;
            } else {
                end = (i + 1) * section;
            }
            System.out.print("begin=" + begin);
            System.out.print(",end=" + end);
            System.out.println(",size=" + (end - begin));

            // 根据总记录数count和线程数Server.threadCount进行分页任务分发
            Future<List<String>> future = executor.submit(new Callable<List<String>>() {
                @Override
                public List<String> call() throws Exception {
                    List<String> data = new ArrayList<String>();
                    try {
                        data = dataDao.find(userId, begin, end);
                    } catch (Throwable t) {
                        t.printStackTrace();
                    }
                    return data;
                }
            });
            futures.add(future);
        }

        this.waitForDone(futures);

        // 结果的组装
        List<String> reuslt = new ArrayList<String>();
        for (Future<List<String>> future : futures) {
            List<String> tmp = future.get();
            reuslt.addAll(tmp);
        }
        return reuslt;
    }

    private void waitForDone(List<Future<List<String>>> futures) {
        boolean done = false;
        while (!done) {
            done = true;
            for (Future<List<String>> future : futures) {
                future.isDone();
                if (!future.isDone()) {
                    done = false;
                    try {
                        Thread.sleep(50);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                    break;
                }
            }
        }
    }
}

【信号量】
代码相比线程轮询更简单，思路清晰，而且可以共用一个线程池来进行全局的控制。当所有线程完成任务的时候，唤醒组装线程进行结果的反馈。

package com.chinaso.search.semaphore.service;

import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

import com.chinaso.search.ServerSemaphore;
import com.chinaso.search.dao.DataDao;
import com.chinaso.search.semaphore.concurrent.DataSemaphore;

/**
 * piaohailin
 * 2014-3-22
 */
public class DataService {
    private Executor executor = Executors.newFixedThreadPool(4); // 共用线程池，是为了从全局角度，叫多线程可控
    private final DataDao dataDao = new DataDao();

    /**
     * 多线程查询数据库
     * @param userId
     * @return
     * @throws Exception
     */
    public List<String> getAllData(final String userId) throws Exception {
        final DataSemaphore semaphore = new DataSemaphore(0);

        final int count = dataDao.getCount(userId); // 总记录数
        System.out.println("count=" + count);
        // 如果总记录数小于设置的阈值，就直接单线程查询
        int threadCount = ServerSemaphore.threadCount;
        if (count < ServerSemaphore.hold) {
            threadCount = 1;
        }
        int section = count / threadCount; // 区间大小
        // 创建线程
        for (int i = 0; i < threadCount; i++) {
            final int begin = i * section;
            final int end;
            // 最后一个区间判断
            if ((i + 1) == threadCount) {
                end = count;
            } else {
                end = (i + 1) * section;
            }
            System.out.print("begin=" + begin);
            System.out.print(",end=" + end);
            System.out.println(",size=" + (end - begin));

            // 根据总记录数count和线程数Server.threadCount进行分页任务分发
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        List<String> data = dataDao.find(userId, begin, end);
                        semaphore.fillData(data);
                    } catch (Throwable t) {
                        t.printStackTrace();
                    } finally {
                        // 执行成功后，发放授权
                        semaphore.release();
                    }
                }
            });
        }

        semaphore.acquire(threadCount); // 等待授权数量满足条件，放行
        return semaphore.getResult();
    }
}

【CompletionService】
CompletionService将Executor和BlockingQueue的功能融合在一起。你可以将Callable任务提交给它来执行，然后使用类似于队列操作的take和poll等方法来获得已经完成的结果，而这些结果会在完成时被封装为Future。ExecutorCompletionService实现了CompletionService，并将计算部分委托给一个Executor。

package com.chinaso.search.completionservice.service;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.Executors;

import com.chinaso.search.ServerSemaphore;
import com.chinaso.search.dao.DataDao;
import com.chinaso.search.semaphore.concurrent.DataSemaphore;

/**
 * piaohailin
 * 2014-3-22
 */
public class DataService {
    private Executor executor = Executors.newCachedThreadPool(); // 共用线程池，是为了从全局角度，叫多线程可控
    private final DataDao dataDao = new DataDao();

    /**
     * 多线程查询数据库
     * 
     * @param userId
     * @return
     * @throws Exception
     */
    public List<String> getAllData(final String userId) throws Exception {
        final int count = dataDao.getCount(userId); // 总记录数
        System.out.println("count=" + count);
        // 如果总记录数小于设置的阈值，就直接单线程查询
        int threadCount = ServerSemaphore.threadCount;
        if (count < ServerSemaphore.hold) {
            threadCount = 1;
        }
        int section = count / threadCount; // 区间大小
        // 创建线程
        CompletionService<List<String>> completionService = new ExecutorCompletionService<List<String>>(executor);// 多线程任务管理
        for (int i = 0; i < threadCount; i++) {
            final int begin = i * section;
            final int end;
            // 最后一个区间判断
            if ((i + 1) == threadCount) {
                end = count;
            } else {
                end = (i + 1) * section;
            }
            System.out.print("begin=" + begin);
            System.out.print(",end=" + end);
            System.out.println(",size=" + (end - begin));

            // 根据总记录数count和线程数Server.threadCount进行分页任务分发

            completionService.submit(new Callable<List<String>>() {
                @Override
                public List<String> call() throws Exception {
                    List<String> data = null;
                    try {
                        data = dataDao.find(userId, begin, end);
                    } catch (Throwable t) {
                        t.printStackTrace();
                    } finally {
                        if (data == null) {
                            data = new ArrayList<String>();
                        }
                    }
                    return data;
                }
            });
        }

        List<String> result = new ArrayList<String>();
        for (int i = 0; i < threadCount; i++) {
            result.addAll(completionService.take().get()); // 取得结果，如果没有返回，则阻塞
        }
        return result;
    }
}

【更新历史】
2014-04-26
上传1.1版本，增加CompletionService的接口用法。

2014-05-04
上传1.2版本，重构service，使代码结构更清晰易读。