如何基于流实现Java MongoDB的批量插入操作？

当然可以解决，核心思路是分批流式插入，不用把所有数据一次性加载到内存里。具体实现方式如下：

• 核心逻辑：
  不要一次性把数百万条数据都塞进List，而是从数据源流式读取数据，每攒够一批（比如1000条）就调用insertMany插入，插入完成后清空当前批次的List，释放内存，继续处理下一批。

• Java驱动示例代码：
  假设你有一个数据输入流（比如从大文件、外部数据库导出的流），可以这么写：

  // 获取目标集合
  MongoCollection<Document> collection = mongoDb.getCollection("target_col");
  
  // 定义批次大小，根据内存情况调整，推荐1000-10000条
  int batchSize = 1000;
  List<Document> batchBuffer = new ArrayList<>(batchSize);
  
  // 流式读取数据（这里用伪代码模拟实际的数据流）
  try (Stream<YourDataModel> dataStream = fetchDataAsStream()) {
      dataStream.forEach(data -> {
          // 把自定义数据模型转成MongoDB Document
          Document doc = new Document()
              .append("field1", data.getField1())
              .append("field2", data.getField2());
          
          batchBuffer.add(doc);
          
          // 达到批次上限就执行插入
          if (batchBuffer.size() >= batchSize) {
              collection.insertMany(batchBuffer);
              batchBuffer.clear(); // 清空缓冲区，释放内存
          }
      });
      
      // 处理最后一批不足batchSize的数据
      if (!batchBuffer.isEmpty()) {
          collection.insertMany(batchBuffer);
      }
  } catch (IOException | MongoException e) {
      // 这里可以加重试逻辑，避免某批次插入失败导致数据丢失
      e.printStackTrace();
  }
  

• 关键注意点：

  • 批次大小要根据你的内存配置调整：太大容易触发OOM，太小会增加网络请求次数拖慢速度，建议先测试1000条的情况再微调。
  • 数据源的流式读取要做好：比如读大文件用BufferedReader逐行处理，读数据库用JDBC的流式查询（Statement.setFetchSize(Integer.MIN_VALUE)）。
  • 异常处理要到位：如果某批次插入失败，最好把这批数据暂存下来，后续重试，避免丢数据。
  • 可以调整MongoDB的写关注：如果对一致性要求不高，用WriteConcern.W1（只确认主节点写入成功）比默认的WriteConcern.ACKNOWLEDGED速度更快。

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