用BlockingQueue实现异步生产者-消费者批量刷库，核心是生产者入队、消费者批量拉取攒批写库；推荐ArrayBlockingQueue（有界防OOM）或LinkedBlockingQueue（可设界）；生产者用offer超时避免阻塞，消费者用drainTo批量获取并控制最小等待时间触发刷库。
用 BlockingQueue 实现异步生产者-消费者批量刷库，核心是：生产者往队列塞数据，消费者从队列批量拉取、攒够一批再统一写入数据库，避免频繁单条 SQL 带来的性能瓶颈。
选择合适的 BlockingQueue 实现
推荐使用 ArrayBlockingQueue 或 LinkedBlockingQueue：
-
ArrayBlockingQueue：有界队列，防止内存无限增长，适合对吞吐和稳定性要求高的场景；设置容量时建议根据平均写入速率和容忍延迟反推（例如每秒 1000 条，最大容忍 2 秒积压 → 容量 ≥ 2000） -
LinkedBlockingQueue：默认无界（注意 OOM 风险），若指定容量则行为类似 Array 版；适合写入波动大、允许短暂积压的场景
定义数据结构与队列实例
先定义要入库的业务对象（如订单、日志等），再声明线程安全的队列：
public class OrderEvent {
private long orderId;
private String status;
private long createTime;
// getter/setter...
}
// 队列容量设为 10000，公平模式可选
private final BlockingQueue<OrderEvent> queue = new ArrayBlockingQueue<>(10000, true);
生产者：非阻塞或带超时地提交
避免生产者因队列满而长时间卡住，建议用带超时的 offer()：
Java Development Manual
下载
Java开发手册规约集合，基于阿里巴巴Java开发手册（嵩山版）。 涵盖7大维度：编程规约、异常日志、单元测试、安全规约、MySQL数据库、工程结构、设计规约。 当用户需要：(1) 编写或审查Java代码 (2) 检查命名/代码规范 (3) 处理异常和日志 (4) 编写单元测试 (5) 安全编码 (6) 数据库设...
立即学习“Java免费学习笔记（深入）”；
- 成功：数据入队，继续下一条
- 失败（超时/满）：可降级处理，比如记录告警、写本地文件暂存、或直接丢弃（按业务容忍度决定）
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();
补充要点
收尾时记得优雅关闭：
- 设置标志位
running = false，消费完剩余数据再退出 - 生产者停止后，可调用
queue.drainTo()清空残留 - 数据库连接、事务、连接池等资源需配合生命周期管理（如用 try-with-resources 或 Spring 的 @Transactional）
