如何从 Python 中的线程获取返回值？

Question

我在 python 中做了一些计算量很大的任务，并找到了用于并行化的线程模块。我有一个函数可以进行计算并返回一个 ndarray 作为结果。现在我想知道如何并行化我的函数并从每个线程取回计算出的数组。

下面的例子通过轻量级的函数和计算得到了极大的简化。

import numpy as np

def calculate_result(input):
    a=np.linspace(1.0, 1000.0, num=10000)   # just an example
    result = input*a
  return(result)

input =[1,2,3,4]

for i in range(0,len(input(i))):
    t.Thread(target=calculate_result, args=(input))
    t. start()  
    #Here I want to receive the return value from the thread

我正在寻找一种从每个线程的线程/函数中获取返回值的方法，因为在我的任务中每个线程计算不同的值。

我发现了另一个问题 (how to get the return value from a thread in python?)，有人正在寻找类似的问题（无 ndarrays），并且使用 ThreadPool 和异步处理...

----------------------------------------------- --------------------------------

感谢您的回答！ 由于您的帮助，我现在正在寻找一种方法来解决我的多处理模块问题。为了让您更好地了解我的工作，请参阅我的以下说明。

说明：

我的“input_data”是一个包含 282240 个 uint32 类型元素的 ndarray

在'calculation_function()'中，我使用一个for循环来计算 每 12 位一个结果并将其放入 'output_data'

因为这非常慢，所以我将 input_data 拆分为例如4 或 8 部分并在计算函数（）中计算每个部分。

现在我正在寻找一种方法，如何使 4 或 8 函数并行化 来电

数据的顺序是基本的，因为数据是图像和 每个像素都必须在正确的位置。所以函数调用没有。 1 计算第一个和最后一个函数调用的最后一个像素 图片。

计算工作正常，图像可以完全重建 来自我的算法，但我需要并行化来加快时间 关键方面。

总结： 一个输入 ndarray 分为 4 或 8 个部分。每个部分都有 70560 或 35280 个 uint32 值。从每个 12 位中，我计算一个具有 4 或 8 个函数调用的像素。每个函数返回一个 188160 或 94080 像素的 ndarray。所有返回值将被放在一起并重新整形为图像。

什么都有效： 计算已经开始，我可以重建我的图像了

问题： 函数调用是连续进行的，但每次图像重建都很慢

主要目标： 通过并行化函数调用来加速函数调用。

代码：

def decompress(payload,WIDTH,HEIGHT):
    # INPUTS / OUTPUTS
    n_threads = 4                                                                           
    img_input = np.fromstring(payload, dtype='uint32')                                      
    img_output = np.zeros((WIDTH * HEIGHT), dtype=np.uint32)                            
    n_elements_part = np.int(len(img_input) / n_threads)                                    
    input_part=np.zeros((n_threads,n_elements_part)).astype(np.uint32)                      
    output_part =np.zeros((n_threads,np.int(n_elements_part/3*8))).astype(np.uint32)        

    # DEFINE PARTS (here 4 different ones)
    start = np.zeros(n_threads).astype(np.int)                          
    end = np.zeros(n_threads).astype(np.int)                            
    for i in range(0,n_threads):
        start[i] = i * n_elements_part
        end[i] = (i+1) * n_elements_part -1

    # COPY IMAGE DATA
    for idx in range(0,n_threads):
        input_part [idx,:] = img_input[start[idx]:end[idx]+1]


    for idx in range(0,n_threads):                          # following line is the function_call that should be parallized
        output_part[idx,:] = decompress_part2(input_part[idx],output_part[idx])



    # COPY PARTS INTO THE IMAGE
    img_output[0     : 188160] = output_part[0,:]
    img_output[188160: 376320] = output_part[1,:]
    img_output[376320: 564480] = output_part[2,:]
    img_output[564480: 752640] = output_part[3,:]

    # RESHAPE IMAGE
    img_output = np.reshape(img_output,(HEIGHT, WIDTH))

    return img_output

请不要照顾我的初学者编程风格 :) 只是在寻找一个解决方案，如何将函数调用与多处理模块并行化并取回返回的 ndarrays。

非常感谢您的帮助！

Answer 1

您可以使用多处理模块中的进程池

        def test(a):
           return a

        from multiprocessing.dummy import Pool
        p = Pool(3)
        a=p.starmap(test, zip([1,2,3]))
        print(a)
        p.close()
        p.join()

Answer 2

kar 的回答有效，但请记住，他正在使用可能受 GIL 限制的 .dummy 模块。这是有关它的更多信息： multiprocessing.dummy in Python is not utilising 100% cpu