我将避免使用executor线程池，并使用以下方法：
创建一个具有blockingqueue（选择实现）和一个持续使用该队列的线程的类。它显然会消耗任务。当它获取任务时，它将执行它。这可以保证队列中的任务是按顺序执行的，并且最多可以执行该队列中的一个任务。
现在创建一些将任务Map到队列的逻辑。e、 g.如果不希望同时执行相同类型的任务，请将它们全部Map到同一队列。例如，a总是转到队列1，b到2，等等。。。它遵循前面提到的属性，通过这种方式，不会有两个类型为a的任务并行进行。
一种常见的方法是将所有队列放在一个数组中，然后根据队列数量的某个值进行分配，这就是调用分区，例如：

queues = new Queue[N];

typeOfTask = task.getType() (to int somehow, for example you can hash or use the char if it is a single letter)

queueIndex = typeTask % queues.length

queues[queueIndex].push(task)

在上面的类中，在将任务传递给线程之前获取任务时，您可以执行任何逻辑，就像您提到的那样。
最后，在使用文件系统时，我将避免使用许多线程（如果有的话），因为磁盘是。。。一个。因此，并行处理没有什么大的意义，除非您必须编写大量不需要缓冲的数据。相反，线程更多的是能够并行工作，使用更多的处理器来提高速度，或者在执行阻塞操作时。

我试着实现消费者用例的一部分，似乎问题多于解决方案。请阅读代码中的内联注解。

public class Consumer implements Runnable {

    private volatile boolean isStopped;
    private final Map<String, Queue<Task>> tasks;
    private final Map<String, AtomicBoolean> tokens;
    //Preferably FixedThreadService
    private final ExecutorService service;

    public Consumer(Map<String, Queue<Task>> tasks, Map<String, AtomicBoolean> tokens, ExecutorService service) {
        this.tasks = tasks;
        this.tokens = tokens;
        this.service = service;
        isStopped = false;
    }

    public void stopConsumer(){
        isStopped = true;
    }

    @Override
    public void run() {
        while(!isStopped){
            // tasks map should have latest values because producer is going to update this map
            // frequently. Should have to use ConcurrentHashMap.
            Set<String> keys = tasks.keySet();
            for ( String key:tasks.keySet() ) {
                if(!tokens.get(key).get()){
                    continue;
                }
              try {
                  //This does not mean the task is in progress, just submitted for execution.
                  service.execute(tasks.get(key).poll());
                  //Now that we submitted the task for exicution, we can mark the token to false.
                  tokens.get(key).compareAndSet(true, false);
                  // But at what point are we going to mark token to True. If this token is updated
                  // From any other thread than consumer, it will create race condition.
              }catch(RejectedExecutionException e){
                  //What will happen to the polled task. Task is removed from the queue.

                  //Token we can mark to true, but task has already polled from queue.
                  // Shall we put it back to the head of the queue and how ?
                  tokens.get(key).compareAndSet(false, true);
                  //What if the task is not completed smoothly. T1 is not completed successfully,
                  // what is backup plan, because next task will start any way and corrupt the file.

              }
            }

        }
    }
}

如果我们能将诸如1.添加文件、2.修改文件、3.提交和4.推送到远程等任务隔离开来，我们可能会想出更好的方法。但在这种情况下也会有限制。
最好使用单线程方法。