使用 ExecutorService 并行处理任务答案

【问题标题】：Using ExecutorService to process tasks in parallel使用 ExecutorService 并行处理任务
【发布时间】：2017-04-16 15:41:45
【问题描述】：

我正在编写一个需要同时监控多台机器的 java 程序。这个数字不是固定的，它可以在程序执行过程中随时变化（增加/减少）。

我正在考虑做这样的事情：

public static void main (String args[]) {

    ExecutorService EXEC1 = Executors.newScheduledThreadPool(1);

    EXEC1.scheduleWithFixedDelay(new Runnable() {

        ExecutorService EXEC2 = Executors.new...
        Map<Integer, Future<Void>> monitoringTasks = new HashMap<Integer, Future<Void>>();

        @Override
        public void run() {

            List<Machine> monitorizedMachines = MachineDao.getMonitorizedMachines();

            for (Machine machine: monitorizedMachines) {

                Future<Void> monitoringTask = monitoringTasks.get(machine.getId());

                if(monitoringTask == null || monitoringTask.isDone()) {

                    monitoringTask = EXEC2.submit(new Runnable() {
                        @Override
                        public void run() throws Exception {

                            // monitor machine....

                        }
                    });

                    monitoringTasks.put(machine.getId(), monitoringTask);
                }

            }
        }

    }, 1, 1, TimeUnit.SECONDS);

}

但我无法为这种情况选择最合适的执行程序 (EXEC2)：FixedThreadPool、CachedThreadPool、自定义实现，...

需要说，每个监控任务的时长大约是 2/3 秒。

谁能给我一些建议？

【问题讨论】：

我有个主意，我需要你的代码在你的机器类中

标签： java multithreading parallel-processing

【解决方案1】：

大多数时候，当您开发基于大型生产的应用程序时，您需要使用ExecutorService EXEC2 = Executors.newFixedThreadPool(THREAD_COUNT);，并且您需要在进行性能测试后正确配置THREAD_COUNT w与预期的请求数/卷数。

您可以查看here 以详细了解为什么newCachedThreadPool() 不适合具有大量请求的应用程序。

【讨论】：

【解决方案2】：

这是一个简单的例子。首先在你的类机器中添加一个公共变量布尔ISWORKING的例子。在 run() 代码中，在变量之间添加代码，如下一个示例：

public static class Machine implements Runnable {

        public boolean ISWORKING = true;

        @Override
        public void run() {
            this.ISWORKING = true;
            //YOUR CODE HERE..................
            this.ISWORKING = false;
        }

    }

第二个示例代码：

    Timer timer = null;
    TimerTask task = null;
    boolean isLocked = false;

    public void main() {

        task = new TimerTask() {

            @Override
            public void run() {

                if (isLocked) {
                    return;
                }

                isLocked = true;

                List<Machine> monitorizedMachines = MachineDao.getMonitorizedMachines();

                //Count the pending job.
                int poolsize = 0;
                for (Machine machine : monitorizedMachines) {

                    if (!machine.ISWORKING) {
                        poolsize++;
                    }

                }

                if (poolsize == 0) {
                    isLocked = false;
                    return;
                }

                //Prevent a lot of poolsize.
                poolsize = Math.min(100, poolsize);

                ThreadPoolExecutor pool = (ThreadPoolExecutor) Executors.newFixedThreadPool(poolsize);

                for (Machine machine : monitorizedMachines) {

                    if (!machine.ISWORKING) {
                        pool.execute(machine);
                    }

                }
                pool.shutdown();


                isLocked = false;

            }

        };

        timer = new Timer();
        timer.schedule(task, 1000, 2000);

    }

【讨论】：