通过覆盖 afterExecute(Runnable r, Throwable t) 来重试线程池

Question

我想实现一个线程池，可以通过覆盖afterExecute 钩子来执行特定时间的任务。我可以再次提交参数Runnable r 吗？

这是我最初的实现。

public class RetriableThreadPool extends ThreadPoolExecutor {

  static final int MAXRETRYTIMES = 5;

  int retryTimes = 0;

  public RetriableThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime,
      TimeUnit unit, BlockingQueue<Runnable> workQueue) {
    super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    retryTimes = 0;
  }

  @Override
  protected void afterExecute(Runnable r, Throwable t) {
    super.afterExecute(r, t);
    if (retryTimes < MAXRETRYTIMES) {
      retryTimes++;
      super.submit(r);
    }
  }

}

在这个初始实现中，我只允许提交一个任务。

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

public class ThreadPoolTest {

  public static void main(String[] args) {
    RetriableThreadPool retriableThreadPool = new RetriableThreadPool(10, 10, 0L,
        TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());
    retriableThreadPool.execute(new Runnable() {
      int num = 0;

      @Override
      public void run() {
        // TODO Auto-generated method stub
        num = num + 123;
        System.out.println(num);
      }

    });
    try {
      Thread.sleep(1000);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    // retriableThreadPool.shutdown();
  }
}

在这个例子中，我得到了奇怪的输出：

123
246

如果可以重新提交 runnable，我想我应该得到 5 个输出。如果无法重新提交。结果应该只有 123。我不明白这个输出的原因。

---

感谢 nogard，我修改了代码

public class RetriableThreadPool extends ThreadPoolExecutor {

  static final int MAXRETRYTIMES = 5;

  int retryTimes = 0;

  public RetriableThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime,
      TimeUnit unit, BlockingQueue<Runnable> workQueue) {
    super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    retryTimes = 0;
  }

  @Override
  protected void afterExecute(Runnable r, Throwable t) {
    super.afterExecute(r, t);
    if (retryTimes < MAXRETRYTIMES) {
      retryTimes++;
      super.execute(r);
    }
  }
}

我还有 3 个问题：

1. 如何以原始状态重试可运行文件。在这种情况下，我预计结果将是 123 的 5 倍
2. 如何为方法submit 添加挂钩，就像afterExecute 为execute
3. 是否已经有可重试线程池的良好实现？当抛出异常或可调用返回某些结果时，我想运行可重试。

Answer 1

我认为这种行为的原因是您在 afterExecute 方法中提交任务而不是 execute，并且提交不会再次触发afterExecute 回调。这就是为什么您在输出中只看到两行：第一行来自原始执行，第二行来自提交。

此外，您永远不会增加重试计数器，您的任务将始终重新提交

@Override
protected void afterExecute(Runnable r, Throwable t) {
    super.afterExecute(r, t);
    ++ retryTimes;
    if (retryTimes < MAXRETRYTIMES) {
        super.execute(r);
    }
}

更新您的 3 个问题：

1. 有多种选择：

   • 不要更改 Runnable 中的状态（不要分配给 num）
   • 创建新的 Runnable 实例（或复制实例）
   • 重置Runnable状态

2. 对于钩子，我会用装饰器模式来实现：像这样：

   public class YourExecutor {
   @Override
   public void submit(Runnable task) {
       return super.submit(new TaskDecorator(task));
   }
   
   protected void onCompletedTask(Runnable task) {
       // callback
   }
   
   private class TaskDecorator implements Runnable {
       private final Runnable delegate;
   
       public TaskDecorator(Runnable delegate) {
           this.delegate = delegate;
       }
   
       @Override
       public void run() {
           this.delegate.run();
           onCompletedTask(delegate);
       }
   }