Java 中怎么用内存队列 BlockingQueue 实现异步生产者消费批量刷库-java教程-PHP中文网

用blockingqueue实现异步生产者-消费者批量刷库，核心是生产者入队、消费者批量拉取攒批写库；推荐arrayblockingqueue（有界防oom）或linkedblockingqueue（可设界）；生产者用offer超时避免阻塞，消费者用drainto批量获取并控制最小等待时间触发刷库。

java 中怎么用内存队列 blockingqueue 实现异步生产者消费批量刷库

用 BlockingQueue 实现异步生产者-消费者批量刷库，核心是：生产者往队列塞数据，消费者从队列批量拉取、攒够一批再统一写入数据库，避免频繁单条 SQL 带来的性能瓶颈。

选择合适的 BlockingQueue 实现

推荐使用 ArrayBlockingQueue 或 LinkedBlockingQueue：

ArrayBlockingQueue：有界队列，防止内存无限增长，适合对吞吐和稳定性要求高的场景；设置容量时建议根据平均写入速率和容忍延迟反推（例如每秒 1000 条，最大容忍 2 秒积压 → 容量 ≥ 2000）
LinkedBlockingQueue：默认无界（注意 OOM 风险），若指定容量则行为类似 Array 版；适合写入波动大、允许短暂积压的场景

定义数据结构与队列实例

先定义要入库的业务对象（如订单、日志等），再声明线程安全的队列：

Java Development Manual

Java开发手册规约集合，基于阿里巴巴Java开发手册（嵩山版）。涵盖7大维度：编程规约、异常日志、单元测试、安全规约、MySQL数据库、工程结构、设计规约。当用户需要：(1) 编写或审查Java代码 (2) 检查命名/代码规范 (3) 处理异常和日志 (4) 编写单元测试 (5) 安全编码 (6) 数据库设...

下载

public class OrderEvent {
    private long orderId;
    private String status;
    private long createTime;
    // getter/setter...
}

// 队列容量设为 10000，公平模式可选
private final BlockingQueue<orderevent> queue = new ArrayBlockingQueue(10000, true);
</orderevent>

生产者：非阻塞或带超时地提交

避免生产者因队列满而长时间卡住，建议用带超时的 offer()：

成功：数据入队，继续下一条
失败（超时/满）：可降级处理，比如记录告警、写本地文件暂存、或直接丢弃（按业务容忍度决定）

if (!queue.offer(event, 100, TimeUnit.MILLISECONDS)) {
    log.warn("OrderEvent dropped due to queue full: {}", event.getOrderId());
}

消费者：批量拉取 + 批量执行 SQL

消费者用一个独立线程循环执行，每次尝试拉取固定数量（如 100 条），并控制最小等待时间防空转：

用 drainTo(Collection, maxElements) 高效批量出队（比循环 poll 更快且线程安全）
攒够一批（如 ≥ 50 条）或等待超时（如 200ms）就触发刷库，平衡延迟与吞吐
JDBC 批量插入推荐用 PreparedStatement.addBatch() + executeBatch()，开启 rewriteBatchedStatements=true（MySQL）提升性能

new Thread(() -> {
    List<orderevent> batch = new ArrayList(100);
    while (running) {
        batch.clear();
        // 尝试一次拉最多 100 条，立即返回（不等）
        queue.drainTo(batch, 100);
        if (batch.isEmpty()) {
            // 没拉到，小睡一下避免忙等
            LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(20));
            continue;
        }
        try {
            orderMapper.insertBatch(batch); // MyBatis 批量插入
        } catch (Exception e) {
            log.error("Failed to flush batch", e);
            // 可重试或落盘补偿，视业务而定
        }
    }
}).start();
</orderevent>