使用ExecutorService并行处理WorldWind Shapefile时抛出IllegalArgumentException

单线程处理Shapefile时一切正常，但使用ExecutorService并行处理以提升效率时，调用record.getCompoundPointBuffer().getPositions().iterator()抛出如下异常：

Caused by: java.lang.IllegalArgumentException: newPosition < 0: (5044 < 0)
        at java.base/java.nio.Buffer.createPositionException(Buffer.java:352)
        at java.base/java.nio.Buffer.position(Buffer.java:327)
        at java.base/java.nio.IntBuffer.position(IntBuffer.java:1551)
        at java.base/java.nio.IntBuffer.position(IntBuffer.java:285)
        at gov.nasa.worldwind.util.CompoundVecBuffer.<init>(Unknown Source)
        at gov.nasa.worldwind.util.VecBufferSequence.<init>(Unknown Source)
        at gov.nasa.worldwind.util.VecBufferSequence.createSlice(Unknown Source)
        at gov.nasa.worldwind.util.CompoundVecBuffer.slice(Unknown Source)

原始并行处理代码：

try {
       Shapefile shapefile = new Shapefile(shpFileStream, shxFileStream, dbfFileStream, prjFileStream);
       
       List<Callable<String>> tasks = new ArrayList<>();
       List<ShapefileRecord> records = new ArrayList<>();
       while (shapefile.hasNext()) { // iterate shapefile records
            ShapefileRecord record = shapefile.nextRecord();
            records.add(record);
            if(records.size() == 5000) { // create tasks with bunch of 5000 records
                    MyParallelTask task = new MyParallelTask (new ArrayList<>(records),...);
                    tasks.add(task);
                    records.clear();
            }
      }

      if(!records.isEmpty()) { // create another task if some more records pending
            MyParallelTask task = new MyParallelTask (new ArrayList<>(records),...);
            tasks.add(task);
            records.clear();
       }

       try(ExecutorService executorService = Executors.newVirtualThreadPerTaskExecutor()) {
            List<Future<String>> shapeFilePersistFutures = executorService.invokeAll(tasks);
            for (Future<String> shapeFilePersistFuture : shapeFilePersistFutures) {
                 shapeFilePersistFuture.get();
            }
            executorService.shutdown();
       }
 } catch(Exception e) {
      e.printStackTrace();
      throw new CustomException(....);
 }

问题根源 #

NASA WorldWind的ShapefileRecord及其内部依赖的CompoundVecBuffer、NIO Buffer类不是线程安全的。虽然你在创建任务时复制了records列表，但ShapefileRecord内部的Buffer是共享引用，并非深拷贝。多线程并行访问时，多个线程会同时修改Buffer的position等状态，导致出现非法的位置值（比如负数）。

解决方案 #

最可靠的修复方式是在单线程阶段提前提取所有几何数据，将纯数据（如Position列表）交给并行任务处理，避免多线程直接访问WorldWind的非线程安全结构。

修改后的代码示例：

try {
    Shapefile shapefile = new Shapefile(shpFileStream, shxFileStream, dbfFileStream, prjFileStream);
    
    List<Callable<String>> tasks = new ArrayList<>();
    List<Position> currentBatch = new ArrayList<>();

    while (shapefile.hasNext()) {
        ShapefileRecord record = shapefile.nextRecord();
        // 单线程下提前提取所有位置数据，避免多线程共享Buffer
        Iterator<Position> posIterator = record.getCompoundPointBuffer().getPositions().iterator();
        while (posIterator.hasNext()) {
            currentBatch.add(posIterator.next());
        }

        // 达到批量大小就创建任务
        if (currentBatch.size() >= 5000) {
            tasks.add(new MyParallelTask(new ArrayList<>(currentBatch), ...));
            currentBatch.clear();
        }
    }

    // 处理剩余数据
    if (!currentBatch.isEmpty()) {
        tasks.add(new MyParallelTask(new ArrayList<>(currentBatch), ...));
        currentBatch.clear();
    }

    // 并行执行任务
    try(ExecutorService executorService = Executors.newVirtualThreadPerTaskExecutor()) {
        List<Future<String>> futures = executorService.invokeAll(tasks);
        for (Future<String> future : futures) {
            future.get();
        }
    }
} catch(Exception e) {
    e.printStackTrace();
    throw new CustomException(...);
}

说明 #

• 所有对WorldWind内部非线程安全组件的访问（如getCompoundPointBuffer()）都在单线程完成，避免了并发修改问题。
• 并行任务仅处理独立的Position数据，完全隔离了线程间的共享状态，从根源上消除了竞态条件。

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