java - TaskScheduler 的多线程版本，它保证高优先级任务永远不会在 java 中的低优先级任务之后开始执行

Question

我需要写一个多线程版本的TaskSchedular

TaskScheduler {
   TaskScheduler(List<Task> tasks)
    .... 
}


   interface Task {
      int getPriority()
      void execute()
    }

这样就保证了高优先级的任务永远不会在低优先级之后开始执行。

这是我的单线程版本示例。我需要创建这个多线程。

public class SingleThreadedTaskScheduler {

    private ExecutorService executorService = Executors.newFixedThreadPool(1);

    private PriorityQueue<Task> taskQueue;
    private static final PriorityComparator PRIORITY_COMPARATOR = new PriorityComparator();


    public SingleThreadedTaskScheduler(List<Task> tasks){
        this.taskQueue = new PriorityQueue<Task>(tasks.size(), PRIORITY_COMPARATOR);
        this.taskQueue.addAll(tasks);
    }

    public void executeAllByPriority(){
        executorService.submit(new Runnable() {
            @Override
            public void run() {
                for (Task task : taskQueue) {
                    task.execute();
                }
            }
        });

    }


    public void executeAllByPriorityWithUnInterruptableFirst(){
        executorService.submit(new Runnable() {
            @Override
            public void run() {
                PriorityQueue<Task> interruptedQueue = new PriorityQueue<Task>(taskQueue.size(),PRIORITY_COMPARATOR);
                for (Task task : taskQueue) {
                    if (task instanceof Interruptible && ((Interruptible) task).isInterrupted() ){
                        interruptedQueue.add(task);
                    } else {
                        task.execute();
                    }
                }

                for (Task task : interruptedQueue ){
                    task.execute();
                }
            }
        });

    }

    @Override
    protected void finalize() throws Throwable {
        super.finalize();
        executorService.shutdown();
    }

}

小代码会有很大帮助。

Answer 1

I am a bit confused with your question. Do you want to execute lower priority task first?

I think if you are asking how to use the TaskScheduler for making sure higher priority task over lower priority and BY-Pass the default FIFO, then you need to use a PriorityBlockingQueue with an object that has implemented java.util.Comparator Interface. With your own comparator, you can also define your own priority.

Answer 2

没关系，我自己想出了解决方案。我使用的线程池大小为 10。并使用 AtomicInteger 来实际查看优先级较高的线程是否在较低优先级的线程之后没有获得 CPU 周期。

如果您看到不正确的内容，请添加您的反馈。

public class MultiThreadedTaskScheduler  {

      AtomicInteger priorityCounter = new AtomicInteger(-1);
private ExecutorService executorService = Executors.newFixedThreadPool(10);

private PriorityQueue<Task> taskQueue;
private static final PriorityComparator PRIORITY_COMPARATOR = new PriorityComparator();


        public MultiThreadedTaskScheduler(List<Task> tasks){
            this.taskQueue = new PriorityQueue<Task>(tasks.size(), PRIORITY_COMPARATOR);
            this.taskQueue.addAll(tasks);
        }

        public void executeAllByPriority(){
            executorService.submit(new Runnable() {
                @Override
                public void run() {
                    for (Task task : taskQueue) {
                        executeMethod(task);
                    }
                }
            });

        }

    private void executeMethod(Task task) {
        if (task.getPriority()  > priorityCounter.get() ){
            task.execute();
            priorityCounter.getAndSet(task.getPriority());
        }

    }



        public void executeAllByPriorityWithUnInterruptableFirst(){
            executorService.submit(new Runnable() {
                @Override
                public void run() {
                    PriorityQueue<Task> interruptedQueue = new PriorityQueue<Task>(taskQueue.size(),PRIORITY_COMPARATOR);
                    for (Task task : taskQueue) {
                        if (task instanceof Interruptible && ((Interruptible) task).isInterrupted() ){
                            interruptedQueue.add(task);
                        } else {
                            executeMethod(task);
                        }
                    }

                    for (Task task : interruptedQueue ){
                        executeMethod(task);
                    }
                }
            });

        }

        @Override
        protected void finalize() throws Throwable {
            super.finalize();
            executorService.shutdown();
        }

}