如何通过多线程加速二级索引映射的填充？

背景 #

foo()方法通过多线程并行处理，核心目标之一是填充mainMap，用于索引计算结果。mainMap的类型为Map<Pair<T, T>, Object>，我们仅关注它的键（Pair对象）。

Pair类结构简单，仅封装两个同类型对象，代码如下：

public class Pair<T, T> {
    private T first; // 键对第一个元素
    private T second; // 键对第二个元素

    public Pair(T first, T second) {
        this.first = first;
        this.second = second;
    }

    public boolean contains(T elem) {
        return this.first.equals(elem) || this.second.equals(elem);
    }

    // Getters and Setters...
}

核心问题 #

处理过程中需要快速从mainMap中获取所有键（Pair）包含指定T类型元素elem的对应值。

已尝试的方案 #

注：示例用String作为泛型T的实现，实际T可以是任意类型。

1. 基础遍历方案 #

遍历mainMap.keySet()，用key.contains(elem)过滤符合条件的Pair，再查询mainMap。
结果：数据量可达数十万条，遍历过滤的性能完全无法满足需求。

2. 二级索引方案 #

创建两个Map<T, Set<Pair<T, T>>>类型的索引映射：firstToPairMap和secondToPairMap，工作逻辑如下：
假设mainMap中有以下条目：

{A, B} -> ...
  {C, B} -> ...
  {X, Z} -> ...

则firstToPairMap的内容为：

A -> [{A, B}]
  C -> [{C, B}]
  X -> [{X, Z}]

secondToPairMap的内容为：

B -> [{A, B}, {C, B}]
  Z -> [{X, Z}]

查询时直接通过这两个索引获取包含目标元素的Pair，查询性能很好，但填充索引的性能极差。因为mainMap是多线程并行填充的，索引映射也需要并行更新，但每次插入都要合并已有集合（比如secondToPairMap中B对应的集合需要合并新Pair），合并操作的开销极大。

为优化，我尝试将并行计算中要插入的Pair存入队列，延迟填充索引，实现了dequeuePairsMap方法，但效果不理想：

private static void dequeuePairsMap(Queue<Pair<T, T>> queue, Map<T, Set<Pair<T, T>>> secondToPairMap, Map<T, Set<Pair<T, T>>> firstToPairMap) {
   while(!queue.isEmpty()) {
        Pair<T, T> pair = queue.poll();
        Set<Pair<T, T>> secondToPairSet = secondToPairMap.getOrDefault(pair.getSecond(), new HashSet<>());
        Set<Pair<T, T>> firstToPairSet = firstToPairMap.getOrDefault(pair.getFirst(), new HashSet<>());
        secondToPairSet.add(pair);
        firstToPairSet.add(pair);
        secondToPairMap.put(pair.getSecond(), secondToPairSet);
        firstToPairMap.put(pair.getFirst(), firstToPairSet);
    }
}

这个实现仍然依赖集合合并操作，没解决性能问题。已经确认瓶颈不是并发访问，而是合并操作本身。

求解决这个性能问题的思路或方案。

编辑：分享一个能提升速度的dequeuePairsMap新实现，思路是批量插入映射、减少合并请求：

private static void dequeuePairsMap(Queue<Pair<T, T>> queue, Map<T, Set<Pair<T, T>>> secondToPairMap, Map<T, Set<Pair<T, T>>> firstToPairMap) {

    while(!queue.isEmpty()) {
        Pair<T, T> pair = queue.poll();
        Set<Pair<T, T>> firstToPairSet = queue.stream().filter(edge -> edge.containsAsFirst(pair.getFirst())).collect(Collectors.toSet());
        Set<Pair<T, T>> secondToPairSet = queue.stream().filter(edge -> edge.containsAsSecond(pair.getSecond())).collect(Collectors.toSet());
        firstToPairSet.add(pair);
        secondToPairSet.add(pair);
        Set<Pair<T, T>> firstToPairExistingSet = firstToPairMap.getOrDefault(pair.getFirst(), new HashSet<>());
        Set<Pair<T, T>> secondToPairExistingSet = secondToPairMap.getOrDefault(pair.getSecond(), new HashSet<>());
        if(!firstToPairExistingSet.containsAll(firstToPairSet))
            firstToPairMap.merge(pair.getFirst(), firstToPairSet, (oldSet, newSet) -> Stream.concat(oldSet.stream(), newSet.stream()).collect(Collectors.toSet()));

        if(!secondToPairExistingSet.containsAll(secondToPairSet))
            secondToPairMap.merge(pair.getSecond(), secondToPairSet, (oldSet, newSet) -> Stream.concat(oldSet.stream(), newSet.stream()).collect(Collectors.toSet()));

    }

}

