具有重试能力的 ThreadpoolExecutor 并在任务多次失败后关闭

Question

我需要一个线程池执行器，它需要完成确切数量（相同）的任务。

它必须能够重新提交失败的任务 n 次。如果任何任务失败超过 n，那么线程池应该关闭并且不再继续处理任何其他任务。

我尝试结合我在不同答案中找到的 2 种方法 - 一种用于通过覆盖 ThreadPoolExecutor.afterExecute 和子类化 CountDownLatch 重新提交失败的任务，以便等待闩锁的线程被中断并关闭执行器。

到目前为止，这是子类倒计时锁存器：

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

public class AbortableCountDownLatch extends CountDownLatch {
protected boolean aborted = false;

public AbortableCountDownLatch(int count) {
    super(count);
}

/**
 * Unblocks all threads waiting on this latch and cause them to receive an
 * AbortedException.  If the latch has already counted all the way down,
 * this method does nothing.
 */
public void abort() {
    if( getCount() == 0 )
        return;

    this.aborted = true;
    while(getCount() > 0)
        countDown();
}

@Override
public boolean await(long timeout, TimeUnit unit) throws InterruptedException {
    final boolean rtrn = super.await(timeout,unit);
    if (aborted)
        throw new AbortedException();
    return rtrn;
}

@Override
public void await() throws InterruptedException {
    super.await();
    if (aborted)
        throw new AbortedException();
}


public static class AbortedException extends InterruptedException {
    public AbortedException() {
    }

    public AbortedException(String detailMessage) {
        super(detailMessage);
    }
}
}

还有线程池执行器：

public class MyThreadPoolExecutor extends ThreadPoolExecutor {

private static final int RETRY_LIMIT = 3;

private Map<Runnable, Integer> retriedTasks = new ConcurrentHashMap<>();
private AbortableCountDownLatch latch;

public MyThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime,
                            TimeUnit unit, BlockingQueue<Runnable> workQueue, AbortableCountDownLatch latch) {
    super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    this.latch = latch;
}

@Override
public void afterExecute(Runnable r, Throwable t) {
    super.afterExecute(r, t);
    // If submit() method is called instead of execute()
    if (t == null && r instanceof Future<?>) {
        try {
            Object result = ((Future<?>) r).get();
        } catch (CancellationException e) {
            t = e;
        } catch (ExecutionException e) {
            t = e.getCause();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
    if (t != null) {
        retriedTasks.put(r, retriedTasks.getOrDefault(r, 0) + 1);
        System.out.println("Retries for " + r + " -> " + retriedTasks.get(r));
        /* check to see if we have retried this task too many times, if so - shutdown */
        if (retriedTasks.containsKey(r) && retriedTasks.get(r) > RETRY_LIMIT) {
            System.err.println("Thread failed for more than " + RETRY_LIMIT + " times, aborting everything..");
            this.latch.abort();
        } else {
            System.err.println("Thread threw  exception " + t.getMessage() + ". Retry-ing task...");
            execute(r);
        }
    } else {
        /* clear any previous retry count for this runnable */
        retriedTasks.remove(r);
    }
}
}

一个主要的人会像这样使用它们：

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

public class MainProcessor {

public static void main(String[] args) {

    AbortableCountDownLatch latch = new AbortableCountDownLatch(5);
    ThreadPoolExecutor threadPoolExecutor = new MyThreadPoolExecutor(8, 8, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), latch);

    for (int i = 0; i < 5; i++) {
        threadPoolExecutor.submit(() -> {
            System.out.println("Started thread " + Thread.currentThread().getName());
            Random random = new Random();
            try {
                Thread.sleep(random.nextInt(7000));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
                if (random.nextBoolean()){
                    System.err.println("Thread " + Thread.currentThread().getName() + " failed - throwing exception..");
                    throw new RuntimeException("Thread " + Thread.currentThread().getName() + "failed! spectacularly :!");
                }
                else {
                    System.out.println("Thread " + Thread.currentThread().getName() + " finished.");
                    latch.countDown();
                }
        });
    }

    try {
        latch.await();
    } catch (InterruptedException e) {
        threadPoolExecutor.shutdownNow();
    }

    threadPoolExecutor.shutdown();
}
}

这种方法看起来正确吗？我不特别喜欢必须将闩锁传递给线程池执行器和实际的 Runnable。有没有实现这一目标的标准方法？我也可以使用 Scala 版本。

我看到其他人建议任务应该在失败的情况下重新提交到池中，但这似乎不是一个好主意，因为任务应该只负责实际运行的逻辑，而不是执行细节。

Answer 1

您可以使用 Task-Wrapper 来完成这项工作，那么它会相当简单：

public class TaskWrapper implements Runnable
{
    private Runnable task;
    private int maxResubmits;
    private ThreadPoolExecutor executor;
    private CountDownLatch latch;

    public TaskWrapper(Runnable task, int maxResubmits, ThreadPoolExecutor executor, CountDownLatch latch) {
        this.task=task;
        this.maxResubmits=maxResubmits;
        this.executor=executor;
        this.latch=latch;
        executor.submit(this);
    }

    public void run() {
        try {
            task.run();
            latch.countdoun();
        }
        catch(Exception e) {
            maxResubmits--;
            if(maxResubmits>0)
                executor.submit(this);
            else
            {
                latch.countdoun();
                executor.shutdownNow()
            }                
        }
    }
}

您现在只需要创建闩锁，调用您的任务，然后等待执行：

List<Runnable> tasks;
int maxResubmits;

CountDownLatch latch=new CountDownLatch(tasks.size());

tasks.forEach(task->new TaskWrapper(task,maxResubmits,executor,latch));

latch.await();

if(!executor.isShutdown())
    executor.shutdown();