La foosiguiente función devuelve una cadena 'foo'. ¿Cómo puedo obtener el valor 'foo'que se devuelve desde el objetivo del hilo?

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()

La "forma obvia de hacerlo", que se muestra arriba, no funciona: thread.join()devuelta None.

En Python 3.2+, el concurrent.futuresmódulo stdlib proporciona una API de nivel superior para threading, incluyendo pasar valores de retorno o excepciones de un subproceso de trabajo al subproceso principal:

import concurrent.futures

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

with concurrent.futures.ThreadPoolExecutor() as executor:
    future = executor.submit(foo, 'world!')
    return_value = future.result()
    print(return_value)

FWIW, el multiprocessingmódulo tiene una buena interfaz para esto usando la Poolclase. Y si desea seguir con hilos en lugar de procesos, puede usar la multiprocessing.pool.ThreadPoolclase como un reemplazo directo.

def foo(bar, baz):
  print 'hello {0}'.format(bar)
  return 'foo' + baz

from multiprocessing.pool import ThreadPool
pool = ThreadPool(processes=1)

async_result = pool.apply_async(foo, ('world', 'foo')) # tuple of args for foo

# do some other stuff in the main process

return_val = async_result.get()  # get the return value from your function.

Una forma que he visto es pasar un objeto mutable, como una lista o un diccionario, al constructor del hilo, junto con un índice u otro identificador de algún tipo. El hilo puede almacenar sus resultados en su ranura dedicada en ese objeto. Por ejemplo:

def foo(bar, result, index):
    print 'hello {0}'.format(bar)
    result[index] = "foo"

from threading import Thread

threads = [None] * 10
results = [None] * 10

for i in range(len(threads)):
    threads[i] = Thread(target=foo, args=('world!', results, i))
    threads[i].start()

# do some other stuff

for i in range(len(threads)):
    threads[i].join()

print " ".join(results)  # what sound does a metasyntactic locomotive make?

Si realmente desea join()devolver el valor de retorno de la función llamada, puede hacerlo con una Threadsubclase como la siguiente:

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={}, Verbose=None):
        Thread.__init__(self, group, target, name, args, kwargs, Verbose)
        self._return = None
    def run(self):
        if self._Thread__target is not None:
            self._return = self._Thread__target(*self._Thread__args,
                                                **self._Thread__kwargs)
    def join(self):
        Thread.join(self)
        return self._return

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

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

Eso se vuelve un poco complicado debido a algunos cambios de nombres, y accede a estructuras de datos "privadas" que son específicas de la Threadimplementación ... pero funciona.

Para python3

class ThreadWithReturnValue(Thread):
    def __init__(self, group=None, target=None, name=None,
                 args=(), kwargs={}, Verbose=None):
        Thread.__init__(self, group, target, name, args, kwargs)
        self._return = None
    def run(self):
        print(type(self._target))
        if self._target is not None:
            self._return = self._target(*self._args,
                                                **self._kwargs)
    def join(self, *args):
        Thread.join(self, *args)
        return self._return

La respuesta de Jake es buena, pero si no desea utilizar un conjunto de subprocesos (no sabe cuántos subprocesos necesitará, pero créelos según sea necesario), entonces una buena forma de transmitir información entre subprocesos es el incorporado Clase Queue.Queue , ya que ofrece seguridad de subprocesos.

Creé el siguiente decorador para que actúe de manera similar al conjunto de subprocesos:

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

Entonces solo lo usas como:

@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

La función decorada crea un nuevo hilo cada vez que se llama y devuelve un objeto Thread que contiene la cola que recibirá el resultado.

ACTUALIZAR

Ha pasado bastante tiempo desde que publiqué esta respuesta, pero aún recibe vistas, así que pensé en actualizarla para reflejar la forma en que lo hago en las versiones más recientes de Python:

Python 3.2 agregado en el concurrent.futuresmódulo que proporciona una interfaz de alto nivel para tareas paralelas. Proporciona ThreadPoolExecutoryProcessPoolExecutor , por lo que puede utilizar un subproceso o grupo de procesos con la misma API.

Un beneficio de esta API es que enviar una tarea a ExecutorunFuture objeto objeto, que se completará con el valor de retorno del invocable que envíe.

Esto hace que adjuntar un queueobjeto sea innecesario, lo que simplifica bastante el decorador:

_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

Esto usará un módulo predeterminado ejecutor de agrupación de hilos de si no se pasa uno.

El uso es muy similar al anterior:

@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