优化建议 #

1. 替换集合实现，避免重复替换Set
   你的核心问题在于每次更新索引时，都要先取出旧Set、添加元素、再把整个Set放回Map，这相当于每次都要替换整个集合，开销极大。改用ConcurrentHashMap搭配线程安全的Set实现，直接在原有集合上添加元素：

   // 初始化索引
   Map<T, Set<Pair<T,T>>> firstToPairMap = new ConcurrentHashMap<>();
   Map<T, Set<Pair<T,T>>> secondToPairMap = new ConcurrentHashMap<>();
   
   // 处理单个Pair时
   Pair<T,T> pair = ...;
   // 懒加载创建线程安全的Set，直接添加元素
   firstToPairMap.computeIfAbsent(pair.getFirst(), k -> Collections.newSetFromMap(new ConcurrentHashMap<>())).add(pair);
   secondToPairMap.computeIfAbsent(pair.getSecond(), k -> Collections.newSetFromMap(new ConcurrentHashMap<>())).add(pair);
   

   这样完全避免了集合的合并和替换操作，直接在原有集合上追加元素，性能会大幅提升。

2. 批量处理队列元素
   不要逐个poll队列元素，改为批量取出（比如每次取1000条），先按first和second分组后再批量更新索引，减少对Map的操作次数：

   List<Pair<T,T>> batch = new ArrayList<>();
   // 每次从队列取最多1000条
   queue.drainTo(batch, 1000);
   if (batch.isEmpty()) break;
   
   // 按first分组
   Map<T, List<Pair<T,T>>> groupByFirst = batch.stream().collect(Collectors.groupingBy(Pair::getFirst));
   // 按second分组
   Map<T, List<Pair<T,T>>> groupBySecond = batch.stream().collect(Collectors.groupingBy(Pair::getSecond));
   
   // 批量更新索引
   groupByFirst.forEach((key, pairs) -> {
       Set<Pair<T,T>> set = firstToPairMap.computeIfAbsent(key, k -> Collections.newSetFromMap(new ConcurrentHashMap<>()));
       set.addAll(pairs);
   });
   groupBySecond.forEach((key, pairs) -> {
       Set<Pair<T,T>> set = secondToPairMap.computeIfAbsent(key, k -> Collections.newSetFromMap(new ConcurrentHashMap<>()));
       set.addAll(pairs);
   });
   

3. 给Pair补充equals和hashCode
   你提供的Pair类没有实现这两个方法，HashSet会基于对象引用判断元素是否存在，导致相同逻辑的Pair被重复添加，既浪费空间又增加无效操作。补充实现：

   @Override
   public boolean equals(Object o) {
       if (this == o) return true;
       if (o == null || getClass() != o.getClass()) return false;
       Pair<?, ?> pair = (Pair<?, ?>) o;
       return Objects.equals(first, pair.first) && Objects.equals(second, pair.second);
   }
   
   @Override
   public int hashCode() {
       return Objects.hash(first, second);
   }
   

4. 改用Guava的Multimap
   Guava的ConcurrentHashMultimap天生就是键对应多值的结构，内部已经优化了集合存储和合并逻辑，比自己手动实现Map+Set高效得多：

   Multimap<T, Pair<T,T>> firstToPairMap = ConcurrentHashMultimap.create();
   Multimap<T, Pair<T,T>> secondToPairMap = ConcurrentHashMultimap.create();
   
   // 添加元素只需一行代码
   firstToPairMap.put(pair.getFirst(), pair);
   secondToPairMap.put(pair.getSecond(), pair);
   

   完全不需要自己处理集合的创建和合并，代码简洁且性能有保障。

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