Groovy基于CompletableFutures的异步Sheet合并方法失效排查

我实现了一个SheetMerger类，用于异步合并从不同Sheet异步获取的SheetModel对象。部分Sheet包含数千行数据，部分仅含少量行。预期按任务完成顺序合并结果，但运行时看不到任何Sheet合并完成的日志，实际合并逻辑也未执行。

完整代码 #

SheetMerger类 #

public class SheetMerger {
    public SheetModel merge(List<Future<SheetModel>> promises) { 
        List<String> mergedInColumnNames = [];
        String keyColumnName;
        List<Map<String, Object>> mergedInData = [];
        
        CompletableFuture[] promisesArray = new CompletableFuture[promises.size()];
        for (int idx = 0; idx < promisesArray.size(); idx++)
            promisesArray[idx] = promises[idx];
        
        CompletableFuture.anyOf(promisesArray).get();
        
        promisesArray.eachWithIndex { CompletableFuture<SheetModel> promise, int idx ->
            println "Starting to merge Sheet number ${idx + 1}"
            
            return promise.thenAcceptAsync { SheetModel model ->
                println "Merging Sheet number ${idx + 1}"
                
                mergedInColumnNames = CollectionUtils.MergeLists(mergedInColumnNames, model.columnNames);
                
                if (keyColumnName == null)
                    keyColumnName = model.keyColumnName;
                
                mergedInData = mergeIntoDataSource(mergedInData, keyColumnName, model);
                
                println "Sheet number ${idx + 1} merged"
            }
        
        }
        
        CompletableFuture.allOf(promisesArray).get();
            
        
        return new SheetModel(mergedInColumnNames, keyColumnName, mergedInData);
    }
    
    public List<Map<String, Object>> mergeIntoDataSource(List<Map<String, Object>> masterDataSource, String masterKey, SheetModel branchModel) {  
        
        branchModel.data.eachWithIndex { Map<String, Object> branchRow, int idx ->  
            Map<String, Object> masterRow = branchRow.findResult { String key, Object value -> 
                if (!key.equals(branchModel.keyColumnName))
                    return null;
                
                return masterDataSource
                    .find { Map<String, Object> row -> return row[masterKey].equals(value) }
            }
            
            if (masterRow == null) {
                masterDataSource.push(branchRow);
                return;
            }
                
            masterDataSource[idx] = CollectionUtils.MergeMaps(masterRow, branchRow);
        }
        
        
        return masterDataSource;
    }
    
}

CollectionUtils类 #

public final class CollectionUtils {
    /**
     * **TODO**: bring in some callback that takes masterMap, branchMap, key, and defines a merge strategy
     * @param masterMap
     * @param branchMap
     * @return
     */
    public static Map MergeMaps(Map masterMap, Map branchMap) { 
        final List<String> listOfKeys = this.MergeLists(masterMap.keySet().toList(), 
            branchMap.keySet().toList());
        
        for (String key : listOfKeys) { 
            if (!branchMap.containsKey(key))
                continue;
                
            final Object branchValue = branchMap[key]; 
            if ((branchValue == null) || (branchValue.toString().trim().equals("")))
                continue;
            
            if ((!masterMap.containsKey(key)) || (masterMap[key].toString().trim().equals("")))
                masterMap[key] = branchValue;
        }
        
        return masterMap;
    }
    
    public static List MergeLists(List masterList, List... branchLists) { 
        return masterList.plus(branchLists)
            .flatten()
            .unique()
    }
}

运行日志 #

reading the Sheets...
merging the Sheets..
Reading Sheet 'ZohoSPOS Reconciliation' in Excel file 'C:\Users\Eliza\Desktop\Upwork jobs\job\com.job.sposZohoMergeScript\excelSourceFiles\zoho_spos_balance-11_42_20_01_2023_WORKING.xlsx'...
Reading Sheet 'Corrections' in Excel file 'C:\Users\Eliza\Desktop\Upwork jobs\job\com.job.sposZohoMergeScript\excelSourceFiles\zoho-spos-balance-changes-v3.xlsx'...
Reading Sheet 'ZohoSPOS Reconciliation' in Excel file 'C:\Users\Eliza\Desktop\Upwork jobs\job\com.job.sposZohoMergeScript\excelSourceFiles\zoho_spos_balance-08_25_30_12_2022.xlERROR StatusLogger Log4j2 could not find a logging implementation. Please add log4j-core to the classpath. Using SimpleLogger to log to the console...
Sheet 'Corrections' in Excel file 'C:\Users\Eliza\Desktop\Upwork jobs\job\com.job.sposZohoMergeScript\excelSourceFiles\zoho-spos-balance-changes-v3.xlsx' has been parsed!
Reading Sheet 'Robert' in Excel file 'C:\Users\Eliza\Desktop\Upwork jobs\job\com.job.sposZohoMergeScript\excelSourceFiles\zoho-spos-balance-changes-v1.xlsx'...
Starting to merge Sheet number 1
Starting to merge Sheet number 2
Starting to merge Sheet number 3
Starting to merge Sheet number 4
Starting to merge Sheet number 5
Sheet 'Robert' in Excel file 'C:\Users\Eliza\Desktop\Upwork jobs\job\com.job.sposZohoMergeScript\excelSourceFiles\zoho-spos-balance-changes-v1.xlsx' has been parsed!
Merging Sheet number 4
Reading Sheet 'Sheet1' in Excel file 'C:\Users\Eliza\Desktop\Upwork jobs\job\com.job.sposZohoMergeScript\excelSourceFiles\zoho-spos-balance-changes-v1.xlsx'...
Sheet 'Sheet1' in Excel file 'C:\Users\Eliza\Desktop\Upwork jobs\job\com.job.sposZohoMergeScript\excelSourceFiles\zoho-spos-balance-changes-v1.xlsx' has been parsed!
Merging Sheet number 5
Merging Sheet number 2
Sheet 'ZohoSPOS Reconciliation' in Excel file 'C:\Users\Eliza\Desktop\Upwork jobs\job\com.job.sposZohoMergeScript\excelSourceFiles\zoho_spos_balance-08_25_30_12_2022.xlsx' has been parsed!
Merging Sheet number 3
Sheet 'ZohoSPOS Reconciliation' in Excel file 'C:\Users\Eliza\Desktop\Upwork jobs\job\com.job.sposZohoMergeScript\excelSourceFiles\zoho_spos_balance-11_42_20_01_2023_WORKING.xlsx' has been parsed!
Merging Sheet number 1

