企业官网建设流程全解析

Julia 多线程编程完整指南：从线程启动、线程池到数据竞争防护

【免费下载链接】juliaThe Julia Programming Language项目地址: https://gitcode.com/gh_mirrors/ju/julia

本文以 Julia 官方手册的 Multi-Threading 章节为主体，结合仓库内base/threadingconstructs.jl、base/lock.jl、base/atomics.jl等源码，系统讲解 Julia 多线程的启动方式、线程池（Threadpool）机制、@threads/@spawn并行原语，以及使用锁（Lock）与原子操作（Atomic）编写无数据竞争并发代码的实战方案。读完本文，你将掌握从命令行配置多线程环境，到写出正确、可复用的并行求和、共享状态防护等完整能力。

使用多个线程启动 Julia

默认线程数与版本演进

Julia 默认启动 2 条执行线程：1 条 worker 线程和 1 条 interactive 线程（该默认行为自 Julia 1.12 起生效）。可以通过Threads.nthreads()验证：

julia> Threads.nthreads(:default) 1 julia> Threads.nthreads(:interactive) 1

兼容性说明（Julia 1.12）：Julia 1.12 之前默认只有 1 条（default 线程池）线程。若通过-t1或JULIA_NUM_THREADS=1将线程数显式设置为 1，则不会额外派生 interactive 线程。

通过命令行参数与环境变量控制线程数

执行线程的数量由两种方式控制，二者同时指定时**-t/--threads命令行参数优先于JULIA_NUM_THREADS环境变量**：

$ julia --threads 4

验证可用线程数：

julia> Threads.nthreads() 4

当前位于主线程（master thread），可用Threads.threadid()检查：

julia> Threads.threadid() 1

环境变量方式（必须在启动 Julia之前设置）：

# Bash（Linux/macOS） export JULIA_NUM_THREADS=4 # C shell（Linux/macOS）或 CMD（Windows） set JULIA_NUM_THREADS=4 # PowerShell（Windows） $env:JULIA_NUM_THREADS=4

版本兼容性：-t/--threads参数要求 Julia 1.5 及以上，更早版本只能用环境变量；JULIA_NUM_THREADS=auto要求 Julia 1.7 及以上，旧版本会忽略该值。

auto自动推断

线程数既可以指定为整数（--threads=4），也可以指定为auto（--threads=auto）。auto会让 Julia 尝试推断一个有用的默认线程数，详细规则见 命令行选项文档。

线程数向 worker 进程的传播

通过-t/--threads指定的线程数会传播给用-p/--procs或--machine-file启动的 worker 进程。例如julia -p2 -t2会启动 1 个主进程加 2 个 worker 进程，三个进程都启用 2 条线程。若需要对 worker 线程做更细粒度的控制，可使用addprocs并通过exeflags传入-t/--threads。

多 GC 线程

垃圾收集器（GC）同样可以使用多线程，其默认使用数量与计算 worker 线程数一致，也可用--gcthreads命令行参数或JULIA_NUM_GC_THREADS环境变量配置：

julia --gcthreads 4

--gcthreads参数要求 Julia 1.10 及以上。更完整的 GC 配置与性能调优细节参见 内存管理与垃圾收集。

从源码看，线程相关配置在 base/options.jl 中通过jl_options结构体承载，其中包括nthreadpools、nthreads、nmarkthreads、nsweepthreads、nthreads_per_pool等字段，分别对应线程池数量、总线程数、标记/清扫 GC 线程数以及每个线程池的线程数，可见线程池机制在内核选项中是一等公民。

线程池（Threadpools）

为什么需要线程池

当一个程序的线程忙于执行大量任务时，任务可能出现延迟，从而影响程序的响应性与交互性。为解决这一问题，可以在用Threads.@spawn派生任务时将其标记为 interactive 任务：

using Base.Threads @spawn :interactive f()

Interactive 任务应避免执行高延迟操作；若是长耗时任务，则应频繁让出（yield）CPU。

配置各线程池的线程数

