背景

我想实现一个合并请求工具类，思路：

1. 把请求放入请求队列，阻塞当前线程 t1 [ Locksupport.park ]
2. 线程 t2 收集一个批次的请求，提交到线程池
3. 线程 tn 执行批量处理，逐个设置返回值，然后逐个唤醒阻塞线程[t1...] [ Locksupport.unpark(t1)]

遇到的问题

测试时发现，打印每个线程的返回值时，偶尔会遇到返回值为 null 。

源码

传递的请求封装对象：

class BizTask {
    private Thread thread;
    private Object param;
    private Object response;

    public Object getParam() {
        return param;
    }

    public void setParam(Object param) {
        this.param = param;
    }

    public Object getResponse() {
        return response;
    }

    public void setResponse(Object response) {
        this.response = response;
    }

    public Thread getThread() {
        return thread;
    }

    public void setThread(Thread thread) {
        this.thread = thread;
    }
}

工具类：

public class BatchHelper {

    private final ThreadPoolExecutor threadPool = new ThreadPoolExecutor(10, 10, 20L, TimeUnit.MINUTES, new LinkedBlockingQueue<>(100000), new ThreadPoolExecutor.CallerRunsPolicy());
    private final BlockingQueue<BizTask> requestQueue = new LinkedBlockingQueue<>();
    private final Consumer<List<BizTask>> function;

    public BatchHelper(Consumer<List<BizTask>> function) {
        this.function = function;
        threadPool.execute(this::autoDispatch);
    }

    public Object take(Object param) {
        BizTask task = new BizTask();
        task.setParam(param);
        task.setThread(Thread.currentThread());
        // synchronized (task) {
        try {
            requestQueue.put(task);
        } catch (InterruptedException ignored) {
        }
        // 阻塞
        LockSupport.park();
        // }
        return task.getResponse();
    }

    public void autoDispatch() {
        while (true) {
            try {
                BizTask t1 = requestQueue.take();
                List<BizTask> tasks = new ArrayList<>(128);
                tasks.add(t1);
                BizTask t2 = requestQueue.poll();
                if (t2 == null) {
                    dispatch(tasks);
                    continue;
                }
                int sum = 1;
                while (t2 != null) {
                    tasks.add(t2);
                    sum = (sum + 1) & 127;
                    if (sum == 0) {
                        dispatch(tasks);
                        tasks = new ArrayList<>(128);
                    }
                    t2 = requestQueue.poll();
                }
                if (sum > 0) {
                    dispatch(tasks);
                }
                Thread.sleep(10);
            } catch (InterruptedException ignored) {
            }
        }
    }

    private void dispatch(List<BizTask> list) {
        threadPool.execute(() -> {
            // 批量处理，设置返回值
            function.accept(list);
            for (BizTask task : list) {
                // synchronized (task) {
                Thread lock = task.getThread();
                // 唤醒
                LockSupport.unpark(lock);
                // }
            }
        });
    }
}

测试案例：

public class TestDemo {
    public static void main(String[] args) {
        BatchHelper batchHelper = new BatchHelper((tasks) -> {
            for (int i = 0, tasksSize = tasks.size(); i < tasksSize; i++) {
                BizTask task = tasks.get(i);
                // System.out.println("requestId = " + requestId);
                task.setResponse(i);
            }
        });

        // 模拟请求线程
        ThreadPoolExecutor threadPool = new ThreadPoolExecutor(20, 20, 20L, TimeUnit.MINUTES, new LinkedBlockingQueue<>(1000), new ThreadPoolExecutor.CallerRunsPolicy());
        for (int i = 0; i < 100; i++) {
            int finalI = i;
            threadPool.execute(() -> {
                Object take = batchHelper.take(finalI + "");
                if (take == null) {
                    System.out.println("take = " + take);
                }
            });
        }


    }
}