利用线程异步调用

Posted 2021-02-28 lingduqianli

定义线程类

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.util.ArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class ThreadFixedPool {

    private static final Logger logger = LogManager.getLogger(ThreadFixedPool.class);

    private static final int CORE_POOL_SIZE = 4;
    private static final int MAX_POOL_SIZE = 200;

    private static ExecutorService executorService;
    ArrayList<Runnable> runnables = new ArrayList<>();

    public ThreadFixedPool build() {
        if (null == executorService) {
            logger.debug("加载默认线程池,初始化线程数为:{}", CORE_POOL_SIZE);
            executorService = new ThreadPoolExecutor(CORE_POOL_SIZE, MAX_POOL_SIZE, 0L, TimeUnit.MILLISECONDS,
                    new LinkedBlockingQueue<Runnable>(1024), new AipcThreadFactory(),
                    new ThreadPoolExecutor.AbortPolicy());
        }
        return this;
    }

    public ThreadFixedPool build(int poolSize) {
        if (null == executorService) {
            if (poolSize == 0) {
                logger.warn("线程池异常 poolSize!");
            }
            if (poolSize < 1 || poolSize >= MAX_POOL_SIZE) {
                logger.warn("线程数不合法");
            }
            logger.debug("加载自定义线程池,初始化线程数为:{}", poolSize);
            executorService = new ThreadPoolExecutor(CORE_POOL_SIZE, MAX_POOL_SIZE, 0L, TimeUnit.MILLISECONDS,
                    new LinkedBlockingQueue<Runnable>(1024), new AipcThreadFactory(),
                    new ThreadPoolExecutor.AbortPolicy());
        }
        return this;
    }

    public void execute(Runnable runnable) {
        executorService.execute(runnable);
    }

    public void shutdown() {
        if (runnables.size() > 0) {
            for (Runnable r : runnables) {
                executorService.execute(r);
            }
        }
        try {
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException e) {
            logger.error("线程池shutdown 休眠异常", e);
        }
        executorService.shutdown();
    }
}

　方法体进行调用

new ThreadFixedPool().build().execute(new getProduct(products, product.getCustomerId()));

　　异步方法体进行处理

private class getProduct extends Thread {
        List<ProductListRsp> listRsp;
        String customerId = "";

        getProduct(List<ProductListRsp> listRsp, String customerId) {
            this.listRsp = listRsp;
            this.customerId = customerId;
        }

        @Override
        public void run() {
            for (ProductListRsp productListRsp : listRsp) {
                ChangeVo vo = new ChangeVo();
                vo.setAmt(1);
                vo.setCustomerId(customerId);
                vo.setId(productListRsp.getId());
                nearService.getProduct(vo);
            }
        }
    }