调用Future.get()抛出ConcurrentModificationException问题求助

错误原因 #

ArrayList.subList()返回的是原列表的视图对象，并非独立的列表副本。它内部会引用原列表，并且在每次操作时检查原列表的修改计数（modCount）。当你的线程在处理这个子视图时，若原filterRecords列表被任何线程修改（包括主线程的结构变更），就会触发ConcurrentModificationException。从堆栈跟踪来看，异常发生在将SubList转为新ArrayList的过程中，此时原列表的modCount已和SubList记录的数值不一致。

修复代码 #

修改列表拆分的逻辑，将subList转换为独立的ArrayList副本，彻底和原列表解耦：

ConcurrentLinkedQueue<Future<Tuple<List<FixedLengthRecord>, BatchDataPoints>>> futures = new ConcurrentLinkedQueue<>();
List<List<Tuple<String, GenericRecord>>> partitions = new ArrayList<>();

int partitionSize = (filterRecords.size() / fixedLengthRecordEnrichmentThreadSize);
for (int i = 0; i < filterRecords.size(); i += partitionSize) {
    // 关键修改：创建新的ArrayList持有子列表元素，而非直接使用subList视图
    List<Tuple<String, GenericRecord>> sublist = new ArrayList<>(
        filterRecords.subList(i, Math.min(i + partitionSize, filterRecords.size()))
    );
    partitions.add(sublist);
}

for(int fixedLengthRecordEnrichmentThread = 0; fixedLengthRecordEnrichmentThread < this.fixedLengthRecordEnrichmentThreadSize; fixedLengthRecordEnrichmentThread++) {
    futures.offer(service.submit(new FixedLengthRecordWrapper(partitions.get(fixedLengthRecordEnrichmentThread),dataEnricher)));
}

// Future处理逻辑可保持不变
while(!futures.isEmpty()){
   Future<Tuple<List<FixedLengthRecord>, BatchDataPoints>> future = futures.poll();
   if (future.isDone() || future.isCancelled()) {
        try {
            Tuple<List<FixedLengthRecord>, BatchDataPoints> fixedLengthRecordsTuple = future.get();
            // 处理结果
        } catch (InterruptedException | ExecutionException e) {
            // 捕获并处理异常，避免程序崩溃
            e.printStackTrace();
        }
   }
}

额外注意事项 #

• 确保filterRecords在完成拆分后，不会被其他线程进行结构修改（如add/remove/clear），否则即使使用副本，若拆分过程中原列表变更，仍可能出现问题。
• 建议在Future处理时添加异常捕获逻辑，避免单个线程的异常导致整个流程终止。

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