默认情况下 Julia 预留 1 条 interactive 线程来运行交互任务，该数量可通过--threads 3,1形式的参数控制——逗号前是:default线程池线程数，逗号后是:interactive线程池线程数：

$ julia --threads 3,1 julia> Threads.nthreads(:interactive) 1 $ julia --threads 3,0 julia> Threads.nthreads(:interactive) 0

环境变量同样支持该语法：

export JULIA_NUM_THREADS=3,1

这会以 3 条:default线程池线程和 1 条:interactive线程池线程启动 Julia：

julia> using Base.Threads julia> nthreadpools() 2 julia> threadpool() # 主线程位于 interactive 线程池 :interactive julia> nthreads(:default) 3 julia> nthreads(:interactive) 1 julia> nthreads() 3

关于上述 API 有几点需要注意：

• 显式请求 1 条线程（-t1或JULIA_NUM_THREADS=1）不会额外增加 interactive 线程。
• 零参数版本的nthreads()返回 default 线程池的线程数。
• 主线程所属线程池不固定：取决于 Julia 是否以 interactive 线程启动，主线程可能在 default 或 interactive 线程池中。
• 两个数字中的任意一个都可以替换为auto，由 Julia 自行选择合理默认值。

源码层面，这些函数定义在 base/threadingconstructs.jl：threadpoolsize(pool)通过_nthreads_in_pool查询线程池大小，threadpool(tid)通过ccall(:jl_threadpoolid, ...)获取线程所在线程池（:default、:interactive或:foreign），nthreadpools()则直接读取jl_n_threadpools全局量。@spawn宏在 base/threadingconstructs.jl 中通过_spawn_set_thrpool调用jl_set_task_threadpoolid将任务绑定到指定线程池，当目标池不可用（线程数为 0）时回退到:default。

@threads宏

基本用法

Julia 使用Threads.@threads宏支持并行循环。创建零数组并让 4 条线程各自把线程 ID 写入对应位置：

julia> a = zeros(10) 10-element Vector{Float64}: 0.0 0.0 0.0 0.0 0.0 0.0 0.0 0.0 0.0 0.0 julia> Threads.@threads for i = 1:10 a[i] = Threads.threadid() end

迭代空间被分割到各线程，之后每条线程把自己的线程 ID 写入分配的位置：

julia> a 10-element Vector{Float64}: 1.0 1.0 1.0 2.0 2.0 2.0 3.0 3.0 4.0 4.0

注意Threads.@threads没有像@distributed那样的可选归约（reduction）参数。@threads还支持并行数组推导式，例如Threads.@threads [i for i in ...]，推导式会保持元素顺序（见 base/threadingconstructs.jl 中的宏实现与文档注释）。

调度选项：:dynamic、:static与:greedy

@threads宏支持可选调度参数Threads.@threads [schedule] for ... end（调度参数自 Julia 1.5 起可用）。宏在 base/threadingconstructs.jl 中仅接受:static、:dynamic、:greedy三者，否则抛出ArgumentError("unsupported schedule argument in @threads")：

• :dynamic（默认，Julia 1.8 起）：将迭代动态分配给可用 worker 线程。任务数被限制为可用线程数（Threads.threadpoolsize()）的某个小常数倍，每个任务处理连续区域。在length(xs)远大于线程数、且单次f(x)运行时间远小于任务派生/同步开销（通常小于 10 微秒）时，它通常比@sync for x in xs; @spawn f(x); end更高效。当前实现假设各迭代负载均匀。
• :greedy（Julia 1.11 起）：派生最多Threads.threadpoolsize()个任务，每个任务贪婪地从迭代器取下一个值；某任务一完成就取下一个。负载不均匀/波动大时通常是最佳选择，且只要求迭代器接口（不要求支持索引）。
• :static：为每个线程创建一个任务，将迭代均分并固定绑定到对应线程；threadid()在一次迭代内保证恒定。但在另一个@threads循环内或非 1 号线程上使用:static会报错。该模式仅为支持 Julia 1.3 之前的旧代码迁移而存在，新写的库函数不推荐使用，因为这类函数无法从任意 worker 线程调用。

