Julia 并行计算：@parallel 与 Threads

Julia 之所以在科学计算领域备受青睐，很大程度上得益于其原生支持的并行计算能力。与需要额外配置 MPI 库的其他语言不同，Julia 从语言层面内置了多种并行机制，其中 @parallel 和 Threads 是两套最常用的方案。

理解这两套机制的适用场景，能帮助你在不同任务中选择最优的并行策略。

并行计算的基本模型

在深入代码之前，需要先弄清楚两个核心概念：共享内存与分布式内存。

共享内存模式下，所有线程访问同一块物理内存。线程间通信无需数据拷贝，直接读写共享变量即可。这种模式适合单台多核机器，编程模型相对简单，但需要注意线程安全问题。

分布式内存模式下，每个进程拥有独立的内存空间。数据需要在进程间显式传递，通信开销更大，但可以跨多台机器扩展。这种模式适合计算集群，理论上可以无限扩展计算规模。

Julia 的 @parallel 对应分布式内存模型，基于多进程实现；而 Threads 对应共享内存模型，基于多线程实现。

@parallel：进程级并行

工作原理

@parallel 宏属于 Distributed 标准库，它会启动多个 Julia 进程（可能分布在不同机器上），将迭代任务分发到各个进程执行。每个进程拥有独立的内存空间，数据需要通过通道（Channel）或远程引用（RemoteRef）在进程间传递。

基础用法

@parallel 最典型的使用场景是 for 循环的归约操作：

using Distributed

# 添加工作进程（根据CPU核心数添加）
addprocs(4)

# 定义并行函数（注意：函数必须在所有进程上可用）
@everywhere function compute_chunk(start, stop)
    result = 0.0
    for i in start:stop
        result += sqrt(i) * sin(i)
    end
    return result
end

# 并行执行归约
n = 10_000_000
chunk_size = div(n, nworkers())
results = @distributed (+) for i in 1:n
    compute_chunk(i, min(i + chunk_size - 1, n))
end

println("结果: $results")
```

上述代码中，`@distributed (+)` 告诉 Julia 将循环迭代分配到各进程，每个进程计算局部结果后通过 `+` 运算符合并。

### 通道式数据传递

当任务需要更复杂的数据交换时，可以使用 `Channel` 和 `@spawn`：

```julia
using Distributed, Base.Iterators

@everywhere function process_data(data)
    sleep(0.1)  # 模拟计算
    sum(data)
end

# 创建通道
results = Channel{Float64}(4)

# 并行生产数据
@sync begin
    for chunk in partition(1:100, 25)
        @spawn results[fetch(myid())] = process_data(chunk)
    end
end

# 收集结果
total = 0.0
while isready(results)
    total += take!(results)
end
```

### @parallel 的适用场景

**适合使用 @parallel 的情况**：

| 场景 | 说明 |
|------|------|
| 大规模数据处理 | 数据量超过单机内存，需要分布到多台机器 |
| 长时间独立任务 | 各任务之间通信很少，天然适合并行 |
| 异构计算资源 | 不同节点计算能力不同，可按需分配任务 |
| 内存密集型任务 | 每个进程需要独立的大内存空间 |

**不适合使用 @parallel 的情况**：

| 场景 | 说明 |
|------|------|
| 频繁的小任务 | 进程通信开销会抵消并行收益 |
| 需要大量共享数据 | 进程间数据传递复杂且慢 |
| 实时性要求高 | 进程启动和通信延迟较高 |

---

## Threads：线程级并行

### 工作原理

`Threads` 模块从 Julia 1.3 开始引入，提供轻量级的共享内存并行。所有线程运行在同一进程内，共享相同的内存空间。这使得线程间通信几乎没有开销，但需要程序员自行处理数据竞争问题。

### 基础配置与用法

使用线程前，需要启动 Julia 时设置环境变量或命令行参数：

```bash
# 启动时指定4个线程
julia --threads=4

# 或通过环境变量
export JULIA_NUM_THREADS=4
julia
```

在代码中检查线程数并执行并行任务：

```julia
using Base.Threads

println("可用线程数: $(nthreads())")

# 简单的并行循环
n = 10_000_000
result = zeros(nthreads())

@threads for i in 1:n
    @inbounds result[threadid()] += sqrt(i) * sin(i)
end

total = sum(result)
println("结果: $total")
```

### 原子操作与线程安全

多线程环境下，对同一变量的并发读写需要同步机制。Julia 提供了多种原子操作：

```julia
using Base.Threads

# 方法1：使用 @atomic 和 @atomic!
result = 0
lock = Threads.Lock()

@threads for i in 1:1_000_000
    # 使用锁保护临界区
    Threads.lock(lock)
    result += sqrt(i)
    Threads.unlock(lock)
end

# 方法2：使用原子加法（更高效）
atomic_result = Atomic{Int64}(0)

@threads for i in 1:1_000_000
    @atomic atomic_result[] += i
end

println("原子加法结果: $(atomic_result[])")

线程局部存储

某些场景下，每个线程需要维护独立的数据副本：

using Base.Threads

# 初始化线程局部数组
thread_local accumulator::Vector{Float64} = [0.0]

