Java ScheduledExecutorService - 防止多个并行任务中的饥饿答案

【问题标题】：Java ScheduledExecutorService - Preventing Starvation in Multiple Parallel TasksJava ScheduledExecutorService - 防止多个并行任务中的饥饿
【发布时间】：2017-02-02 21:44:39
【问题描述】：

我正在开发一个需要并行和定期检查多个资源的程序：

public class JobRunner {

    private final SensorService sensorService;
    private ScheduledExecutorService executor = Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors());

    public void run() {
        sensorService.finalAll().forEach(sensor -> {
            Runnable task = () -> {
                // read and save new data to log
                List<Double> data = sensor.fetchNewData();
                this.save(data);
            };

            // execute every 10 sec
            executor.scheduleWithFixedDelay(task, 0, 10, TimeUnit.SECONDS);
        });
    }

    public void save(List<Double> data) {
        // ...
    }
}

findAll 调用返回大约 50 个传感器的列表，但是当我运行程序时，我看到虽然在第一个周期查询了所有传感器，但在后续执行中只调用了 2-3 个（例如 - 在 20 秒, 30 秒等)。我在想，由于某些传感器比其他传感器返回得更快，它们会更早地完成任务的等待周期，并被池中的下一个线程抓取，从而使其他完成较慢的任务饿死。

如何确保所有任务（传感器）都得到平等对待？这里有哪些最佳实践；我应该使用作业队列还是不同的并发机制？谢谢。

【问题讨论】：

标签： java concurrency scheduled-tasks starvation

【解决方案1】：

在您的代码中有N=count service.findAll() 计时器，这使得调试和测试更加困难。此外，不能保证旧任务会在合理的时间内被执行并且不会被新任务取代。如果你

使用单个计时器，在所有传感器检查完成后 10 秒触发传感器检查
当计时器触发检查时，同时通过传感器

请以下面的代码为例。它每 10 秒打印 50 个整数，然后 EOL。使用Stream API实现并行化

ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);
executor.scheduleWithFixedDelay(new Runnable() {
    @Override
    public void run() {
        IntStream.range(0, 50).parallel().forEach(i -> System.out.print(i + " "));
        System.out.println();
    }
}, 0, 10, TimeUnit.SECONDS);

您可以将ScheduledExecutorService 替换为Timer 以使代码更清晰。并且，作为一个选项，您可以使用另一个 ExecutorService，而不是使用并行流，在 Timer 上提交下一个 N 任务并等待它们完成：

ExecutorService workerExecutor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
Timer timer = new Timer();
timer.schedule(new TimerTask() {
    @Override
    public void run() {
        List<Future<Void>> futures = new ArrayList<>();
        for (int i = 0; i < 50; i++) {
            final int index = i;
            Future<Void> future = workerExecutor.submit(new Callable<Void>() {
                @Override
                public Void call() throws Exception {
                    System.out.print(index + " ");
                    return null;
                }
            });
            futures.add(future);
        }
        for (Future<Void> future : futures) {
            try {
                future.get();
            } catch (InterruptedException|ExecutionException e) {
                throw new RuntimeException();
            }
        }
        System.out.println();
    }
}, 0, 10_000);

【讨论】：