语义约束

@threads以未指定的顺序、可能并发地执行循环体，且不保证迭代与任务、worker 线程的具体分配关系（每次执行可能不同）。循环体代码（包括其传递调用的代码）不能假设迭代如何分配给任务或线程；每次迭代必须能独立推进且无数据竞争。这意味着：

• 在一次迭代中获取的锁必须在该次迭代内释放；
• 用Channel等阻塞原语跨迭代通信是错误的；
• 除非使用锁或原子操作，只能写入迭代间不共享的位置；
• 除非使用:static调度，threadid()在一次迭代内也可能变化（参见下文任务迁移）。

另外，@threads派生的任务调度在:default线程池上，因此即使从主线程或 interactive 池中的任务调用，@threads也不会使用:interactive线程池的线程。

无数据竞争地使用@threads

数据竞争的概念详见下文“线程间通信与数据竞争”。先看一个串行求和函数：

julia> function sum_single(a) s = 0 for i in a s += i end s end sum_single (generic function with 1 method) julia> sum_single(1:1_000_000) 500000500000

直接加@threads会让多条线程同时读写共享变量s，暴露出数据竞争：

julia> function sum_multi_bad(a) s = 0 Threads.@threads for i in a s += i end s end sum_multi_bad (generic function with 1 method) julia> sum_multi_bad(1:1_000_000) 70140554652

结果不是正确的500000500000，而且每次求值大概率不同。

正确做法：使用任务专属的缓冲把求和分段为无竞争的分块。复用sum_single（其内部缓冲s是任务私有的），把输入向量a切成至多nthreads()块并行求和，最后再用sum_single汇总各块结果：

julia> function sum_multi_good(a) chunks = Iterators.partition(a, cld(length(a), Threads.nthreads())) tasks = map(chunks) do chunk Threads.@spawn sum_single(chunk) end chunk_sums = fetch.(tasks) return sum_single(chunk_sums) end sum_multi_good (generic function with 1 method) julia> sum_multi_good(1:1_000_000) 500000500000

重要警告：缓冲不要基于threadid()管理（如buffers = zeros(Threads.nthreads())）。因为并发任务可能让出（yield），多条并发任务可能在同一线程上共用同一缓冲，引入数据竞争风险；且当线程数多于 1 时，任务可能在让出点换线程（即下文的任务迁移）。

另一个选项是在跨任务/线程共享的变量上使用原子操作，根据操作特征不同可能性能更高（详见下文原子操作章节）。

线程间通信与数据竞争

虽然 Julia 线程可以通过共享内存通信，但编写正确且无数据竞争的并发代码是出了名的困难。Julia 的Channel是线程安全的，可用于安全通信；下面的小节还会介绍用锁和原子操作避免数据竞争。在某些情况下（特别是死锁或已知不安全操作，如让出当前运行任务），Julia 会抛出ConcurrencyViolationError。

数据竞争自由是程序员的责任

你完全有责任保证程序无数据竞争，不满足该要求时本文档的任何承诺都不成立，观测到的结果可能高度反直觉。引入数据竞争后，Julia不保证内存安全——如果另一个线程可能正在写入某数据，请极其小心地读取它，否则可能导致段错误甚至更糟。下面是一些多线程下访问全局变量的不安全方式：

Thread 1: global b = false global a = rand() global b = true Thread 2: while !b; end bad_read1(a) # 此处访问 a 是不安全的！ Thread 3: while !@isdefined(a); end bad_read2(a) # 此处访问 a 也是不安全的

使用锁避免数据竞争

锁是避免数据竞争、编写线程安全代码的重要工具。锁可被锁定与解锁；某线程锁定后未解锁即视为"持有"该锁。若只有一把锁，并要求持锁才能访问某些数据，即可保证多条线程永远不会同时访问同一数据。

