java - ScheduledExecutorService每次循环后调用方法

Question

我需要以下场景:

在一个周期内运行所有ScheduledFutures，并在所有任务完成执行后每次调用tasksCompleted()方法。在实际调度周期之后调用 tasksCompleted() 时，下一个调度周期不得等待。

简而言之:在实际调度周期完成后调用方法，并且不停止下一个调度周期

以下代码创建任务并进行调度。但是，当一个周期内的所有任务完成时，我无法调用 tasksCompleted() 。

import java.time.LocalTime;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

public class Scheduler {

    public static void main(String[] args) {
        final ScheduledExecutorService ses = Executors.newScheduledThreadPool(10);
        System.out.println("- [" + LocalTime.now() + "] run parent-task...");

        // create 3 tasks: each task needs 7 seconds.
        var tasks = createTasks("child", 3, 7);
        List<ScheduledFuture<?>> futures = new ArrayList<>();
        tasks.forEach(t ->
        {
            ScheduledFuture<?> future = ses.scheduleWithFixedDelay(t, 0, 2, TimeUnit.SECONDS);
            futures.add(future);
        });
        // this does not work..
        var scheduleCycleCompleted = futures.stream().allMatch(f -> f.isDone());
        System.out.println("scheduleCycleCompleted: " + scheduleCycleCompleted);

        // maybe a solution with CompletableFuture?
        CompletableFuture[] cfs = futures.toArray(new CompletableFuture[futures.size()]);

    }

    static void tasksCompleted() {
        System.out.println("schedule cycle completed");
    }

    static List<Runnable> createTasks(String group, int numbersOfTasks, long taskDuration) {
        var tasks = new ArrayList<Runnable>();
        for (var i = 0; i < numbersOfTasks; i++) {
            int taskNr = i;
            Runnable task = () ->
            {
                System.out.println("- [" + LocalTime.now() + "] Running " + group + "-task" + taskNr + "...[needs "
                        + taskDuration + " seconds]");
                try {
                    TimeUnit.SECONDS.sleep(taskDuration);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                    Thread.currentThread().interrupt();
                }
            };
            tasks.add(task);
        }
        return tasks;
    }

}

Answer 1

已更新

希望它能起作用。

CountDownLatch 将解决这里的问题。

import java.time.LocalTime;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

public class Scheduler {

    public static void main(String[] args) throws InterruptedException {
        final ScheduledExecutorService ses = Executors.newScheduledThreadPool(10);

        System.out.println("- [" + LocalTime.now() + "] run parent-task...");

        int noOfTask=3;
        CountDownLatch countDownLatch = new CountDownLatch(noOfTask);

        TaskComplete taskComplete=new TaskCompleteImpl(noOfTask,countDownLatch);

        // create 3 tasks: each task needs 7 seconds.
        List<Runnable> tasks = createTasks("child", noOfTask, 7,countDownLatch,taskComplete);

        List<ScheduledFuture<?>> futures = new ArrayList<>();
        tasks.forEach(t ->
        {
            ScheduledFuture<?> future = ses.scheduleWithFixedDelay(t, 0, 2, TimeUnit.SECONDS);
            futures.add(future);
        });
        // this does not work..





    }

    interface TaskComplete{
        void tasksCompleted();
    }

    static class TaskCompleteImpl implements TaskComplete {

        int totalTask=0;
        int index=0;
        CountDownLatch countDownLatch;
        public TaskCompleteImpl(int totalTask){

        }

        public TaskCompleteImpl(int noOfTask, CountDownLatch countDownLatch) {
            this.totalTask=noOfTask;
            this.countDownLatch=countDownLatch;
        }

        @Override
        public synchronized void tasksCompleted() {
            index=index+1;
            if(index==totalTask){
                System.out.println("schedule cycle completed");
                index=0;
                countDownLatch=new CountDownLatch(totalTask);
            }

        }
    }


    static List<Runnable> createTasks(String group, int numbersOfTasks, long taskDuration, CountDownLatch countDownLatch, TaskComplete taskComplete) {
        List tasks = new ArrayList<Runnable>();
        for (int i = 0; i < numbersOfTasks; i++) {
            int taskNr = i;
            Runnable task = () ->
            {
                System.out.println("- [" + LocalTime.now() + "] Running " + group + "-task" + taskNr + "...[needs "
                        + taskDuration + " seconds]");
                try {
                    TimeUnit.SECONDS.sleep(taskDuration);
                    countDownLatch.countDown();
                    countDownLatch.await();
                    taskComplete.tasksCompleted();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                    Thread.currentThread().interrupt();
                }
            };
            tasks.add(task);
        }
        return tasks;
    }

}