从线程返回值

Question

在 Python 中，如何让线程将元组或我选择的任何值返回给父级？

Answer 1

我建议您在启动之前实例化一个 Queue.Queue线程，并将其作为线程的参数之一传递：在线程完成之前，它会将结果.put放在它作为参数接收的队列上。父级可以随意 .get 或 .get_nowait 。

队列通常是在 Python 中安排线程同步和通信的最佳方式：它们本质上是线程安全的消息传递工具——通常是组织多任务处理的最佳方式！-)

Answer 2

您应该传递一个 Queue 实例作为参数，然后您应该将返回对象 .put() 放入队列中。无论您放置什么对象，您都可以通过queue.get()收集返回值。

示例：

queue = Queue.Queue()
thread_ = threading.Thread(
                target=target_method,
                name="Thread1",
                args=[params, queue],
                )
thread_.start()
thread_.join()
queue.get()

def target_method(self, params, queue):
 """
 Some operations right here
 """
 your_return = "Whatever your object is"
 queue.put(your_return)

用于多线程：

#Start all threads in thread pool
    for thread in pool:
        thread.start()
        response = queue.get()
        thread_results.append(response)

#Kill all threads
    for thread in pool:
        thread.join()

我使用这个实现，它对我来说非常有用。我希望你这样做。

Answer 3

使用lambda包装目标线程函数，并使用队列将其返回值传递回父线程。 （您的原始目标函数保持不变，无需额外的队列参数。）

示例代码：

import threading
import queue
def dosomething(param):
    return param * 2
que = queue.Queue()
thr = threading.Thread(target = lambda q, arg : q.put(dosomething(arg)), args = (que, 2))
thr.start()
thr.join()
while not que.empty():
    print(que.get())

输出：

4

Answer 4

如果您调用 join() 来等待线程完成，您可以简单地将结果附加到 Thread 实例本身，然后在 join() 返回后从主线程检索它。

另一方面，您没有告诉我们您打算如何发现线程已完成并且结果可用。如果您已经有办法做到这一点，它可能会向您（以及我们，如果您要告诉我们的话）指出获得结果的最佳方法。

Answer 5

我很惊讶没有人提到你可以将它传递给一个可变的：

from threading import Thread

def task(thread_return):
    thread_return['success'] = True

thread_return={'success': False}
Thread(target=task, args=(thread_return,)).start()

print(thread_return)
{'success': True}

也许这有我不知道的重大问题。

Answer 6

另一种方法是将回调函数传递给线程。这提供了一种简单、安全且灵活的方法，可以随时从新线程向父级返回值。

# A sample implementation

import threading
import time

class MyThread(threading.Thread):
    def __init__(self, cb):
        threading.Thread.__init__(self)
        self.callback = cb

    def run(self):
        for i in range(10):
            self.callback(i)
            time.sleep(1)


# test

import sys

def count(x):
    print x
    sys.stdout.flush()

t = MyThread(count)
t.start()

Answer 7

您可以使用同步 queue 模块。
考虑您需要从数据库中检查具有已知 ID 的用户信息：

def check_infos(user_id, queue):
    result = send_data(user_id)
    queue.put(result)

现在您可以像这样获取数据：

import queue, threading
queued_request = queue.Queue()
check_infos_thread = threading.Thread(target=check_infos, args=(user_id, queued_request))
check_infos_thread.start()
final_result = queued_request.get()

Answer 8

对于简单的程序，上面的答案对我来说看起来有点矫枉过正。我会 en-nicen 可变方法：

class RetVal:
 def __init__(self):
   self.result = None


def threadfunc(retVal):
  retVal.result = "your return value"

retVal = RetVal()
thread = Thread(target = threadfunc, args = (retVal))

thread.start()
thread.join()
print(retVal.result)

Answer 9

概念验证：

import random
import threading

class myThread( threading.Thread ):
    def __init__( self, arr ):
        threading.Thread.__init__( self )
        self.arr = arr
        self.ret = None

    def run( self ):
        self.myJob( self.arr )

    def join( self ):
        threading.Thread.join( self )
        return self.ret

    def myJob( self, arr ):
        self.ret = sorted( self.arr )
        return

#Call the main method if run from the command line.
if __name__ == '__main__':
    N = 100

    arr = [ random.randint( 0, 100 ) for x in range( N ) ]
    th = myThread( arr )
    th.start( )
    sortedArr = th.join( )

    print "arr2: ", sortedArr

Answer 10

嗯，在 Python 线程模块中，有与锁关联的条件对象。一种方法 acquire() 将返回从底层方法返回的任何值。有关更多信息：Python 条件对象

Answer 11