注意：锁与变量之间的关联是由程序员建立的，而非程序本身。辅助类型Base.Lockable可以帮助你把锁和值关联起来，通常比自行维护关联更安全。

创建锁并在变更变量期间持有它，最简单的方式是@lock宏：

julia> my_lock = ReentrantLock(); julia> my_variable = [1, 2, 3]; julia> @lock my_lock my_variable[1] = 100 100

在其他线程上使用相同锁与变量执行同样模式，即可保证操作无数据竞争。@lock宏定义在 base/lock.jl，ReentrantLock是AbstractLock的可重入实现（内部通过ThreadSynchronizer实现阻塞与唤醒）。

同样操作也可以用函数式lock完成，以下两种方式与前文等价：

julia> lock(my_lock) do my_variable[1] = 100 end 100 julia> begin lock(my_lock) try my_variable[1] = 100 finally unlock(my_lock) end end 100

三种方式完全等价。注意最后一种必须显式使用try块确保锁一定被解锁，而前两种在内部自动处理。在修改被其他线程访问的数据（如给全局或闭包变量赋值）时，应始终使用上述锁模式，否则可能造成无法预见的严重后果。lock(f, l)的函数形式在 base/lock.jl 中实现，负责上锁、调用f、在finally中解锁的完整流程。

使用 Base.Lockable 将锁与值关联

Base.Lockable可以程序化地保证锁与值的关联，相比仅靠约定维护关联，更不易出错、更易读。任何对象都可以被包装进Base.Lockable（其定义在 base/lock.jl，默认使用ReentrantLock）：

julia> my_array = []; julia> my_locked_array = Base.Lockable(my_array);

持有锁时，可用空索引记法访问底层对象：

julia> begin lock(my_locked_array) try push!(my_locked_array[], 1) finally unlock(my_locked_array) end end 1-element Vector{Any}: 1

通常更简单、更安全的做法是把函数作为lock的第一个参数，函数作用于未锁定的对象，加锁/解锁自动处理：

julia> lock(x -> push!(x, 2), my_locked_array); julia> lock(display, my_locked_array) 2-element Vector{Any}: 1 2 julia> lock(my_locked_array) do x x[1] = π display(x) end 2-element Vector{Any}: π = 3.1415926535897... 2

原子操作

Julia 支持以原子方式访问和修改值，即以线程安全的方式避免竞态条件。原语类型的值可以包装为Threads.Atomic以标示必须以原子方式访问：

julia> i = Threads.Atomic{Int}(0); julia> ids = zeros(4); julia> old_is = zeros(4); julia> Threads.@threads for id in 1:4 old_is[id] = Threads.atomic_add!(i, id) ids[id] = id end julia> old_is 4-element Vector{Float64}: 0.0 1.0 7.0 3.0 julia> i[] 10 julia> ids 4-element Vector{Float64}: 1.0 2.0 3.0 4.0

如果去掉原子标记做加法，可能因竞态得到错误结果。下面的对比很直观：

julia> using Base.Threads julia> Threads.nthreads() 4 julia> acc = Ref(0) Base.RefValue{Int64}(0) julia> @threads for i in 1:1000 acc[] += 1 end julia> acc[] 926 julia> acc = Atomic{Int64}(0) Atomic{Int64}(0) julia> @threads for i in 1:1000 atomic_add!(acc, 1) end julia> acc[] 1000

非原子版本acc[] += 1在 1000 次增量后只得到 926，而原子版本严格得到 1000。Threads.Atomic{T}结构定义在 base/atomics.jl，atomic_add!等函数通过@atomic :acquire_release语义实现，atomic_cas!则基于@atomicreplace :acquire_release :acquire。

@atomic引用接口

虽然上面的Threads.atomic_add!函数族仍然受支持，但单个原子位置的推荐接口是@atomic、@atomicswap、@atomicreplace和@atomiconce宏的引用形式。它显式写出每个操作，让a[] += 1这类读-改-写更新明确成为原子操作（而不是静默地竞态），并允许把内存序作为可选首参（默认为:sequentially_consistent）：

