java - 如何在Spring TaskExecutor框架中管理线程

Question

我有一个 Runnable 的 BlockingQueue - 我可以使用 TaskExecutor 实现之一简单地执行所有任务，并且所有任务都将并行运行。然而，有些Runnable依赖于其他Runnable，这意味着它们需要等待Runnable完成，然后才能执行。

规则非常简单:每个Runnable都有一个代码。具有相同代码的两个 Runnable 不能同时运行，但如果代码不同，则应并行运行。换句话说，所有正在运行的 Runnable 都需要有不同的代码，所有“重复项”都应该等待。

问题是线程结束时没有事件/方法/任何内容。我可以将这样的通知构建到每个 Runnable 中，但我不喜欢这种方法，因为它将在线程结束之前完成，而不是在线程结束之后完成

java.util.concurrent.ThreadPoolExecutor有方法afterExecute，但需要实现 - Spring仅使用默认实现，并且该方法被忽略。

即使我这样做，它也会变得复杂，因为我需要跟踪两个额外的集合:已执行的 Runnable 集合(没有实现可以访问此信息)和由于已执行而推迟的集合重复的代码。

我喜欢 BlockingQueue 方法，因为没有轮询，当队列中有新内容时，线程就会简单地激活。但也许有更好的方法来管理 Runnable 之间的此类依赖关系，所以我应该放弃 BlockingQueue 并使用不同的策略？

Answer 1

如果不同代码的数量不是那么大，则为每个可能的代码使用单独的单线程执行器的方法，由 BarrySW19 提供。 ，很好。如果线程总数变得 Not Acceptable ，那么我们可以使用 actor(来自 Akka 或其他类似库)来代替单线程执行器:

public class WorkerActor extends UntypedActor {
  public void onReceive(Object message) {
    if (message instanceof Runnable) {
      Runnable work = (Runnable) message;
      work.run();
    } else {
      // report an error
    }
  }
}

与原始解决方案一样，ActorRef为WorkerActor s 被收集在 HashMap 中。当 ActorRef workerActorRef获得(检索或创建)与给定代码相对应的 Runnable job提交执行 workerActorRef.tell(job) .

如果你不想依赖actor库，你可以编程WorkerActor从头开始:

public class WorkerActor implements Runnable, Executor {
  Executor executor=ForkJoinPool.commonPool(); // or can by assigned in constructor
  LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueu<>();
  boolean running = false;

  public synchronized void execute(Runnable job) {
    queue.put(job);
    if (!running) {
      executor.execute(this); // execute this worker, not job!
      running=true;
  }

  public void run() {
    for (;;) {
      Runnable work=null;
      synchronized (this) {
         work = queue.poll();
         if (work==null) {
           running = false;
           return;
         }
      }
      work.run();
    }
  }
}

当 WorkerActor worker获得(检索或创建)与给定代码相对应的 Runnable job提交执行 worker.execute(job) .