Si está utilizando Python 3.4+, una característica realmente agradable de usar este método (y los objetos Futuros en general) es que el futuro devuelto se puede ajustar para convertirlo en un asyncio.Futurecon asyncio.wrap_future. Esto hace que funcione fácilmente con corutinas:

result = await asyncio.wrap_future(long_task(10))

Si no necesita acceso al concurrent.Futureobjeto subyacente , puede incluir el ajuste en el decorador:

_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

Luego, cada vez que necesite eliminar el código de bloqueo o de uso intensivo de la CPU del hilo del bucle de eventos, puede colocarlo en una función decorada:

@threadpool
def some_long_calculation():
    ...

# this will suspend while the function is executed on a threadpool
result = await some_long_calculation()

Otra solución que no requiere cambiar su código existente:

import Queue
from threading import Thread

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

que = Queue.Queue()

t = Thread(target=lambda q, arg1: q.put(foo(arg1)), args=(que, 'world!'))
t.start()
t.join()
result = que.get()
print result

También se puede ajustar fácilmente a un entorno multiproceso:

import Queue
from threading import Thread

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

que = Queue.Queue()
threads_list = list()

t = Thread(target=lambda q, arg1: q.put(foo(arg1)), args=(que, 'world!'))
t.start()
threads_list.append(t)

# Add more threads here
...
threads_list.append(t2)
...
threads_list.append(t3)
...

# Join all the threads
for t in threads_list:
    t.join()

# Check thread's return value
while not que.empty():
    result = que.get()
    print result

La respuesta join / returnrespuesta de Parris / kindall portó a 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

Tenga en cuenta que la Threadclase se implementa de manera diferente en Python 3.

Robé la respuesta de Kindall y la limpié un poco.

La parte clave es agregar * args y ** kwargs a join () para manejar el tiempo de espera

class threadWithReturn(Thread):
    def __init__(self, *args, **kwargs):
        super(threadWithReturn, self).__init__(*args, **kwargs)

        self._return = None

    def run(self):
        if self._Thread__target is not None:
            self._return = self._Thread__target(*self._Thread__args, **self._Thread__kwargs)

    def join(self, *args, **kwargs):
        super(threadWithReturn, self).join(*args, **kwargs)

        return self._return

RESPUESTA ACTUALIZADA A CONTINUACIÓN

Esta es mi respuesta popularmente votada, así que decidí actualizar con un código que se ejecutará tanto en py2 como en py3.

Además, veo muchas respuestas a esta pregunta que muestran una falta de comprensión con respecto a Thread.join (). Algunos fallan completamente en manejar el timeoutargumento. Pero también hay un caso de esquina que debe tener en cuenta con respecto a las instancias en las que tiene (1) una función de destino que puede devolver Noney (2) también pasa el timeoutargumento para unirse (). Consulte "PRUEBA 4" para comprender este caso de esquina.

Clase ThreadWithReturn que funciona con py2 y py3:

import sys
from threading import Thread
from builtins import super    # https://stackoverflow.com/a/30159479

if sys.version_info >= (3, 0):
    _thread_target_key = '_target'
    _thread_args_key = '_args'
    _thread_kwargs_key = '_kwargs'
else:
    _thread_target_key = '_Thread__target'
    _thread_args_key = '_Thread__args'
    _thread_kwargs_key = '_Thread__kwargs'

class ThreadWithReturn(Thread):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._return = None

    def run(self):
        target = getattr(self, _thread_target_key)
        if not target is None:
            self._return = target(
                *getattr(self, _thread_args_key),
                **getattr(self, _thread_kwargs_key)
            )

    def join(self, *args, **kwargs):
        super().join(*args, **kwargs)
        return self._return

Algunas pruebas de muestra se muestran a continuación:

import time, random

# TEST TARGET FUNCTION
def giveMe(arg, seconds=None):
    if not seconds is None:
        time.sleep(seconds)
    return arg

# TEST 1
my_thread = ThreadWithReturn(target=giveMe, args=('stringy',))
my_thread.start()
returned = my_thread.join()
# (returned == 'stringy')

# TEST 2
my_thread = ThreadWithReturn(target=giveMe, args=(None,))
my_thread.start()
returned = my_thread.join()
# (returned is None)

# TEST 3
my_thread = ThreadWithReturn(target=giveMe, args=('stringy',), kwargs={'seconds': 5})
my_thread.start()
returned = my_thread.join(timeout=2)
# (returned is None) # because join() timed out before giveMe() finished