julia> a = Threads.Atomic{Int}(0) Base.Threads.Atomic{Int64}(0) julia> @atomic a[] = 10 # 原子存储 10 julia> @atomic a[] # 原子加载 10 julia> @atomic :monotonic a[] # 显式内存序的原子加载 10 julia> @atomic a[] += 1 # 原子读-改-写，返回新值 11 julia> @atomicswap a[] = 0 # 原子交换，返回旧值 11 julia> @atomicreplace a[] 0 => 5 # 原子比较并交换 (old = 0, success = true)

这些宏同样可作用于AtomicMemory的元素以及@atomic结构体字段（见下文 per-field atomics），因此同一套语法覆盖标量、数组与字段。

Threads.Atomic类型是独立、类似Ref的原子单元。与Ref一样，它是实用的构建块且不会被移除；但当你有选择时，可变结构体的@atomic字段通常更优，因为避免了额外的间接层。

Threads.atomic_*函数早于这些宏出现且仍可用，但宏是操作原子单元的推荐方式——可读性更好且能指定内存序。下表给出对应转换（注意：atomic_*函数返回旧值，而@atomic a[] op= v返回新值，@atomic a[] op v返回old => new对，可用.first/.second取单个值）：

在Threads.Atomic上使用普通a[] = v存储已被弃用（因为a[] += 1这类写法看似原子实则不是），应改用@atomic a[] = v。另外，@atomic宏在Threads.Atomic上的引用形式要求 Julia 1.14 及以上版本。

这些宏在 base/exports.jl 中作为 Base 的公开 API 导出（@atomic、@atomicswap、@atomicreplace、@atomiconce），可直接使用。

字段级原子（Per-field atomics）

还可以用@atomic、@atomicswap、@atomicreplace、@atomiconce宏在更细粒度上使用原子操作。内存模型的具体细节与设计参见 Julia Atomics Manifesto（将正式发布）。

结构体声明中的任意字段都可以用@atomic修饰，之后任何写入都必须同样标记@atomic，并且必须使用定义好的原子序之一（:monotonic、:acquire、:release、:acquire_release或:sequentially_consistent）。对原子字段的读取也可以标注原子序约束；不指定时以 monotonic（宽松）序执行。字段级原子要求 Julia 1.7 及以上版本。

副作用与可变函数参数

使用多线程时必须小心对待非纯函数，否则可能得到错误结果。例如按命名惯例以!结尾的函数会修改其参数，因此不是纯函数（参见!命名约定）。将它们应用于多线程循环时，必须自行确保对共享可变状态的访问是受保护的。

@threadcall

外部库（例如通过ccall调用的库）对 Julia 基于任务的 I/O 机制构成挑战：如果 C 库执行阻塞操作，Julia 调度器在该调用返回前无法执行任何其他任务。（例外情况是：回调进 Julia 的自定义 C 代码——它可能让出；或者调用 C 等价物jl_yield()的 C 代码。）

@threadcall宏为此提供了一种避免执行停滞的方案：它把 C 函数调度到独立线程上执行，使用默认大小为 4 的线程池，线程池大小由环境变量UV_THREADPOOL_SIZE控制。在等待空闲线程期间以及获得线程后的函数执行期间，发起请求的任务（在主 Julia 事件循环上）会向其他任务让出。注意@threadcall在执行完成前不会返回，从用户角度看它与其他 Julia API 一样是阻塞调用。

极其重要：被调用的函数不能回调进 Julia，否则会段错误。

@threadcall可能在未来的 Julia 版本中被移除或改变，使用时需评估风险。

注意事项（Caveats）

目前，只要用户代码无数据竞争，Julia 运行时和标准库中的大多数操作都可以线程安全地使用。但在某些领域，线程支持的稳定性工作仍在进行中。多线程编程本身有很多固有难点，若使用线程的程序表现出异常或非预期行为（如崩溃或神秘结果），应首先怀疑线程交互。

使用 Julia 线程时需要了解以下具体限制与警告：

