使用 ExecutorService 控制任务执行顺序 [英] Controlling Task execution order with ExecutorService

问题描述

我有一个将异步任务委托给线程池的进程.我需要确保某些任务按顺序执行.所以例如

I have a process which delegates asynch tasks to a pool of threads. I need to ensure that certain tasks are executed in order. So for example

任务按顺序到达

任务 a1、b1、c1、d1、e1、a2、a3、b2、f1

Tasks a1, b1, c1, d1 , e1, a2, a3, b2, f1

任务可以按任何顺序执行，除非存在自然依赖性，因此 a1,a2,a3 必须按该顺序处理，方法是分配给同一线程或阻塞这些线程，直到我知道前一个 a# 任务已完成.

Tasks can be executed in any order except where there is a natural dependancy, so a1,a2,a3 must be processed in that order by either allocating to the same thread or blocking these until I know the previous a# task was completed.

目前它不使用 Java Concurrency 包，但我正在考虑更改以利用线程管理.

Currently it doesn't use the Java Concurrency package, but I'm considering changing to take avantage of the thread management.

有没有人有类似的解决方案或如何实现这一目标的建议

Does anyone have a similar solution or suggestions of how to achieve this

推荐答案

我编写了自己的 Executor 来保证具有相同键的任务的任务排序.它对具有相同键的订单任务使用队列映射.每个键控任务使用相同的键执行下一个任务.

I write own Executor that warrants task ordering for tasks with same key. It uses map of queues for order tasks with same key. Each keyed task execute next task with the same key.

此解决方案不处理 RejectedExecutionException 或来自委托执行器的其他异常！所以委托的Executor应该是无限"的.

This solution don't handle RejectedExecutionException or other exceptions from delegated Executor! So delegated Executor should be "unlimited".

import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.Executor;

/**
* This Executor warrants task ordering for tasks with same key (key have to implement hashCode and equal methods correctly).
*/
public class OrderingExecutor implements Executor{

    private final Executor delegate;
    private final Map<Object, Queue<Runnable>> keyedTasks = new HashMap<Object, Queue<Runnable>>();

    public OrderingExecutor(Executor delegate){
        this.delegate = delegate;
    }

    @Override
    public void execute(Runnable task) {
        // task without key can be executed immediately
        delegate.execute(task);
    }

    public void execute(Runnable task, Object key) {
        if (key == null){ // if key is null, execute without ordering
            execute(task);
            return;
        }

        boolean first;
        Runnable wrappedTask;
        synchronized (keyedTasks){
            Queue<Runnable> dependencyQueue = keyedTasks.get(key);
            first = (dependencyQueue == null);
            if (dependencyQueue == null){
                dependencyQueue = new LinkedList<Runnable>();
                keyedTasks.put(key, dependencyQueue);
            }

            wrappedTask = wrap(task, dependencyQueue, key);
            if (!first)
                dependencyQueue.add(wrappedTask);
        }

        // execute method can block, call it outside synchronize block
        if (first)
            delegate.execute(wrappedTask);

    }

    private Runnable wrap(Runnable task, Queue<Runnable> dependencyQueue, Object key) {
        return new OrderedTask(task, dependencyQueue, key);
    }

    class OrderedTask implements Runnable{

        private final Queue<Runnable> dependencyQueue;
        private final Runnable task;
        private final Object key;

        public OrderedTask(Runnable task, Queue<Runnable> dependencyQueue, Object key) {
            this.task = task;
            this.dependencyQueue = dependencyQueue;
            this.key = key;
        }

        @Override
        public void run() {
            try{
                task.run();
            } finally {
                Runnable nextTask = null;
                synchronized (keyedTasks){
                    if (dependencyQueue.isEmpty()){
                        keyedTasks.remove(key);
                    }else{
                        nextTask = dependencyQueue.poll();
                    }
                }
                if (nextTask!=null)
                    delegate.execute(nextTask);
            }
        }
    }
}

这篇关于使用 ExecutorService 控制任务执行顺序的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！