Java线程池自定义报警

当线程池任务过多，使用拒绝策略时，你也希望收到消息，现在实现这一功能 : )

@AllArgsConstructor
public class ThreadRejectTest {
    @SneakyThrows
    public static void main(String[] args) {
        AtomicLong atomicLong = new AtomicLong();
        //1.创建线程工厂
        ThreadFactory threadFactory = r -> {
            Thread t = new Thread(r);
            t.setName("coolT");
            t.setDaemon(true);
            return t;
        };
        int process = Runtime.getRuntime().availableProcessors() + 1;
        System.out.println("System core size " + process);
        //2.设置线程池 采用四种其中一种的拒绝策略
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                process + 1,
                process * 2,
                20L,
                TimeUnit.SECONDS,
                new linkedBlockingDeque<>(100),
                threadFactory,
                new ThreadPoolExecutor.AbortPolicy()
        );

		//3.将之前设置的拒绝策略变为拒绝策越代理类
        executor.setRejectedExecutionHandler(RejectedProxyUtil.createProxy(
                executor.getRejectedExecutionHandler(),
                "customThreadPool", atomicLong));

		//4.执行仍任务
        int taskNum = 1000;
        while (taskNum-->0){
            executor.execute(() -> {
                //doing tasks
                try {
                    TimeUnit.MICROSECONDS.sleep(100);
//                    System.out.println("拒绝数量: "+atomicLong);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        TimeUnit.SECONDS.sleep(1000000);
    }
}

//代理工具类  生成代理类返回
public class RejectedProxyUtil {

    public static RejectedExecutionHandler createProxy(RejectedExecutionHandler rejectedExecutionHandler,
                                                       String threadPoolId,
                                                       AtomicLong rejectCount){
        RejectedExecutionHandler rejectProxy = (RejectedExecutionHandler)Proxy.newProxyInstance(rejectedExecutionHandler.getClass().getClassLoader(),
                new Class[]{RejectedExecutionHandler.class},
                new RejectedProxyInvocationHandler(rejectedExecutionHandler, rejectCount, threadPoolId));
        return rejectProxy;
    }
}

//添加代理类中发送告警功能
@AllArgsConstructor
public class RejectedProxyInvocationHandler implements InvocationHandler {

    private Object target;

    private AtomicLong rejectCount;

    private final String threadPoolId;

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        rejectCount.getAndIncrement();
        // 发送警告信息 钉钉、邮箱...
        System.out.println("ALARM thread pool " + threadPoolId + " has rejected mission");
        //alarm.sendAlarm(threadPoolId);
        return method.invoke(target, args);
    }
}

Shift+F9执行