[Java知识库]通过队列和线程池提升任务的执行效率

业务流程长，通过异步操作提升返回时间
首先确定流程是可以异步的，在进行一些业务检查和补偿确定异步是成功的
或者记录好日志 也方便排查

首先创建一个线城池的管理service

package com.reformer.invoice.service;

public interface ThreadPoolManagers<T> {

    /**
     * 任务提交执行
     * @param commonService
     * @param message
     */
    void execute(T commonService, String message);
    /**
     * 最大队列容量
     */
    int getMaxThreadQueueSize();
    /**
     * 核心容量
     */
    int getCorePoolSize();
    /**
     * 最大容量
     */
    int getMaxPoolSize();
    /**
     * 存活时间
     */
    int getKeepAliveTime();
    /**
     * 最大缓存队列容量
     */
    int getMaxCacheQueueSize();
    /**
     * 标识
     */
    boolean getOffline();

}

package com.reformer.invoice.service.impl;

import com.reformer.invoice.service.CommonService;
import com.reformer.invoice.service.ThreadPoolManagers;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

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



@Slf4j
@Service
public class ThreadPoolManagersImpl<T> implements ThreadPoolManagers<T> {

    /**
     * 缓存队列
     */
    private volatile LinkedBlockingQueue<Runnable> cacheLinkedQueue;
    /**
     * 最大队列容量
     */
    private final int MAX_THREAD_QUEUE_SIZE = 1000;
    /**
     * 核心容量
     */
    private final int CORE_POOL_SIZE = 20;
    /**
     * 最大容量
     */
    private final int MAX_POOL_SIZE = 40;
    /**
     * 存活时间
     */
    private final int KEEP_ALIVE_TIME = 0;
    /**
     * 最大缓存队列容量
     */
    private final int MAX_CACHE_QUEUE_SIZE = 10000;
    /**
     * 线程池
     */
    private volatile ThreadPoolExecutor threadPoolExecutor;
    /**
     * 标识
     */
    private volatile static boolean offline = false;


    public ThreadPoolManagersImpl() {
        getCacheQueue();
        getExecutorService();
    }

    @Override
    public void execute(T commonService, String message) {
        int size = getExecutorService().getQueue().size();
        log.info("线程池queue,队列size :" + size);
        if (offline) {
            addQueue((CommonService) commonService, message);
            log.info("缓存,队列开始缓存,size:{} ", cacheLinkedQueue.size());
            if (size == 0) {
                log.info("开始恢复");
                new OfflineResumeThread().start();
            }
            return;
        }
        if (size >= MAX_THREAD_QUEUE_SIZE) {
            setOffline(true);
            return;
        }
        getExecutorService().submit(addTask((CommonService) commonService, message));
    }

    private void addQueue(CommonService commonService, String message) {

        getCacheQueue().add(addTask(commonService, message));

    }

    /**
     * 生成任务
     *
     * @param commonService
     * @param message
     * @return
     */
    private Runnable addTask(CommonService commonService, String message) {
        return new Runnable() {
            @Override
            public void run() {
                commonService.process(message);
            }
        };
    }


    private ThreadPoolExecutor getExecutorService() {
        if (threadPoolExecutor == null) {
            synchronized (ThreadPoolManagersImpl.class) {
                if (threadPoolExecutor == null) {
                    threadPoolExecutor = new ThreadPoolExecutor(
                            CORE_POOL_SIZE
                            , MAX_POOL_SIZE
                            , KEEP_ALIVE_TIME
                            , TimeUnit.MILLISECONDS
                            , new LinkedBlockingQueue<>(MAX_THREAD_QUEUE_SIZE)
                            , new RejectedExecutionHandler() {
                        @Override
                        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                            log.info("线程被拒绝掉,核心线程队列容量{}", executor.getQueue().size());
                            // 可以做一些补偿机制
                        }
                    }
                    );
                }
            }
        }
        return threadPoolExecutor;
    }

    private LinkedBlockingQueue getCacheQueue() {
        if (cacheLinkedQueue == null) {
            synchronized (ThreadPoolManagersImpl.class) {
                if (cacheLinkedQueue == null) {
                    cacheLinkedQueue = new LinkedBlockingQueue<>(MAX_CACHE_QUEUE_SIZE);
                }
            }
        }
        return cacheLinkedQueue;
    }


    class OfflineResumeThread extends Thread {
        @Override
        public void run() {
            while (true) {
                if (getExecutorService().getQueue().size() >= MAX_THREAD_QUEUE_SIZE) {
                    continue;
                }
                Runnable runnable = null;
                try {
                    runnable = cacheLinkedQueue.take();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                log.info("移除 queue,size :{}", cacheLinkedQueue.size());
                if (runnable != null) {
                    getExecutorService().submit(runnable);
                }
                if (cacheLinkedQueue.size() == 0) {
                    setOffline(false);
                    log.info("恢复完毕");
                    return;
                }
            }
        }
    }


    @Override
    public int getMaxThreadQueueSize() {
        return MAX_THREAD_QUEUE_SIZE;
    }

    @Override
    public int getCorePoolSize() {
        return CORE_POOL_SIZE;
    }

    @Override
    public int getMaxPoolSize() {
        return MAX_POOL_SIZE;
    }

    @Override
    public int getKeepAliveTime() {
        return KEEP_ALIVE_TIME;
    }

    @Override
    public int getMaxCacheQueueSize() {
        return MAX_CACHE_QUEUE_SIZE;
    }

    @Override
    public boolean getOffline() {
        return offline;
    }

    private void setOffline(boolean offline) {
        ThreadPoolManagersImpl.offline = offline;
    }
}

CommonService 是我自己的业务实现类 ，以上就是操作就可以使程序性能提升