# TEST 4
my_thread = ThreadWithReturn(target=giveMe, args=(None,), kwargs={'seconds': 5})
my_thread.start()
returned = my_thread.join(timeout=random.randint(1, 10))

¿Puedes identificar el caso de esquina que posiblemente podamos encontrar con TEST 4?

El problema es que esperamos que giveMe () devuelva None (ver PRUEBA 2), pero también esperamos que join () devuelva None si se agota el tiempo de espera.

returned is None significa ya sea:

(1) eso es lo que giveMe () devolvió, o

(2) unirse () agotó el tiempo de espera

Este ejemplo es trivial ya que sabemos que giveMe () siempre devolverá None. Pero en el caso del mundo real (donde el objetivo puede devolver legítimamente Ninguno u otra cosa), querríamos verificar explícitamente qué sucedió.

A continuación se muestra cómo abordar este caso de esquina:

# TEST 4
my_thread = ThreadWithReturn(target=giveMe, args=(None,), kwargs={'seconds': 5})
my_thread.start()
returned = my_thread.join(timeout=random.randint(1, 10))

if my_thread.isAlive():
    # returned is None because join() timed out
    # this also means that giveMe() is still running in the background
    pass
    # handle this based on your app's logic
else:
    # join() is finished, and so is giveMe()
    # BUT we could also be in a race condition, so we need to update returned, just in case
    returned = my_thread.join()

Usando la cola:

import threading, queue

def calc_square(num, out_queue1):
  l = []
  for x in num:
    l.append(x*x)
  out_queue1.put(l)


arr = [1,2,3,4,5,6,7,8,9,10]
out_queue1=queue.Queue()
t1=threading.Thread(target=calc_square, args=(arr,out_queue1))
t1.start()
t1.join()
print (out_queue1.get())

Mi solución al problema es envolver la función y el hilo en una clase. No requiere el uso de agrupaciones, colas o paso variable de tipo c. También es no bloqueante. Verifica el estado en su lugar. Vea un ejemplo de cómo usarlo al final del código.

import threading

class ThreadWorker():
    '''
    The basic idea is given a function create an object.
    The object can then run the function in a thread.
    It provides a wrapper to start it,check its status,and get data out the function.
    '''
    def __init__(self,func):
        self.thread = None
        self.data = None
        self.func = self.save_data(func)

    def save_data(self,func):
        '''modify function to save its returned data'''
        def new_func(*args, **kwargs):
            self.data=func(*args, **kwargs)

        return new_func

    def start(self,params):
        self.data = None
        if self.thread is not None:
            if self.thread.isAlive():
                return 'running' #could raise exception here

        #unless thread exists and is alive start or restart it
        self.thread = threading.Thread(target=self.func,args=params)
        self.thread.start()
        return 'started'

    def status(self):
        if self.thread is None:
            return 'not_started'
        else:
            if self.thread.isAlive():
                return 'running'
            else:
                return 'finished'

    def get_results(self):
        if self.thread is None:
            return 'not_started' #could return exception
        else:
            if self.thread.isAlive():
                return 'running'
            else:
                return self.data

def add(x,y):
    return x +y

add_worker = ThreadWorker(add)
print add_worker.start((1,2,))
print add_worker.status()
print add_worker.get_results()

joinsiempre regreso None, creo que deberías subclase Threadpara manejar los códigos de retorno y así.

Tomando en consideración @iman comentario sobre @JakeBiesinger respuesta que he recompuesto a tener varias número de hilos:

from multiprocessing.pool import ThreadPool

def foo(bar, baz):
    print 'hello {0}'.format(bar)
    return 'foo' + baz

numOfThreads = 3 
results = []

pool = ThreadPool(numOfThreads)

for i in range(0, numOfThreads):
    results.append(pool.apply_async(foo, ('world', 'foo'))) # tuple of args for foo)

# do some other stuff in the main process
# ...
# ...

results = [r.get() for r in results]
print results

pool.close()
pool.join()

Salud,

Chico.

Puede definir una mutable por encima del alcance de la función de subprocesos y agregar el resultado a eso. (También modifiqué el código para que sea compatible con 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)

Esto vuelve {'world!': 'foo'}

Si usa la entrada de función como la clave para su dictado de resultados, se garantiza que cada entrada única dará una entrada en los resultados

Estoy usando este contenedor, que convierte cómodamente cualquier función para ejecutar en un Thread- cuidando su valor de retorno o excepción. No agrega Queuegastos generales.

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

Ejemplos de uso

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.")

Notas sobre el threadingmódulo

