Julia Task 与并发

Task（也称协程 coroutine）是 Julia 的轻量级执行单元，可以在计算之间灵活地暂停和恢复，是 Julia 处理并发编程的基础。

Julia 支持以下几类并发与并行编程：

类别	说明
异步任务（协程）	Task 暂停/恢复，适合 IO 密集场景
多线程	多个 Task 同时跑在多个 CPU 线程上，共享内存
分布式计算	多个 Julia 进程协作，各进程有独立内存
GPU 计算	通过 CUDA.jl 等包在 GPU 上运行 Julia 代码

本章介绍最常用的协程与多线程，分布式计算只做简要说明。

Task 基础

用 @task 宏可以把一段代码包装成 Task，此时任务还没有运行；调用 schedule() 让它开始执行：

实例

# 创建一个任务（还没有运行）
julia> t = @task begin
sleep(1)
println("done")
end
Task (runnable) @0x00007f13a40c0eb0

# 调度执行，schedule 会立即返回
julia> schedule(t);

# wait 阻塞等待任务完成
julia> wait(t);
done

实际开发中更常用的是 Threads.@spawn 宏，它创建任务并立即调度执行：

实例

julia> t = Threads.@spawn begin
sleep(1)
println("done")
end
Task (runnable) @0x00007f13a40c11c0

fetch(t) 会等待任务完成并返回其结果：

实例

julia> t = Threads.@spawn 1 + 1
Task (runnable) @0x00007f13a40c13c0

julia> fetch(t)
2

@sync 等待一组任务

@sync 包住的代码块会阻塞到其中所有任务都执行完：

实例

julia> @sync begin
Threads.@spawn begin
sleep(1)
println("任务 1 完成")
end
Threads.@spawn begin
sleep(0.5)
println("任务 2 完成")
end
end
任务 2 完成
任务 1 完成

Channel：任务间通信

Channel 是一个先进先出（FIFO）的队列，可以让多个任务安全地传递数据：

实例

# 创建一个容量为 32、存放 Int 的通道
julia> chnl = Channel{Int}(32);

# put! 放入数据
julia> put!(chnl, 1);

julia> put!(chnl, 2);

# take! 取出数据
julia> take!(chnl)
1

julia> take!(chnl)
2

Channel 的关键行为：

通道满了，put! 会阻塞等待空位；通道空了，take! 会阻塞等待数据。
put! 和 take! 是线程安全的，多个任务可以同时读写同一个通道。
close(ch) 关闭通道，关闭后不能再 put!，但可以先取完已有的数据。

下面是一个经典的生产者-消费者例子：4 个工作任务从 jobs 通道取任务，把结果写进 results 通道：

生产者-消费者 Channel 模型示意图

实例

using Base.Threads

# jobs 通道：生产任务；results 通道：收集结果
const jobs = Channel{Int}(32);
const results = Channel{Tuple}(32);

function do_work()
# for 循环遍历通道会一直取值直到通道关闭
for job_id in jobs
exec_time = rand()
sleep(exec_time) # 模拟实际工作耗时
put!(results, (job_id, exec_time)) # 写回结果
end
end

function make_jobs(n)
for i in 1:n
put!(jobs, i)
end
close(jobs) # 所有任务发完后关闭 jobs 通道
end

n = 4
errormonitor(Threads.@spawn make_jobs(n)); # 生产任务

for i in 1:4 # 启动 4 个工作任务
errormonitor(Threads.@spawn do_work())
end

@elapsed while n > 0 # 收集结果
job_id, exec_time = take!(results)
println("$job_id 完成于 $(round(exec_time; digits=2)) 秒")
global n -= 1
end
1 完成于 0.04 秒
4 完成于 0.5 秒
2 完成于 0.56 秒
3 完成于 0.73 秒

提示：errormonitor(t) 会在任务抛出异常时打印错误信息，比静默失败好得多，建议所有 @spawn 都包上它。

多线程

启动多线程

Julia 1.10 之前默认单线程，可以用 -t 参数或环境变量指定线程数：

$ julia --threads 4
# 或者
$ export JULIA_NUM_THREADS=4
$ julia

用 Threads.nthreads() 查看可用线程数：

实例

julia> Threads.nthreads()
4

julia> Threads.threadid() # 当前代码运行在哪个线程
1

@threads 并行循环

Threads.@threads 加在 for 循环前面，会把迭代范围分配给多个线程并行执行：

实例

julia> a = zeros(10);

julia> Threads.@threads for i = 1:10
a[i] = Threads.threadid()
end

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

数据竞争警告：多线程并行写同一个变量会产生错误结果，这是并发编程最常见的坑：

错误示例：数据竞争

julia> function sum_multi_bad(a)
s = 0
Threads.@threads for i in a
s += i # 多个线程同时读写 s，结果不可预测！
end
s
end
sum_multi_bad (generic function with 1 method)

julia> sum_multi_bad(1:1_000_000) # 正确答案应该是 500000500000
62501142778

正确的做法是先分块各自求和（互相独立，没有竞争），再合并：

正确示例

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

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

用锁保护共享数据

多线程修改共享数据时，用 ReentrantLock 保证同一时刻只有一个线程访问：

实例

julia> my_lock = ReentrantLock();

julia> my_variable = [1, 2, 3];

# @lock 宏：加锁执行表达式，结束后自动解锁
julia> @lock my_lock my_variable[1] = 100
100

# 函数式写法等价
julia> lock(my_lock) do
my_variable[1] = 100
end
100

原子操作

对简单数值的累加，原子操作比锁更高效：

实例

julia> acc = Threads.Atomic{Int64}(0)
Base.Threads.Atomic{Int64}(0)

julia> @threads for i in 1:1000
Threads.atomic_add!(acc, 1)
end

julia> acc[]
1000

异步编程的典型场景

协程最大的优势是 IO 密集场景：等待网络、磁盘时把 CPU 让给别的任务，看似串行的代码实际在并发执行：

实例

using Sockets

# 并发连接三个网站，总耗时约等于最慢的那个，而不是三个之和
julia> @sync for hostname in ("google.com", "github.com", "julialang.org")
Threads.@spawn begin
conn = connect(hostname, 80)
write(conn, "GET / HTTP/1.1\r\nHost:$(hostname)\r\n\r\n")
readline(conn, keep=true)
println("完成连接 $(hostname)")
end
end
完成连接 google.com
完成连接 julialang.org
完成连接 github.com

经验法则：IO 密集（网络、磁盘、sleep）用 @spawn 协程就够；CPU 密集需要真并行时，用 @threads 或 @spawn 并确保 Julia 以多线程启动。

返回顶部

菜鸟教程