• Base 集合类型若被多条线程同时使用且至少一条线程在修改集合（常见如对数组push!、向Dict插入），需要手动加锁。
• @spawn使用的调度是非确定性的，不应依赖它。
• 计算密集、不分配内存的任务会阻止其他正在分配内存的线程执行垃圾回收。此时可能需要手动调用GC.safepoint()让 GC 得以运行（该限制未来会移除）。
• 避免并行执行顶层操作，例如include，或对类型、方法、模块定义的eval。
• 注意库注册的**终结器（finalizer）**在启用线程后可能失效。这可能需要在生态系统中做过渡性工作，之后线程才能被广泛放心采用。详见下文"Finalizer 的安全使用"。

任务迁移（Task Migration）

任务在某线程上开始运行后，如果该任务让出，它可能迁移到另一条线程。这类任务可能由@spawn或@threads启动，不过@threads的:static调度选项会冻结threadid()。

这意味着在大多数情况下，不应把threadid()视为任务内的常量，因此也不应用它来索引缓冲向量或有状态对象。任务迁移自 Julia 1.7 引入；在此之前，任务始终停留在其启动线程上。

Finalizer 的安全使用

由于终结器可以打断任何代码，它们与全局状态的交互必须极其小心。但不幸的是，终结器的主要用途恰恰是更新全局状态（纯函数作为终结器通常没有意义），这带来了一个两难问题。处理该问题有几种策略：

1. 单线程时，代码可调用内部 C 函数jl_gc_enable_finalizers防止终结器在临界区内被调度。内部上，某些函数（如 C 锁）使用它来防止在做特定操作（增量包加载、codegen 等）时发生递归。将锁与该标志组合使用可以使终结器安全。

2. 第二种策略（Base 在少数地方采用）是显式延迟终结器，直到它能够非递归地获取其锁。以下示例展示如何将该策略应用于Distributed.finalize_ref：

   function finalize_ref(r::AbstractRemoteRef) if r.where > 0 # 检查终结器是否已运行 if islocked(client_refs) || !trylock(client_refs) # 若无法自由获取锁，将终结器延迟到稍后执行 finalizer(finalize_ref, r) return nothing end try # `lock` 后应始终跟随 `try` if r.where > 0 # 必须在此再次检查 # 在此执行真正的清理 r.where = 0 end finally unlock(client_refs) end end nothing end

3. 第三种相关策略是使用无让出（yield-free）队列。目前 Base 中未实现无锁队列，但Base.IntrusiveLinkedListSynchronized{T}是合适的候选。这通常非常适合事件循环代码，例如Gtk.jl用它管理生命周期引用计数。在这种方式下，不在终结器内部做显式工作，而是把对象加入队列，在更安全的时机执行。事实上 Julia 的任务调度器已经在使用这一策略，因此把终结器定义为x -> @spawn do_cleanup(x)就是该方式的一个示例。注意这样无法控制do_cleanup在哪个线程运行，所以do_cleanup仍需获取锁；如果你实现自己的队列，则可以显式地只从自己的线程中排空该队列，从而避免这个要求。

结语

Julia 的多线程模型围绕"任务 + 线程池 + 显式同步"展开：启动时用-t/--threads或JULIA_NUM_THREADS精确控制:default与:interactive线程池的规模，运行时用@threads（:dynamic/:static/:greedy调度）与@spawn表达并行，用锁（@lock、Base.Lockable）与原子操作（Threads.Atomic、@atomic宏族、per-field atomics）保证无数据竞争。同时要牢记数据竞争自由是程序员的责任、threadid()不可视为常量（任务迁移）、以及终结器与集合类型等特殊场景的限制。本手册章节对应的完整源码实现可进一步查阅 base/threadingconstructs.jl、base/lock.jl 与 base/atomics.jl，测试用例可参考 test/threads.jl 与 test/threads_exec.jl，以验证上述全部行为。

【免费下载链接】juliaThe Julia Programming Language项目地址: https://gitcode.com/gh_mirrors/ju/julia