如何并行处理HashMap中HashSet键与ArrayList值的对应元素

看起来你踩了两个坑：一是HashSet的无序性导致元素没法和ArrayList的位置对应，二是直接把整个集合传给方法而不是逐个元素配对。我给你整理两个可行的方案，帮你实现并行调用的需求：

第一步：先解决顺序匹配的问题 #

HashSet是无序的，这意味着你插入的B1/B2/B3在集合里的顺序是不确定的，根本没法和R1/R2/R3的位置一一对应。所以首先要把存储批次ID的集合换成LinkedHashSet，它会严格保留你的插入顺序：

Set<String> hs_batchesId= new LinkedHashSet<String>(); // 替换成LinkedHashSet
hs_batchesId.add("B1");
hs_batchesId.add("B2");
hs_batchesId.add("B3");

第二步：配对元素并并行调用方法 #

接下来我们需要把每个Map条目里的批次ID和结果ID按位置配对，然后用并行流来处理每一对，实现无需等待前一次执行的效果：

方案1：Java 9+原生实现（无需第三方库）

如果你的项目用的是Java 9及以上，可以直接用Stream.zip来合并两个流，代码更简洁：

String jobId = "J1";
Map<Set<String>, List<String>> map_batch_result_details = new HashMap<>();

// 初始化部分用LinkedHashSet保证顺序
Set<String> hs_batchesId = new LinkedHashSet<>();
List<String> list_resultId = new ArrayList<>();
hs_batchesId.add("B1");
hs_batchesId.add("B2");
hs_batchesId.add("B3");
list_resultId.add("R1");
list_resultId.add("R2");
list_resultId.add("R3");
map_batch_result_details.put(hs_batchesId, list_resultId);

// 处理每个Map条目
map_batch_result_details.entrySet().forEach(entry -> {
    // 把LinkedHashSet转成有序List，保证和结果列表的顺序匹配
    List<String> batchList = new ArrayList<>(entry.getKey());
    List<String> resultList = entry.getValue();
    
    // 合并两个流，并行处理每一对元素
    Stream.zip(batchList.stream(), resultList.stream(), (batchId, resultId) -> new String[]{batchId, resultId})
          .parallel() // 启用并行执行，无需等待前一个调用完成
          .forEach(pair -> {
              InputStream inputStream = Connection.getQueryResultStream(jobId, pair[0], pair[1]);
              // 这里添加你对inputStream的后续处理逻辑
          });
});

方案2：兼容Java 8（用Apache Commons Lang）

如果你的项目还在Java 8，可以用Apache Commons Lang的Pair类来封装配对元素，通过索引遍历实现：

import org.apache.commons.lang3.tuple.Pair;

// 初始化代码和方案1一致...

map_batch_result_details.entrySet().forEach(entry -> {
    List<String> batchList = new ArrayList<>(entry.getKey());
    List<String> resultList = entry.getValue();
    
    // 通过索引生成配对，然后并行处理
    IntStream.range(0, Math.min(batchList.size(), resultList.size()))
             .mapToObj(index -> Pair.of(batchList.get(index), resultList.get(index)))
             .parallel()
             .forEach(pair -> {
                 InputStream inputStream = Connection.getQueryResultStream(jobId, pair.getLeft(), pair.getRight());
                 // 后续处理逻辑
             });
});

几个关键注意点： #

• 并行流的作用：parallel()方法会让流的操作在多个线程中执行，完全满足你“无需等待前一次执行完成”的需求。
• 顺序一致性：一定要用LinkedHashSet，否则HashSet的无序性会导致批次ID和结果ID的配对完全混乱。
• 边界处理：代码里用了Math.min()来避免两个集合大小不一致时出现索引越界，如果你确定两个集合的大小一定相等，可以去掉这个判断，甚至加个校验抛出异常，比如：

  if (batchList.size() != resultList.size()) {
      throw new IllegalArgumentException("批次ID和结果ID的数量不匹配");
  }
  

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