下面的函数foo返回一个字符串'foo'。我如何才能获得从线程的目标返回的值'foo' ?
from threading import Thread
def foo(bar):
print('hello {}'.format(bar))
return 'foo'
thread = Thread(target=foo, args=('world!',))
thread.start()
return_value = thread.join()
上面所示的“一种明显的方法”不起作用:thread.join()返回None。
当前回答
Jake的回答很好，但如果您不想使用线程池(您不知道需要多少线程，但可以根据需要创建它们)，那么在线程之间传输信息的一个好方法是内置的Queue。队列类，因为它提供线程安全性。
我创建了以下装饰器，使其以类似于线程池的方式工作:
def threaded(f, daemon=False):
import Queue
def wrapped_f(q, *args, **kwargs):
'''this function calls the decorated function and puts the
result in a queue'''
ret = f(*args, **kwargs)
q.put(ret)
def wrap(*args, **kwargs):
'''this is the function returned from the decorator. It fires off
wrapped_f in a new thread and returns the thread object with
the result queue attached'''
q = Queue.Queue()
t = threading.Thread(target=wrapped_f, args=(q,)+args, kwargs=kwargs)
t.daemon = daemon
t.start()
t.result_queue = q
return t
return wrap
然后你就把它用作:
@threaded
def long_task(x):
import time
x = x + 5
time.sleep(5)
return x
# does not block, returns Thread object
y = long_task(10)
print y
# this blocks, waiting for the result
result = y.result_queue.get()
print result
装饰函数每次被调用时都会创建一个新线程，并返回一个thread对象，其中包含将接收结果的队列。
更新
自从我发布这个答案已经有一段时间了，但它仍然得到了观看，所以我想我应该更新它，以反映我在新版本的Python中这样做的方式:
Python 3.2并发添加。期货模块，为并行任务提供高级接口。它提供了ThreadPoolExecutor和ProcessPoolExecutor，因此您可以使用具有相同api的线程或进程池。
该api的一个好处是将任务提交给Executor将返回一个Future对象，该对象将以您提交的可调用对象的返回值结束。
这使得附加队列对象成为不必要的，这大大简化了装饰器:
_DEFAULT_POOL = ThreadPoolExecutor()
def threadpool(f, executor=None):
@wraps(f)
def wrap(*args, **kwargs):
return (executor or _DEFAULT_POOL).submit(f, *args, **kwargs)
return wrap
如果没有传入，将使用默认的模块线程池执行器。
用法和前面的非常相似:
@threadpool
def long_task(x):
import time
x = x + 5
time.sleep(5)
return x
# does not block, returns Future object
y = long_task(10)
print y
# this blocks, waiting for the result
result = y.result()
print result
如果您使用的是Python 3.4+，那么使用此方法(以及一般的Future对象)的一个非常好的特性是可以将返回的Future对象包装起来以将其转换为asyncio。使用asyncio.wrap_future。这使得它很容易与协程一起工作:
result = await asyncio.wrap_future(long_task(10))
如果您不需要访问底层并发。对象，你可以在装饰器中包含wrap:
_DEFAULT_POOL = ThreadPoolExecutor()
def threadpool(f, executor=None):
@wraps(f)
def wrap(*args, **kwargs):
return asyncio.wrap_future((executor or _DEFAULT_POOL).submit(f, *args, **kwargs))
return wrap
然后，当你需要将cpu密集型代码或阻塞代码从事件循环线程中推出时，你可以将它放在装饰函数中:
@threadpool
def some_long_calculation():
...
# this will suspend while the function is executed on a threadpool
result = await some_long_calculation()
其他回答
Jake的回答很好，但如果您不想使用线程池(您不知道需要多少线程，但可以根据需要创建它们)，那么在线程之间传输信息的一个好方法是内置的Queue。队列类，因为它提供线程安全性。
我创建了以下装饰器，使其以类似于线程池的方式工作:
def threaded(f, daemon=False):
import Queue
def wrapped_f(q, *args, **kwargs):
'''this function calls the decorated function and puts the
result in a queue'''
ret = f(*args, **kwargs)
q.put(ret)
def wrap(*args, **kwargs):
'''this is the function returned from the decorator. It fires off
wrapped_f in a new thread and returns the thread object with
the result queue attached'''
q = Queue.Queue()
t = threading.Thread(target=wrapped_f, args=(q,)+args, kwargs=kwargs)
t.daemon = daemon
t.start()
t.result_queue = q
return t
return wrap
然后你就把它用作:
@threaded
def long_task(x):
import time
x = x + 5
time.sleep(5)
return x
# does not block, returns Thread object
y = long_task(10)
print y
# this blocks, waiting for the result
result = y.result_queue.get()
print result
装饰函数每次被调用时都会创建一个新线程，并返回一个thread对象，其中包含将接收结果的队列。
更新
自从我发布这个答案已经有一段时间了，但它仍然得到了观看，所以我想我应该更新它，以反映我在新版本的Python中这样做的方式:
Python 3.2并发添加。期货模块，为并行任务提供高级接口。它提供了ThreadPoolExecutor和ProcessPoolExecutor，因此您可以使用具有相同api的线程或进程池。
该api的一个好处是将任务提交给Executor将返回一个Future对象，该对象将以您提交的可调用对象的返回值结束。
这使得附加队列对象成为不必要的，这大大简化了装饰器:
_DEFAULT_POOL = ThreadPoolExecutor()
def threadpool(f, executor=None):
@wraps(f)
def wrap(*args, **kwargs):
return (executor or _DEFAULT_POOL).submit(f, *args, **kwargs)
return wrap
如果没有传入，将使用默认的模块线程池执行器。
用法和前面的非常相似:
@threadpool
def long_task(x):
import time
x = x + 5
time.sleep(5)
return x
# does not block, returns Future object
y = long_task(10)
print y
# this blocks, waiting for the result
result = y.result()
print result
如果您使用的是Python 3.4+，那么使用此方法(以及一般的Future对象)的一个非常好的特性是可以将返回的Future对象包装起来以将其转换为asyncio。使用asyncio.wrap_future。这使得它很容易与协程一起工作:
result = await asyncio.wrap_future(long_task(10))
如果您不需要访问底层并发。对象，你可以在装饰器中包含wrap:
_DEFAULT_POOL = ThreadPoolExecutor()
def threadpool(f, executor=None):
@wraps(f)
def wrap(*args, **kwargs):
return asyncio.wrap_future((executor or _DEFAULT_POOL).submit(f, *args, **kwargs))
return wrap
然后，当你需要将cpu密集型代码或阻塞代码从事件循环线程中推出时，你可以将它放在装饰函数中:
@threadpool
def some_long_calculation():
...
# this will suspend while the function is executed on a threadpool
result = await some_long_calculation()
这是我根据@Kindall的回答创建的版本。
这个版本使得您所要做的就是输入带有参数的命令来创建新线程。
这是用Python 3.8做的:
from threading import Thread
from typing import Any
def test(plug, plug2, plug3):
print(f"hello {plug}")
print(f'I am the second plug : {plug2}')
print(plug3)
return 'I am the return Value!'
def test2(msg):
return f'I am from the second test: {msg}'
def test3():
print('hello world')
def NewThread(com, Returning: bool, *arguments) -> Any:
"""
Will create a new thread for a function/command.
:param com: Command to be Executed
:param arguments: Arguments to be sent to Command
:param Returning: True/False Will this command need to return anything
"""
class NewThreadWorker(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
ntw = NewThreadWorker(target = com, args = (*arguments,))
ntw.start()
if Returning:
return ntw.join()
if __name__ == "__main__":
print(NewThread(test, True, 'hi', 'test', test2('hi')))
NewThread(test3, True)
您可以在线程函数的作用域之上定义一个可变变量，并将结果添加到该变量中。(我还修改了代码，使其与python3兼容)
returns = {}
def foo(bar):
print('hello {0}'.format(bar))
returns[bar] = 'foo'
from threading import Thread
t = Thread(target=foo, args=('world!',))
t.start()
t.join()
print(returns)
返回{'world!”:“foo”}
如果使用函数input作为结果字典的键，则保证每个惟一的输入都在结果中给出一个条目
如上所述，多处理池比基本线程要慢得多。使用一些回答中提出的队列是一种非常有效的替代方法。我已经将它与字典一起使用，以便能够运行许多小线程，并通过将它们与字典结合来恢复多个答案:
#!/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)
我正在使用这个包装器，它可以轻松地将任何函数转换为在线程中运行-照顾它的返回值或异常。它不会增加队列开销。
def threading_func(f):
"""Decorator for running a function in a thread and handling its return
value or exception"""
def start(*args, **kw):
def run():
try:
th.ret = f(*args, **kw)
except:
th.exc = sys.exc_info()
def get(timeout=None):
th.join(timeout)
if th.exc:
raise th.exc[0], th.exc[1], th.exc[2] # py2
##raise th.exc[1] #py3
return th.ret
th = threading.Thread(None, run)
th.exc = None
th.get = get
th.start()
return th
return start
用法示例
def f(x):
return 2.5 * x
th = threading_func(f)(4)
print("still running?:", th.is_alive())
print("result:", th.get(timeout=1.0))
@threading_func
def th_mul(a, b):
return a * b
th = th_mul("text", 2.5)
try:
print(th.get())
except TypeError:
print("exception thrown ok.")
线程模块注意事项
线程函数的舒适返回值和异常处理是“python”的常见需求，而且threading模块应该已经提供了——可能直接在标准Thread类中。对于简单的任务，ThreadPool有太多的开销——3个管理线程，很多官僚主义。不幸的是，线程的布局最初是从Java中复制的——例如，从仍然无用的构造函数参数组1 (!)
