Julia多线程共享向量数据竞争规避及并行优化问题咨询

首先，你的原并行代码虽然加了锁，但存在两个核心问题：一是全局锁会导致线程大量等待，完全抵消并行带来的性能提升；二是用动态数组存储哈希，h ∉ hashes是线性查找，临界区执行效率极低，甚至可能因为锁持有时间过长引发隐性问题。另外，如果compute(g)是纯函数，理论上锁保护的逻辑应该能得到正确结果，但结果不一致大概率是因为线性查找在高并发下的隐性问题，或者你忽略了compute(g)的非纯特性（比如有副作用、返回值不稳定）。

最优并行方案：本地去重+全局合并 #

正确的思路是让每个线程独立处理自己的任务块，先完成compute(g)和本地去重，最后再全局合并去重。这种方式完全避免了全局锁，最大化并行效率，同时用Set存储哈希把查找复杂度降到O(1)，整体性能远优于锁方案。

function parallel_output(G)
    # 为每个线程初始化本地哈希集合和结果数组
    thread_data = [
        (Set{UInt64}(), Vector{typeof(compute(first(G)))}()) 
        for _ in 1:Threads.nthreads()
    ]

    # 静态分配任务块，每个线程处理固定范围的G元素
    @threads :static for g in G
        tid = Threads.threadid()
        local_hashes, local_objs = thread_data[tid]

        ng = compute(g)
        h = hash(ng)

        # 本地去重，无需锁
        if !in(h, local_hashes)
            push!(local_hashes, h)
            push!(local_objs, ng)
        end
    end

    # 全局合并所有线程的结果，再次去重
    global_hashes = Set{UInt64}()
    global_objs = Vector{typeof(compute(first(G)))}()

    for (lh, lo) in thread_data
        for (h, obj) in zip(lh, lo)
            if !in(h, global_hashes)
                push!(global_hashes, h)
                push!(global_objs, obj)
            end
        end
    end

    return global_objs
end

关键优化点说明 #

1. 本地独立处理：每个线程维护自己的哈希集合和结果数组，完全不需要锁，compute(g)的并行效率拉满。
2. 用Set存储哈希：不管是本地还是全局，Set的查找是O(1)，比原代码的线性查找快几个数量级，这对单线程版本也是重要优化：

   # 优化后的单线程版本
   function optimized_output(G)
       hashes = Set{UInt64}()
       objects = Vector{typeof(compute(first(G)))}()
       for g ∈ G
           ng = compute(g)
           h = hash(ng)
           if h ∉ hashes
               push!(hashes, h)
               push!(objects, ng)
           end
       end
       return objects
   end
   

3. 静态任务分配：@threads :static让每个线程处理固定的任务块，减少线程调度开销，适合计算密集型任务。

关于原锁方案的补充 #

如果一定要用锁方案，必须把哈希存储换成Set，减少临界区的执行时间，降低锁竞争：

function locked_parallel_output(G)
    hashes = Set{UInt64}()
    objects = Vector{typeof(compute(first(G)))}()
    lk = ReentrantLock()

    @threads for g ∈ G
        ng = compute(g)
        h = hash(ng)

        lock(lk) do
            if !in(h, hashes)
                push!(hashes, h)
                push!(objects, ng)
            end
        end
    end
    return objects
end

但这种方案的并行效率远不如本地去重+全局合并，因为每次进入锁都要等待，只有当compute(g)的开销极大（远大于锁等待时间）时才勉强可用。

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