问题原因与修复方案 #

核心问题 #

1. 合并回调未被等待：thenAcceptAsync返回的新CompletableFuture未被保存，CompletableFuture.allOf(promisesArray).get()仅等待原始Sheet解析任务完成，不等待合并逻辑的异步回调执行。
2. 并发修改风险：合并操作直接修改共享集合，无同步机制，会导致数据竞争和结果不一致。
3. 行索引错误：mergeIntoDataSource中用分支Sheet的行索引替换主数据源行，索引不匹配会覆盖错误数据。

修复步骤 #

1. 等待合并回调完成

保存合并任务的CompletableFuture，并等待所有合并逻辑执行完毕：

public SheetModel merge(List<Future<SheetModel>> promises) { 
    List<String> mergedInColumnNames = Collections.synchronizedList(new ArrayList<>());
    String keyColumnName;
    List<Map<String, Object>> mergedInData = Collections.synchronizedList(new ArrayList<>());
    Object lock = new Object();
    List<CompletableFuture<Void>> mergeFutures = new ArrayList<>();

    CompletableFuture[] promisesArray = promises.toArray(new CompletableFuture[0]);
    
    promisesArray.eachWithIndex { CompletableFuture<SheetModel> promise, int idx ->
        println "Starting to merge Sheet number ${idx + 1}"
        
        CompletableFuture<Void> mergeFuture = promise.thenAcceptAsync { SheetModel model ->
            println "Merging Sheet number ${idx + 1}"
            
            synchronized (lock) {
                mergedInColumnNames.addAll(model.columnNames);
                mergedInColumnNames = new ArrayList<>(mergedInColumnNames.unique());
                if (keyColumnName == null)
                    keyColumnName = model.keyColumnName;
            }
            
            mergeIntoDataSource(mergedInData, keyColumnName, model);
            
            println "Sheet number ${idx + 1} merged"
        }
        mergeFutures.add(mergeFuture);
    }
    
    CompletableFuture.allOf(promisesArray).get();
    CompletableFuture.allOf(mergeFutures.toArray(new CompletableFuture[0])).get();
        
    return new SheetModel(mergedInColumnNames, keyColumnName, mergedInData);
}

2. 修复行索引错误

public List<Map<String, Object>> mergeIntoDataSource(List<Map<String, Object>> masterDataSource, String masterKey, SheetModel branchModel) {  
    branchModel.data.each { Map<String, Object> branchRow ->  
        Map<String, Object> masterRow = masterDataSource.find { row -> 
            row[masterKey] == branchRow[branchModel.keyColumnName]
        }
        
        if (masterRow == null) {
            masterDataSource.add(new HashMap<>(branchRow));
            return;
        }
            
        CollectionUtils.MergeMaps(masterRow, branchRow);
    }
    return masterDataSource;
}

3. 避免修改原始集合

修改工具类方法，基于副本操作避免并发问题：

public static List MergeLists(List masterList, List... branchLists) { 
    List merged = new ArrayList<>(masterList);
    for (List branch : branchLists) {
        merged.addAll(branch);
    }
    return merged.unique();
}

public static Map MergeMaps(Map masterMap, Map branchMap) { 
    Map merged = new HashMap<>(masterMap);
    final List<String> listOfKeys = MergeLists(branchMap.keySet().toList());
    
    for (String key : listOfKeys) { 
        Object branchValue = branchMap[key]; 
        if (branchValue == null || branchValue.toString().trim().isEmpty())
            continue;
        
        if (!merged.containsKey(key) || merged.get(key).toString().trim().isEmpty()) {
            merged.put(key, branchValue);
        }
    }
    return merged;
}

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