[开发工具]自定义线程池

线程池大家都会，再不济用executors直接生成，但是使用线程池时候有一些痛点

1、线程抛异常，线程讲死掉，线程池新建一个线程，而异常被线程池吃掉了，外部感知不到

2、我们可能有需要对线程的各个状态时候执行一些操作，比如完成时，异常时

3、有时候我们不希望进入阻塞队列，希望有线程就执行没线程就等待到有线程为止

public class ThreadPoolExecutorWrapper {
    private ThreadPoolExecutor threadPoolExecutor;

    public static ThreadPoolExecutorWrapper buildBlockTPWithTimeout(int corePoolSize,
                                                                    int maximumPoolSize, long timeout, TimeUnit unit) {
        return build(corePoolSize, maximumPoolSize, 100, TimeUnit.MILLISECONDS, new SynchronousQueue<>(), Executors.defaultThreadFactory(), (r, executor) -> {
            try {
                executor.getQueue().offer(r, timeout, unit);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
    }

    public static ThreadPoolExecutorWrapper buildBlockTP(int corePoolSize,
                                                         int maximumPoolSize) {
        return build(corePoolSize, maximumPoolSize, 100, TimeUnit.MILLISECONDS, new SynchronousQueue<>(), Executors.defaultThreadFactory(), (r, executor) -> {
            try {
                executor.getQueue().put(r);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
    }

    public static ThreadPoolExecutorWrapper build(int corePoolSize,
                                                  int maximumPoolSize,
                                                  long keepAliveTime,
                                                  TimeUnit unit,
                                                  BlockingQueue<Runnable> workQueue,
                                                  ThreadFactory threadFactory,
                                                  RejectedExecutionHandler handler) {
        ThreadPoolExecutorWrapper threadPoolExecutorWrapper = new ThreadPoolExecutorWrapper();
        threadPoolExecutorWrapper.threadPoolExecutor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
        return threadPoolExecutorWrapper;
    }

    public void execute(Runnable command) {
        execute(command, null, null);
    }

    public void execute(Runnable command, Runnable doneHandle) {
        execute(command, doneHandle, null);
    }

    public void execute(Runnable command, Consumer<Throwable> exceptionHandle) {
        execute(command, null, exceptionHandle);
    }

    public void execute(Runnable command, Runnable doneHandle, Consumer<Throwable> exceptionHandle) {
        FutureTaskWrapper<?> ft = new FutureTaskWrapper<>(command, doneHandle, exceptionHandle);
        threadPoolExecutor.execute(ft);
    }

    public <V> FutureTaskWrapper<V> submit(Callable<V> callable) {
        return submit(callable, null, null);
    }

    public <V> FutureTaskWrapper<V> submit(Callable<V> callable, Runnable doneHandle) {
        return submit(callable, doneHandle, null);
    }

    public <V> FutureTaskWrapper<V> submit(Callable<V> callable, Consumer<Throwable> exceptionHandle) {
        return submit(callable, null, exceptionHandle);
    }

    public <V> FutureTaskWrapper<V> submit(Callable<V> callable, Runnable doneHandle, Consumer<Throwable> exceptionHandle) {
        FutureTaskWrapper<V> ft = new FutureTaskWrapper<>(callable, doneHandle, exceptionHandle);
        threadPoolExecutor.submit(ft);
        return ft;
    }
}

public class FutureTaskWrapper<V> extends FutureTask<V> {
    private Runnable doneHandle;
    private Consumer<Throwable> exceptionHandle;

    public FutureTaskWrapper(Callable<V> callable) {
        this(callable, null, null);
    }

    public FutureTaskWrapper(Runnable runnable) {
        this(runnable, null, null);
    }

    public FutureTaskWrapper(Callable<V> callable, Runnable doneHandle, Consumer<Throwable> exceptionHandle) {
        super(callable);
        this.doneHandle = doneHandle;
        this.exceptionHandle = Objects.nonNull(exceptionHandle) ? exceptionHandle : throwable -> {
            throw new RuntimeException(throwable);
        };
    }

    public FutureTaskWrapper(Runnable runnable, Runnable doneHandle, Consumer<Throwable> exceptionHandle) {
        super(runnable, null);
        this.doneHandle = doneHandle;
        this.exceptionHandle = exceptionHandle;
    }

    @Override
    protected void done() {
        if (Objects.isNull(doneHandle)) return;
        doneHandle.run();
    }

    @Override
    protected void setException(Throwable t) {
        super.setException(t);

        if (Objects.isNull(exceptionHandle)) return;
        exceptionHandle.accept(t);
    }

    @Override
    public V get() throws InterruptedException {
        try {
            return super.get();
        } catch (ExecutionException ignored) {

        }

        return null;
    }

    @Override
    public V get(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException {
        try {
            return super.get(timeout, unit);
        } catch (ExecutionException ignored) {

        }

        return null;
    }

    public Runnable getDoneHandle() {
        return doneHandle;
    }

    public void setDoneHandle(Runnable doneHandle) {
        this.doneHandle = doneHandle;
    }

    public Consumer<Throwable> getExceptionHandle() {
        return exceptionHandle;
    }

    public void setExceptionHandle(Consumer<Throwable> exceptionHandle) {
        this.exceptionHandle = exceptionHandle;
    }
}

创作打卡挑战赛

赢取流量/现金/CSDN周边激励大奖