Java递归图遍历中的并发问题——顶点重复处理

看起来你遇到了多线程图遍历里典型的竞态条件问题——明明用了ConcurrentHashMap的keySet来做已访问集合，还加了synchronized方法，可v8还是被重复处理了。我来帮你拆解下问题根源，再给你可行的修复方案。

问题出在哪？ #

你的核心问题在于检查节点是否可处理和标记节点为已访问这两个操作不是原子的。举个例子：

1. 线程A（处理v5）和线程B（处理v6）同时检查v8：此时v8还没被加入visited，且它们的父节点（v5、v6）都已被访问，所以两个线程都通过了check方法。
2. 接着两个线程都会启动新线程去处理v8，之后才执行visited.get().add(vertex)——这时候已经晚了，重复处理已经发生。

另外，你用InheritableThreadLocal来存visited集合其实没必要，反而容易混淆：所有线程应该共享同一个已访问集合，而不是每个线程继承一份（虽然这里继承的是同一个ConcurrentHashMap实例，但ThreadLocal在这里没有发挥积极作用）。

怎么修复？ #

我们需要把“检查父节点是否全部访问 + 标记当前节点为已访问”做成一个原子操作，确保只有一个线程能成功触发节点处理。

具体步骤：

1. 改用共享的并发集合：把ThreadLocal去掉，直接用一个类级别的ConcurrentHashMap.newKeySet()作为所有线程共享的已访问集合。
2. 原子化检查与标记：把原来的check方法改成同时完成“父节点检查”和“原子标记已访问”的逻辑，利用ConcurrentHashMap的原子性add方法来保证只有一个线程能成功标记节点。

修改后的代码示例：

import org.jgrapht.Graph;
import org.jgrapht.graph.DefaultDirectedGraph;
import org.jgrapht.graph.DefaultEdge;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

public class VectorVisitedLogicTest2 {
    // 所有线程共享的已访问集合
    private final Set<String> visited = ConcurrentHashMap.newKeySet();

    public static void main(String[] args) {
        VectorVisitedLogicTest2 test = new VectorVisitedLogicTest2();
        test.run();
    }

    private void run() {
        System.out.println("Starting process");
        Graph<String, DefaultEdge> directedGraph
                = new DefaultDirectedGraph<>(DefaultEdge.class);
        // 初始化顶点和边
        directedGraph.addVertex("v1");
        directedGraph.addVertex("v2");
        directedGraph.addVertex("v3");
        directedGraph.addVertex("v4");
        directedGraph.addVertex("v5");
        directedGraph.addVertex("v6");
        directedGraph.addVertex("v7");
        directedGraph.addVertex("v8");
        directedGraph.addVertex("v9");
        directedGraph.addVertex("v10");
        directedGraph.addVertex("v11");
        directedGraph.addVertex("v12");
        directedGraph.addVertex("v13");
        directedGraph.addEdge("v1", "v2");
        directedGraph.addEdge("v1", "v3");
        directedGraph.addEdge("v1", "v13");
        directedGraph.addEdge("v2", "v4");
        directedGraph.addEdge("v4", "v5");
        directedGraph.addEdge("v4", "v6");
        directedGraph.addEdge("v3", "v7");
        directedGraph.addEdge("v7", "v8");
        directedGraph.addEdge("v5", "v8");
        directedGraph.addEdge("v6", "v8");
        directedGraph.addEdge("v8", "v9");
        directedGraph.addEdge("v8", "v10");
        directedGraph.addEdge("v8", "v12");
        directedGraph.addEdge("v9", "v11");
        directedGraph.addEdge("v10", "v11");
        directedGraph.addEdge("v11", "v12");
        directedGraph.addEdge("v12", "v13");

        // 先标记初始节点v1为已访问
        visited.add("v1");
        Thread thread = new Thread(() -> {
            try {
                execute(directedGraph, "v1");
            } catch (InterruptedException e) {
                System.out.println(e);
                throw new RuntimeException(e);
            }
        });
        thread.start();
    }

    public void execute(Graph<String, DefaultEdge> graph, String vertex) throws InterruptedException {
        System.out.println(vertex + "-" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date()));

        for (DefaultEdge edge : graph.outgoingEdgesOf(vertex)) {
            String neighbor = graph.getEdgeTarget(edge);
            Set<String> neighbourParents = new HashSet<>();
            for (DefaultEdge parentEdge : graph.incomingEdgesOf(neighbor)) {
                neighbourParents.add(graph.getEdgeSource(parentEdge));
            }

            // 原子化检查父节点并标记已访问
            if (checkAndMarkVisited(neighbor, neighbourParents)) {
                Thread thread = new Thread(() -> {
                    try {
                        System.out.println("Calling : " + neighbor + " from : " + vertex );
                        execute(graph, neighbor);
                    } catch (Exception e) {
                        System.out.println(e);
                        throw new RuntimeException(e);
                    }
                });
                thread.start();
            }
        }
    }

    private boolean checkAndMarkVisited(String neighbor, Set<String> neighbourParents) {
        // 先检查所有父节点是否都已被访问
        if (!visited.containsAll(neighbourParents)) {
            return false;
        }
        // 原子性添加到已访问集合：只有当neighbor不存在时才添加成功，返回true
        return visited.add(neighbor);
    }
}

为什么这样能解决问题？ #

• visited.add(neighbor)是原子操作：当多个线程同时调用这个方法时，只有第一个线程能成功添加并返回true，其他线程会返回false，这样就保证了只有一个线程会触发该节点的处理。
• 移除了ThreadLocal，所有线程共享同一个已访问集合，避免了不必要的线程隔离带来的混淆。
• 把父节点检查和标记操作放在同一个方法里，确保了逻辑的连贯性和原子性。

这样修改后，v8就只会被处理一次，不会再出现重复调用的情况了。

备注：内容来源于stack exchange，提问作者Rahul R