El valor de retorno cómodo y el manejo de excepciones de una función roscada es una necesidad frecuente "Pythonic" y, de hecho, el threadingmódulo ya debería ofrecerlo , posiblemente directamente en la Threadclase estándar . ThreadPooltiene demasiados gastos generales para tareas simples: 3 subprocesos de gestión, mucha burocracia. Lamentablemente Thread, el diseño se copió originalmente de Java, que puede ver, por ejemplo, desde el parámetro del constructor 1st (!), Que sigue siendo inútil group.

Defina su objetivo para
1) tomar un argumento q
2) reemplazar cualquier declaración return fooconq.put(foo); return

entonces una función

def func(a):
    ans = a * a
    return ans

se convertiría

def func(a, q):
    ans = a * a
    q.put(ans)
    return

y luego procederías como tal

from Queue import Queue
from threading import Thread

ans_q = Queue()
arg_tups = [(i, ans_q) for i in xrange(10)]

threads = [Thread(target=func, args=arg_tup) for arg_tup in arg_tups]
_ = [t.start() for t in threads]
_ = [t.join() for t in threads]
results = [q.get() for _ in xrange(len(threads))]

Y puede usar decoradores / envoltorios de funciones para hacerlo, de modo que pueda usar sus funciones existentes targetsin modificarlas, pero siga este esquema básico.

Como se mencionó, el grupo de multiprocesamiento es mucho más lento que el subproceso básico. Usar colas como se propone en algunas respuestas aquí es una alternativa muy efectiva. Lo he usado con diccionarios para poder ejecutar muchos hilos pequeños y recuperar múltiples respuestas combinándolas con diccionarios:

#!/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)

La idea de GuySoft es genial, pero creo que el objeto no necesariamente tiene que heredar de Thread y start () podría eliminarse de la interfaz:

from threading import Thread
import queue
class ThreadWithReturnValue(object):
    def __init__(self, target=None, args=(), **kwargs):
        self._que = queue.Queue()
        self._t = Thread(target=lambda q,arg1,kwargs1: q.put(target(*arg1, **kwargs1)) ,
                args=(self._que, args, kwargs), )
        self._t.start()

    def join(self):
        self._t.join()
        return self._que.get()


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

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

print(twrv.join())   # prints foo

Una solución habitual es envolver su función foocon un decorador como

result = queue.Queue()

def task_wrapper(*args):
    result.put(target(*args))

Entonces todo el código puede verse así

result = queue.Queue()

def task_wrapper(*args):
    result.put(target(*args))

threads = [threading.Thread(target=task_wrapper, args=args) for args in args_list]

for t in threads:
    t.start()
    while(True):
        if(len(threading.enumerate()) < max_num):
            break
for t in threads:
    t.join()
return result

Nota

Una cuestión importante es que los valores de retorno pueden estar desordenados . (De hecho, return valueno se guarda necesariamente en el queue, ya que puede elegir una estructura de datos arbitraria segura para subprocesos )

¿Por qué no solo usar variable global?

import threading


class myThread(threading.Thread):
    def __init__(self, ind, lock):
        threading.Thread.__init__(self)
        self.ind = ind
        self.lock = lock

    def run(self):
        global results
        with self.lock:
            results.append(self.ind)



results = []
lock = threading.Lock()
threads = [myThread(x, lock) for x in range(1, 4)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(results)

La respuesta de Kindall en Python3

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

    def run(self):
        try:
            if self._target:
                self._return = self._target(*self._args, **self._kwargs)
        finally:
            del self._target, self._args, self._kwargs 

    def join(self,timeout=None):
        Thread.join(self,timeout)
        return self._return

Si solo se debe validar Verdadero o Falso a partir de la llamada de una función, una solución más simple que encuentro es actualizar una lista global.

import threading

lists = {"A":"True", "B":"True"}

def myfunc(name: str, mylist):
    for i in mylist:
        if i == 31:
            lists[name] = "False"
            return False
        else:
            print("name {} : {}".format(name, i))

t1 = threading.Thread(target=myfunc, args=("A", [1, 2, 3, 4, 5, 6], ))
t2 = threading.Thread(target=myfunc, args=("B", [11, 21, 31, 41, 51, 61], ))
t1.start()
t2.start()
t1.join()
t2.join()

for value in lists.values():
    if value == False:
        # Something is suspicious 
        # Take necessary action 

Esto es más útil cuando desea encontrar si alguno de los hilos ha devuelto un estado falso para tomar las medidas necesarias.