Java之多线程中的Master-Worker模式

该模式的好处是，将大任务拆解成若干小任务并并行执行，从而提高系统吞吐量。

定义Worker进程，负责处理实际任务。

/*具体工作对象*/
static abstract class Worker<T, R> implements Runnable {
    private static final UtilsLog lg = UtilsLog.getLogger(Worker.class);
    protected Queue<T> workQueue;//持有Master的任务队列
    protected Map<String, R> resultMap;//用于存储结果集,key为任务对应的唯一标识符

    public void setWorkQueue(Queue<T> workQueue) {
        this.workQueue = workQueue;
    }

    public void setResultMap(Map<String, R> resultMap) {
        this.resultMap = resultMap;
    }

    public abstract R handler(T entity);

    @Override
    public void run() {
        while (true) {
            T childWork = workQueue.poll();
            if (childWork == null) {
                lg.e("已经没有任务在队列中等待执行");
                break;
            }
            //处理子任务
            R result = handler(childWork);
            resultMap.put(Integer.toString(childWork.hashCode()), result);
        }
    }
}

定 义Master进程，负责接收和分配任务。Master会在提交任务的同时立即返回结果集，由于此处的 getResultMap属于引用传递，因此属性resultMap的修改会同步至业务层。

public static class Master<T, R> {
    private static final UtilsLog lg = UtilsLog.getLogger(Master.class);
    protected Queue<T> workQueue;//用于存储任务集
    protected Map<String, Thread> threadMap;//存储执行任务的线程集
    protected Map<String, R> resultMap;//存储相关结果

    @TargetApi(Build.VERSION_CODES.LOLLIPOP)
    public Master(Worker<T, R> work, int threadCount) {
        workQueue = new ConcurrentLinkedDeque<T>();
        threadMap = new HashMap<>();
        resultMap = new HashMap<>();

        work.setWorkQueue(workQueue);
        work.setResultMap(resultMap);
        for (int i = 0; i < threadCount; i++) {
            threadMap.put(Integer.toString(i), new Thread(work, "thread tag with " + Integer.toString(i)));
        }
    }

    //是否所有的子任务都结束了
    public boolean isComplete() {
        for (Map.Entry<String, Thread> entry : threadMap.entrySet()) {
            if (entry.getValue().getState() != Thread.State.TERMINATED) {
                return false;
            }
        }
        return true;
    }

    public Map<String, R> getResultMap() {
        return resultMap;
    }

    public Master addJob(T job) {
        workQueue.add(job);
        return this;
    }

    public void execute() {
        for (Map.Entry<String, Thread> entry : threadMap.entrySet()) {
            entry.getValue().start();
        }
    }
}

在 业务层调用方案如下，

        Master<Integer, Integer> master = new Master<>(new Worker<Integer, Integer>() {
            @Override
            public Integer handler(Integer entity) {
                int max = 50, min = 0;
                UtilsThread.sleepIgnoreInteruptedException(new Random().nextInt(max) % (max - min + 1) + min);//随机模拟耗时操作
                lg.e("执行handler程序 with value:" + entity);
                return entity * entity;
            }
        }, 3);
        int jobCount = 10;//任务数
        for (int i = 0; i < jobCount; i++) {
            master.addJob(i);
        }
        master.execute();

        Map<String, Integer> resultMap = master.getResultMap();
        while (true) {
            int resultMapSize = resultMap.size();
//            lg.e("并行执行结果集中已有数据量：" + resultMapSize);//此处resultMap持有Master中结果集的引用，因此在线程不断执行的过程中不断刷新结果姐，会连带导致这里值的改变
            if (master.isComplete()) {
                break;
            }
        }

执行结果如下，

另外，也应注意到，在执行 handler方式，到结果集中写入数据是有延时的， 这在开发中需要格外注意，务必使用master. isComplete判断任务完成状况 。

来自为知笔记(Wiz)