我找到的大多数答案都很长，需要熟悉其他模块或高级python特性，除非他们已经熟悉答案所谈论的一切，否则会让人感到困惑。

简化方法的工作代码:

import threading

class ThreadWithResult(threading.Thread):
    def __init__(self, group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None):
        def function():
            self.result = target(*args, **kwargs)
        super().__init__(group=group, target=function, name=name, daemon=daemon)

示例代码:

import time, random


def function_to_thread(n):
    count = 0
    while count < 3:
            print(f'still running thread {n}')
            count +=1
            time.sleep(3)
    result = random.random()
    print(f'Return value of thread {n} should be: {result}')
    return result


def main():
    thread1 = ThreadWithResult(target=function_to_thread, args=(1,))
    thread2 = ThreadWithResult(target=function_to_thread, args=(2,))
    thread1.start()
    thread2.start()
    thread1.join()
    thread2.join()
    print(thread1.result)
    print(thread2.result)

main()

解释: 我想大大简化事情，所以我创建了一个ThreadWithResult类，并让它继承threading.Thread。__init__中的嵌套函数函数调用我们想要保存值的线程函数，并将该嵌套函数的结果保存为实例属性self。线程执行完成后的结果。

创建this的实例与创建threading.Thread的实例是相同的。将希望在新线程上运行的函数传递给目标参数，将函数可能需要的任何参数传递给args参数，将任何关键字参数传递给kwargs参数。

e.g.

my_thread = ThreadWithResult(target=my_function, args=(arg1, arg2, arg3))

我认为这比绝大多数答案更容易理解，而且这种方法不需要额外的导入!我加入了time和random模块来模拟线程的行为，但它们并不是实现最初问题中所要求的功能所必需的。

我知道我是在这个问题被问到很久之后才回答的，但我希望这能在未来帮助更多的人!

编辑:我创建了保存线程结果的PyPI包，允许你访问上面相同的代码，并在项目中重用它(GitHub代码在这里)。PyPI包完全扩展了线程。线程类，因此您可以设置在线程上设置的任何属性。线程在ThreadWithResult类!

上面的原始答案介绍了这个子类背后的主要思想，但要了解更多信息，请参阅这里更详细的解释(来自模块docstring)。

快速使用示例:

pip3 install -U save-thread-result     # MacOS/Linux
pip  install -U save-thread-result     # Windows

python3     # MacOS/Linux
python      # Windows

from save_thread_result import ThreadWithResult

# As of Release 0.0.3, you can also specify values for
#`group`, `name`, and `daemon` if you want to set those
# values manually.
thread = ThreadWithResult(
    target = my_function,
    args   = (my_function_arg1, my_function_arg2, ...)
    kwargs = {my_function_kwarg1: kwarg1_value, my_function_kwarg2: kwarg2_value, ...}
)

thread.start()
thread.join()
if getattr(thread, 'result', None):
    print(thread.result)
else:
    # thread.result attribute not set - something caused
    # the thread to terminate BEFORE the thread finished
    # executing the function passed in through the
    # `target` argument
    print('ERROR! Something went wrong while executing this thread, and the function you passed in did NOT complete!!')

# seeing help about the class and information about the threading.Thread super class methods and attributes available:
help(ThreadWithResult)

根据上面提到的，下面是适用于Python3的更通用的解决方案。

import threading

class ThreadWithReturnValue(threading.Thread):
    def __init__(self, *init_args, **init_kwargs):
        threading.Thread.__init__(self, *init_args, **init_kwargs)
        self._return = None
    def run(self):
        self._return = self._target(*self._args, **self._kwargs)
    def join(self):
        threading.Thread.join(self)
        return self._return

使用

        th = ThreadWithReturnValue(target=requests.get, args=('http://www.google.com',))
        th.start()
        response = th.join()
        response.status_code  # => 200

join总是返回None，我认为你应该子类化Thread来处理返回代码等。

如上所述，多处理池比基本线程要慢得多。使用一些回答中提出的队列是一种非常有效的替代方法。我已经将它与字典一起使用，以便能够运行许多小线程，并通过将它们与字典结合来恢复多个答案:

#!/usr/bin/env python3

import threading
# use Queue for python2
import queue
import random

LETTERS = 'abcdefghijklmnopqrstuvwxyz'
LETTERS = [ x for x in LETTERS ]

NUMBERS = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

def randoms(k, q):
    result = dict()
    result['letter'] = random.choice(LETTERS)
    result['number'] = random.choice(NUMBERS)
    q.put({k: result})

threads = list()
q = queue.Queue()
results = dict()

for name in ('alpha', 'oscar', 'yankee',):
    threads.append( threading.Thread(target=randoms, args=(name, q)) )
    threads[-1].start()
_ = [ t.join() for t in threads ]
while not q.empty():
    results.update(q.get())

print(results)

这是一个很老的问题，但我想分享一个简单的解决方案，它对我的开发过程有帮助。

这个答案背后的方法论是这样一个事实，即“新的”目标函数，内部是将原始函数的结果(通过__init__函数传递)通过所谓的闭包分配给包装器的结果实例属性。

这允许包装器类保留返回值以供调用者随时访问。

注意:这个方法不需要使用线程的任何mangded方法或私有方法。线程类，虽然没有考虑屈服函数(OP没有提到屈服函数)。

享受吧!

from threading import Thread as _Thread


class ThreadWrapper:
    def __init__(self, target, *args, **kwargs):
        self.result = None
        self._target = self._build_threaded_fn(target)
        self.thread = _Thread(
            target=self._target,
            *args,
            **kwargs
        )

    def _build_threaded_fn(self, func):
        def inner(*args, **kwargs):
            self.result = func(*args, **kwargs)
        return inner

此外，你可以用下面的代码运行pytest(假设你已经安装了它)来演示结果:

import time
from commons import ThreadWrapper


def test():

    def target():
        time.sleep(1)
        return 'Hello'

    wrapper = ThreadWrapper(target=target)
    wrapper.thread.start()

    r = wrapper.result
    assert r is None

    time.sleep(2)

    r = wrapper.result
    assert r == 'Hello'

Parris / kindall的answer join/return answer移植到Python 3:

from threading import Thread

def foo(bar):
    print('hello {0}'.format(bar))
    return "foo"

class ThreadWithReturnValue(Thread):
    def __init__(self, group=None, target=None, name=None, args=(), kwargs=None, *, daemon=None):
        Thread.__init__(self, group, target, name, args, kwargs, daemon=daemon)

        self._return = None

    def run(self):
        if self._target is not None:
            self._return = self._target(*self._args, **self._kwargs)

    def join(self):
        Thread.join(self)
        return self._return


twrv = ThreadWithReturnValue(target=foo, args=('world!',))

twrv.start()
print(twrv.join())   # prints foo

注意，Thread类在Python 3中实现的方式不同。