@threads for i in 1:1_000_000
    # 每个线程操作自己的累加器
    accumulator[1] += sqrt(i)
end

# 汇总各线程结果
total = sum([acc[1] for acc in thread_local_storage_values()])

Threads 的适用场景

适合使用 Threads 的情况：

不适合使用 Threads 的情况：

核心对比与选择指南

性能特征对比

选择决策树

                    开始
                      │
        ┌─────────────┴─────────────┐
        │   数据是否超过单机内存？   │
        └─────────────┬─────────────┘
              是 ↙         ↘ 否
             ┌───┐         ┌─────────────────┐
             │@parallel│         │任务是否相互独立？│
             └───┘         └────────┬────────┘
                              是 ↙      ↘ 否
                          ┌─────┐    ┌──────┐
                          │@parallel│  │Threads│
                          └─────┘    └──────┘

典型场景推荐

场景一：Monte Carlo 模拟

如果需要运行大量独立模拟，使用 @parallel 更为合适：

using Distributed

@everywhere function monte_carlo_trials(n)
    count = 0
    for _ in 1:n
        x, y = rand(), rand()
        count += (x^2 + y^2 <= 1)
    end
    return count
end

total_trials = 100_000_000
trials_per_proc = div(total_trials, nworkers())

results = @distributed (+) for _ in 1:nworkers()
    monte_carlo_trials(trials_per_proc)
end

pi_estimate = 4 * results / total_trials

场景二：矩阵运算优化

对于矩阵元素的独立计算，使用 Threads 能获得更好的性能：

using Base.Threads

function parallel_matrix_op!(A, B, C)
    n, m = size(C)
    @threads for j in 1:m
        for i in 1:n
            C[i, j] = A[i, j] * B[i, j] + sqrt(A[i, j])
        end
    end
end

A = rand(10000, 10000)
B = rand(10000, 10000)
C = zeros(10000, 10000)

@time parallel_matrix_op!(A, B, C)

性能调优要点

@parallel 优化策略

1. 任务粒度：避免过细的任务划分，每个任务的计算量应远大于进程通信开销。

# 反模式：任务太细
@distributed (+) for i in 1:1_000_000
    i^2  # 每个任务仅计算一个平方
end

# 正模式：合理的任务粒度
@distributed (+) for chunk in partition(1:1_000_000, 10000)
    sum(x^2 for x in chunk)
end

2. 减少数据传递：只在进程间传递必要的结果数据。

# 反模式：传递大数组
@spawnat pid1 deepcopy(large_array)  # 糟糕

# 正模式：传递计算结果
@spawnat pid1 sum(large_array)  # 只传一个数

Threads 优化策略

1. 避免伪共享：让不同线程操作不同的内存区域。

# 伪共享示例（慢）
@threads for i in 1:n
    for j in 1:m
        C[i, j] = A[i, j] + B[i, j]
    end
end

# 按列划分（更好，避免同一缓存行被多线程竞争）
@threads for col in 1:nthreads()
    range = partition(1:m, nthreads())[col]
    for j in range
        for i in 1:n
            C[i, j] = A[i, j] + B[i, j]
        end
    end
end

2. 使用 @inbounds 和 @simd：加速内层循环。

@threads for j in 1:m
    for i in 1:n
        @inbounds @simd C[i, j] = A[i, j] * B[i, j]
    end
end

常见问题与解决方案

死锁

使用 @parallel 时注意避免循环引用通道：

# 可能导致死锁
@sync begin
    @async put!(ch1, take!(ch2))  # 等待对方先放入
    @async put!(ch2, take!(ch1))  # 死锁
end

解决方案：确保资源的获取顺序一致。

数据竞争

使用 Threads 时最常见的问题：

# 有数据风险的代码
@threads for i in 1:n
    global_counter += 1  # 多个线程同时写
end

# 解决方案：原子操作
@threads for i in 1:n
    @atomic global_counter += 1
end

线程数设置

并非线程越多越好。对于 CPU 密集型任务，线程数应等于物理核心数：

# 获取物理核心数
using Hwloc
num_physical_cores = Hwloc.num_physical_cores()

# 启动 Julia
# julia --threads=$num_physical_cores

混合使用策略

对于超大规模问题，可以同时使用两种并行模式：

using Distributed, Base.Threads

# 每台机器启动多个进程，每个进程内使用多线程
@everywhere function hybrid_compute(data)
    result = 0.0
    @threads for x in data
        result += sqrt(x) * log(x + 1)
    end
    return result
end

# 进程间并行归约
final_result = @distributed (+) for chunk in partition(data, 1000)
    hybrid_compute(chunk)
end

这种模式结合了 @parallel 的跨机扩展能力和 Threads 的低开销优势，适合在计算集群上处理超大规模问题。

总结

选择并行策略的核心原则：根据数据规模和任务特性决定。

• 数据量大、需要跨机器部署时，选择 @parallel
• 数据量适中、完全在单机上执行时，选择 Threads
• 超大规模问题考虑混合使用

实践中最有效的做法是先用 Threads 编写原型，因为其调试成本低、启动快。当单机性能达到瓶颈后，再根据需要迁移到 @parallel 或混合模式。