记一次利用Semaphore处理大批次数据计算的解决方案

/**
 * 2022-01-22更新: 尽量不要用DataBatchUtils batchUtils = new DataBatchUtils(list, INIT_NUM)的方式进行创建,
 * 这样做的效率很差!使用（guava的）DataBatchUtils.batchInsert会好很多
 * @author 94391
 */
public class DataBatchUtils<E> implements Iterator<List<E>> {

    private int batchSize;

    private List<E> origin;

    private int index = 0;

    private List<E> result;

    private int size = 0;

    public DataBatchUtils(List<E> origin, int batchSize) {
        if (0 >= batchSize) {
            throw new RuntimeException();
        }

        this.batchSize = batchSize;
        this.origin = origin;
        this.size = null == origin ? 0 : origin.size();
        result = new ArrayList<E>(batchSize);
    }

    @Override
    public boolean hasNext() {
        return index < size;
    }

    @Override
    public List<E> next() {
        result.clear();
        for (int i = 0; i < batchSize && index < size; i++) {
            result.add(origin.get(index++));
        }
        return result;
    }

    @Override
    public void remove() {
        throw new UnsupportedOperationException();
    }

    /**
     * 批量插入ByGuava
     * @param origin    需要批量操作的数据信任List
     * @param batchSize 分批数(每多少条分割处理一次)
     * @param <T>
     * @return
     */
    public static <T> List<List<T>> batchInsert(List<T> origin, int batchSize) {
        Iterable<List<T>> subSets = Iterables.partition(origin, batchSize);
        Iterator<List<T>> partitions = subSets.iterator();
        List<List<T>> list = new LinkedList<>();
        while (partitions.hasNext()) {
            List<T> sub = partitions.next();
            list.add(sub);
        }
        return list;
    }
}

/**
 * @author 94391
 */
public class ThreadPoolExecutorBuilder {

    private int corePoolSize;
    private int maximumPoolSize;
    private long keepAliveTime;
    private TimeUnit unit;
    private BlockingQueue<Runnable> workQueue;
    private RejectedExecutionHandler handler;
    private ThreadFactory threadFactory;
    private static AtomicInteger threadId = new AtomicInteger();

    /**
     * 默认构造方法设置默认参数
     */
    public ThreadPoolExecutorBuilder() {

        int processors = Runtime.getRuntime().availableProcessors();

        // 核心工作线程以及最大空闲线程数目
        this.corePoolSize = processors;

        this.maximumPoolSize = 2 * processors;

        // 空闲线程的最大存活时间（注意参数生效的条件）
        // 当线程池中线程数量大于corePoolSize（核心线程数量）或设置了allowCoreThreadTimeOut（是否允许空闲核心线程超时）时，
        // 线程会根据keepAliveTime的值进行活性检查，一旦超时便销毁线程
        this.keepAliveTime = 8;

        this.unit = TimeUnit.SECONDS;

        // 任务队列以及饱和处理策略
        this.workQueue = new ArrayBlockingQueue<>(200);
        this.handler = new ThreadPoolExecutor.AbortPolicy();

        // 创建线程的工厂，一般可以用来对线程命名
        this.threadFactory = new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "threadPoolWorker-" + threadId.getAndIncrement());
            }
        };
    }

    public ThreadPoolExecutor build() {
        return new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
    }

    public ThreadPoolExecutorBuilder setCapacity(int corePoolSize, int maximumPoolSize) {
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        return this;
    }

    public ThreadPoolExecutorBuilder setKeepAliveTime(long keepAliveTime, TimeUnit unit) {
        this.keepAliveTime = keepAliveTime;
        this.unit = unit;
        return this;
    }

    public ThreadPoolExecutorBuilder setWorkQueueAndRejectHandler(BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
        this.workQueue = workQueue;
        this.handler = handler;
        return this;
    }

    public ThreadPoolExecutorBuilder setThreadFactory(ThreadFactory threadFactory) {
        this.threadFactory = threadFactory;
        return this;
    }
}