性能优化之-批处理提升吞吐量

public class BatchTool<T> {
    private ScheduledExecutorService scheduledExecutorService;
    private BlockingQueue<T> queue;
    private Consumer<List<T>> consumer;

    // 锁batch()方法，避免调用线程和定时线程并发执行
    private final Object batchLock = new Object();

    public BatchTool(Integer maxSize, Integer intervalTime, TimeUnit timeUnit, Consumer<List<T>> consumer) {
        // 参数校验

        this.queue = new LinkedBlockingQueue<>(maxSize);
        // 消费数据处理
        this.consumer = consumer;

        // 固定频率取走队列数据
        scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(r -> new Thread(r, "scheduledExecutor"));
        scheduledExecutorService.scheduleAtFixedRate(this::batch, intervalTime, intervalTime, timeUnit);
    }

    /**
     * 提交任务元素
     * @param e
     * @throws InterruptedException
     */
    public void put(T e) throws InterruptedException {
        // offer 加锁处理，满了放不进去返回false，则进行批处理
        if (!queue.offer(e)) {
            batch();
            // 批处理后再阻塞放进去该元素
            queue.put(e);
        }
    }

    /**
     * 取出队列数据
     * 封装调用方的批处理方法
     */
    private void batch() {
        List<T> list = new ArrayList<>();
        // 从 drainTo()方法的原子性，不加锁处理也不影响数据安全，最好加上
        // 尽量减少锁范围
        synchronized (batchLock) {
            if (queue.isEmpty()) {
                return;
            }
            // 内部加锁处理写到list
            queue.drainTo(list);
        }
        if (list.isEmpty()) {
            return;
        }
        try {
            // 调用方可异步处理，避免阻塞；或者调用方传入线程池参数；
            consumer.accept(list);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void shutdown() {
        // 队列剩余数据消费掉
        batch();
        if (scheduledExecutorService == null) {
            return;
        }
        try {
            // 温和关闭，不会再进来任务
            scheduledExecutorService.shutdown();
            // 等待线程池关闭（最多等待10秒）
            if (!scheduledExecutorService.awaitTermination(10, TimeUnit.SECONDS)) {
                // 强制关闭
                scheduledExecutorService.shutdownNow();
            }
        } catch (InterruptedException e) {
            scheduledExecutorService.shutdownNow();
        }
    }
}

public class BatchToolTest {

    @Data
    @AllArgsConstructor
    static class Obj {
        private String name;
    }

    /**
     * 调用方处理方法
     * @param objList
     */
    public void consume(List<Obj> objList) {
        if (CollectionUtils.isEmpty(objList)) {
            return;
        }
        objList.forEach(e -> {
            System.out.println(e + " " + Thread.currentThread().getName() + " " + System.currentTimeMillis());
        });
    }

    /**
     * @param args
     * @throws InterruptedException
     * 输出：
     * 前5个元素是主线程输出
     * 后5个元素是定时线程输出
     */
    public static void main(String[] args) throws InterruptedException {
        BatchToolTest batchToolTest = new BatchToolTest();

        BatchTool<Obj> objBatchTool = new BatchTool<>(5, 10, TimeUnit.SECONDS, batchToolTest::consume);

        for (int i = 0; i < 10; i++) {
            objBatchTool.put(new Obj(String.valueOf(i)));
        }

        // 测试到时间读队列
        TimeUnit.SECONDS.sleep(15);

        objBatchTool.shutdown();
    }
}