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.12Julia 1.12 之前默认只有 1 条default 线程池线程。若通过-t1或JULIA_NUM_THREADS1将线程数显式设置为 1则不会额外派生 interactive 线程。通过命令行参数与环境变量控制线程数执行线程的数量由两种方式控制二者同时指定时**-t/--threads命令行参数优先于JULIA_NUM_THREADS环境变量**$ julia --threads 4验证可用线程数julia Threads.nthreads() 4当前位于主线程master thread可用Threads.threadid()检查julia Threads.threadid() 1环境变量方式必须在启动 Julia之前设置# BashLinux/macOS export JULIA_NUM_THREADS4 # C shellLinux/macOS或 CMDWindows set JULIA_NUM_THREADS4 # PowerShellWindows $env:JULIA_NUM_THREADS4版本兼容性-t/--threads参数要求 Julia 1.5 及以上更早版本只能用环境变量JULIA_NUM_THREADSauto要求 Julia 1.7 及以上旧版本会忽略该值。auto自动推断线程数既可以指定为整数--threads4也可以指定为auto--threadsauto。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 任务应避免执行高延迟操作若是长耗时任务则应频繁让出yieldCPU。配置各线程池的线程数默认情况下 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_THREADS3,1这会以 3 条:default线程池线程和 1 条:interactive线程池线程启动 Juliajulia 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_THREADS1不会额外增加 interactive 线程。零参数版本的nthreads()返回 default 线程池的线程数。主线程所属线程池不固定取决于 Julia 是否以 interactive 线程启动主线程可能在 default 或 interactive 线程池中。两个数字中的任意一个都可以替换为auto由 Julia 自行选择合理默认值。源码层面这些函数定义在 base/threadingconstructs.jlthreadpoolsize(pool)通过_nthreads_in_pool查询线程池大小threadpool(tid)通过ccall(:jl_threadpoolid, ...)获取线程所在线程池:default、:interactive或:foreignnthreadpools()则直接读取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与:greedythreads宏支持可选调度参数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更高效。当前实现假设各迭代负载均匀。:greedyJulia 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.jlReentrantLock是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默认使用ReentrantLockjulia 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.jlatomic_add!等函数通过atomic :acquire_release语义实现atomic_cas!则基于atomicreplace :acquire_release :acquire。atomic引用接口虽然上面的Threads.atomic_add!函数族仍然受支持但单个原子位置的推荐接口是atomic、atomicswap、atomicreplace和atomiconce宏的引用形式。它显式写出每个操作让a[] 1这类读-改-写更新明确成为原子操作而不是静默地竞态并允许把内存序作为可选首参默认为:sequentially_consistentjulia 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取单个值旧函数调用atomic等价写法atomic_add!(a, v)atomic a[] vatomic_sub!(a, v)atomic a[] - vatomic_and!(a, v)atomic a[] vatomic_or!(a, v)atomic a[] \| vatomic_xor!(a, v)atomic a[] ⊻ vatomic_max!(a, v)atomic a[] max vatomic_min!(a, v)atomic a[] min vatomic_xchg!(a, v)atomicswap a[] vatomic_cas!(a, cmp, new)atomicreplace a[] cmp new在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 的安全使用由于终结器可以打断任何代码它们与全局状态的交互必须极其小心。但不幸的是终结器的主要用途恰恰是更新全局状态纯函数作为终结器通常没有意义这带来了一个两难问题。处理该问题有几种策略单线程时代码可调用内部 C 函数jl_gc_enable_finalizers防止终结器在临界区内被调度。内部上某些函数如 C 锁使用它来防止在做特定操作增量包加载、codegen 等时发生递归。将锁与该标志组合使用可以使终结器安全。第二种策略Base 在少数地方采用是显式延迟终结器直到它能够非递归地获取其锁。以下示例展示如何将该策略应用于Distributed.finalize_reffunction 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第三种相关策略是使用无让出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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考