julia 多线程计算结果的收集（二）

一、预分配空间，为每个任务定义一个单独的cell，存放计算结果

1、定义一个数组，数组的总个数（长度）就是for循环的总次数（任务的个数）。
2、每次for循环，只修改数组对应的元素。比如，第一次循环，修改第一个元素，array[1]；第二次循环，修改第二个元素，array[2]，以此类推。

tasks = rand(tasks)        #计算的任务
tasks_num = length(tasks)  #任务的数量
ary = zeros(tasks_num)     #预分配一个数组容器，用于保存计算结果
items = [(idx = i,task = t) for (i,t) in zip(1:tasks_num,tasks) ]  #每次循环的【序号】和【任务】配对
Threads.@threads for item in items
    ary[item.idx] = item.task + 1
end

运行说明

image.png

二、为每个线程分配一个收集结果的容器

比如你的CPU是8核，你的线程数为16
那么，为每个线程创建一个收集计算结果的容器，这样每个线程只对一个容器进行push、append等操作，不会产生数据竞争。
容器可以是数组，字典等，它可以增删查改，不像以上第一种，只能update。

#数组作为容器
tasks = rand(1000007)        #计算的任务
res_ary = [[] for _ in 1:Threads.nthreads()]
@time Threads.@threads for task in tasks
    rtn = task * 2  #dosomething()
    push!(res_ary[Threads.threadid()],rtn)
end

res_ary

运行说明

image.png

#字典作为容器
tasks = rand(1000007)        #计算的任务
res_ary = [Dict() for _ in 1:Threads.nthreads()]
@time Threads.@threads for task in tasks
    rtn = task * 2  #dosomething()
    res_ary[Threads.threadid()][task] = rtn
end

#所有的字典合并到一个字典
all_dict = Dict()
for d in res_ary
    merge!(all_dict,d)
end
all_dict

运行说明

image.png

三、官方的用法，用lock

tasks = rand(1000007)        #计算的任务
res_ary = Dict()
splock = Threads.SpinLock()
@time Threads.@threads for task in tasks
    rtn = task * 2  #dosomething()    
    lock(splock)
    try
        res_ary[task] = rtn
    finally
        unlock(splock)
    end        
end
res_ary

用锁的话，会增加消耗。我也没咋用过，后期增加。