批量投递队列高效实现问询：优化线程安全同步机制

你的场景是数千万行数据定时批量HTTP发送，当前用synchronized保证线程安全，但高并发下每次push都加锁会影响性能，下面给几个更高效优雅的实现方案：

1. 用并发队列替代手动同步的ArrayList #

Java提供的LinkedBlockingQueue/ArrayBlockingQueue本身就是线程安全的，不需要手动加锁，生产者的push操作直接调用队列方法即可，消费者定时批量取出所有元素处理，避免频繁锁竞争：

class Producer{
    ScheduledExecutorService service = new ScheduledThreadPoolExecutor(1);
    BlockingQueue<String> dataQueue = new LinkedBlockingQueue<>();

    public Producer() {
        // 定时执行处理逻辑
        service.scheduleAtFixedRate(this::processBatch, 0, 2000, TimeUnit.MILLISECONDS);
    }

    public void push(String str) {
        if (str != null && !str.isEmpty()) {
            dataQueue.offer(str); // 非阻塞添加，避免因队列满阻塞生产者
        }
    }

    private void processBatch(){
        List<String> batch = new ArrayList<>();
        // 一次性取出队列中所有元素，减少锁竞争次数
        dataQueue.drainTo(batch);
        if (batch.isEmpty()) {
            return;
        }
        // 数据转换
        List<String> convertedData = batch.stream()
                                         .map(this::convert) // 替换为你的转换逻辑
                                         .collect(Collectors.toList());
        // 执行HTTP发送
        sendHttp(convertedData);
    }

    // 模拟数据转换方法
    private String convert(String raw) {
        return raw;
    }

    // 模拟HTTP发送方法
    private void sendHttp(List<String> data) {
        // 发送逻辑
    }
}

2. 双缓冲无锁切换方案（高并发场景最优） #

用AtomicReference维护两个列表：一个用于接收生产者的数据，一个用于消费者处理。定时任务触发时，原子性地切换列表，生产者全程无锁写入，性能拉满：

class Producer{
    ScheduledExecutorService service = new ScheduledThreadPoolExecutor(1);
    // 原子引用维护当前接收数据的列表
    private final AtomicReference<List<String>> currentBuffer = new AtomicReference<>(new ArrayList<>());
    // 单独的线程池处理HTTP发送，避免阻塞定时任务
    private ExecutorService sendPool = Executors.newFixedThreadPool(4);

    public Producer() {
        service.scheduleAtFixedRate(this::processBatch, 0, 2000, TimeUnit.MILLISECONDS);
    }

    public void push(String str) {
        if (str != null && !str.isEmpty()) {
            // 直接往当前缓冲列表添加，无锁
            currentBuffer.get().add(str);
        }
    }

    private void processBatch(){
        // 原子切换：把当前缓冲替换为空列表，拿到要处理的批量数据
        List<String> batch = currentBuffer.getAndSet(new ArrayList<>());
        if (batch.isEmpty()) {
            return;
        }
        // 异步处理发送，不阻塞定时线程
        sendPool.submit(() -> {
            List<String> convertedData = batch.parallelStream()
                                             .map(this::convert)
                                             .collect(Collectors.toList());
            sendHttp(convertedData);
        });
    }

    private String convert(String raw) {
        return raw;
    }

    private void sendHttp(List<String> data) {
        // 发送逻辑
    }
}

额外优化点 #

• 定时任务选择：如果HTTP发送耗时可能超过2秒，建议用scheduleWithFixedDelay替代scheduleAtFixedRate，保证两次任务执行间隔固定为2秒，避免任务堆积。
• 线程池配置：发送HTTP是IO密集型操作，可根据服务器带宽、目标接口QPS调整发送线程池的大小，不要用默认的固定线程池。
• 内存控制：如果数据量极大，并发队列可以设置固定容量（比如ArrayBlockingQueue），避免内存溢出；双缓冲方案也可以限制单批次最大数据量（不过你要求必须定时，这个可以忽略）。

内容的提问来源于stack exchange，提问作者lant