基于 jcomeau_ictx 的建议。我遇到的最简单的一个。这里的要求是从服务器上运行的三个不同进程获取退出状态状态，并在三个进程都成功时触发另一个脚本。这似乎工作正常

  class myThread(threading.Thread):
        def __init__(self,threadID,pipePath,resDict):
            threading.Thread.__init__(self)
            self.threadID=threadID
            self.pipePath=pipePath
            self.resDict=resDict

        def run(self):
            print "Starting thread %s " % (self.threadID)
            if not os.path.exists(self.pipePath):
            os.mkfifo(self.pipePath)
            pipe_fd = os.open(self.pipePath, os.O_RDWR | os.O_NONBLOCK )
           with os.fdopen(pipe_fd) as pipe:
                while True:
                  try:
                     message =  pipe.read()
                     if message:
                        print "Received: '%s'" % message
                        self.resDict['success']=message
                        break
                     except:
                        pass

    tResSer={'success':'0'}
    tResWeb={'success':'0'}
    tResUisvc={'success':'0'}


    threads = []

    pipePathSer='/tmp/path1'
    pipePathWeb='/tmp/path2'
    pipePathUisvc='/tmp/path3'

    th1=myThread(1,pipePathSer,tResSer)
    th2=myThread(2,pipePathWeb,tResWeb)
    th3=myThread(3,pipePathUisvc,tResUisvc)

    th1.start()
    th2.start()
    th3.start()

    threads.append(th1)
    threads.append(th2)
    threads.append(th3)

    for t in threads:
        print t.join()

    print "Res: tResSer %s tResWeb %s tResUisvc %s" % (tResSer,tResWeb,tResUisvc)
    # The above statement prints updated values which can then be further processed

Answer 12

以下包装函数将包装现有函数并返回一个指向线程的对象（以便您可以调用 start()、join() 等它）以及访问/查看其最终返回值。

def threadwrap(func,args,kwargs):
   class res(object): result=None
   def inner(*args,**kwargs): 
     res.result=func(*args,**kwargs)
   import threading
   t = threading.Thread(target=inner,args=args,kwargs=kwargs)
   res.thread=t
   return res

def myFun(v,debug=False):
  import time
  if debug: print "Debug mode ON"
  time.sleep(5)
  return v*2

x=threadwrap(myFun,[11],{"debug":True})
x.thread.start()
x.thread.join()
print x.result

它看起来不错，并且 threading.Thread 类似乎很容易使用这种功能进行扩展（*），所以我想知道为什么它还不存在。上述方法有缺陷吗？

(*) 请注意，husanu 对这个问题的回答正是这样做的，子类化 threading.Thread 产生了 join() 给出返回值的版本。

Answer 13

这是实现多线程的代码。

线程 1 正在将 10 到 20 之间的数字相加。
线程 2 正在将 21 到 30 之间的数字相加。

最后，输出返回到主程序，在那里它可以执行最终的加法。 （本程序中未显示）但您可以使用 numpy 调用。

import threading
import os
import queue

def task1(num, queue): 
    print("\n Current thread: {}".format(threading.current_thread().name)) 
    count = 0
    sum1 = 0
    while count <= 10:
        sum1 = sum1 + num
        num = num + 1
        count = count + 1
    print('\n'+str(sum1))
    queue.put(sum1)


if __name__ == "__main__":

    queue = queue.Queue()

    # print ID of current process 
    print("\n Process ID is: {}".format(os.getpid())) 

    # print name of main thread 
    print("\n Main thread is: {}".format(threading.main_thread().name)) 

    # creating threads 
    t1 = threading.Thread(target=task1, name='t1',args=[10,queue]) 
    t2 = threading.Thread(target=task1, name='t2',args=[21,queue])

    #Store thread names in a list
    pool = [t1,t2]

    #Used to store temporary values
    thread_results = []

    # starting threads
    #Start all threads in thread pool
    for thread in pool:
        thread.start()
        response = queue.get()
        thread_results.append(response)

    #Kill all threads
    for thread in pool:
        thread.join()

    print(thread_results)

Answer 14

我认为 threading.Thread 子类化以更清晰的方式工作，因为结果仅与线程实例相关，而不求助于可能具有其他含义的其他对象（即队列）。
这是我的 2 美分示例：

class RThread(threading.Thread):
    def __init__(self, target, args, daemon: bool = False):
        super().__init__(group=None, target=target, args=args, daemon=daemon)
        self.target = target
        self.args = args
        self.daemon = daemon
        self.result = None

    def run(self):
        self.result = self.target(*self.args)

def foo(secs:int=10):
    print(f"Starting thread at {time.strftime('%X')}")
    time.sleep(secs)
    print(f"Ending  thread at {time.strftime('%X')}")
    return secs

t = RThread(target=foo, args=(random.randint(1, 10),), daemon=False)

def main():
    global t
    os.system("clear")
    try:
        name = input("Please input your name: ")
        t.start()
        print(f"Hi, {name}.")
        t.join()
        print(f"The result of thread is: {t.result}")
        print(f"Bye, {name}.")
    except KeyboardInterrupt:
        print("\nBye, bye.")

Key 是子类线程的 run 方法，它将目标函数的返回值链接到线程的“结果”属性。简单明了。