类内的Python线程

Question

我最近开始使用python的线程模块。经过一番尝试和错误之后，我使用大多数教程中提供的以下示例代码成功地实现了基本线程。

class SomeThread(threading.Thread):
    def __init__(self, count):
        threading.Thread.__init__(self)

    def run(self):
        print "Do something"

我的问题是：我有一个类，它有一个类变量，还有一个函数，我想在一个单独的线程中运行。但是，该函数使用类变量，也写入类变量。就像这样：

class MyClass:
    somevar = 'someval'

    def func_to_be_threaded(self):
        # Uses other class functions
        # Do something with class variables

所以我应该如何“把线程类放在MyClass中”。因此，如果调用MyClass（）.func_to_threaded（），它将在线程中运行。

Answer 1

可以将类实例传递给线程：

class SomeThread(threading.Thread):
    def __init__(self, count, instance):
        threading.Thread.__init__(self)
        self.instance = instance

    def run(self):
        print "Do something"
        self.instance.some_param = data
        self.instance.some_function()

Answer 2

如果我理解正确，您想在单独的线程中运行函数吗？有几种方法可以做到这一点。但基本上你的函数是这样包装的：

class MyClass:
    somevar = 'someval'

    def _func_to_be_threaded(self):
        # main body

    def func_to_be_threaded(self):
        threading.Thread(target=self._func_to_be_threaded).start()

它可以通过一个decorator来缩短：

def threaded(fn):
    def wrapper(*args, **kwargs):
        threading.Thread(target=fn, args=args, kwargs=kwargs).start()
    return wrapper

class MyClass:
    somevar = 'someval'

    @threaded
    def func_to_be_threaded(self):
        # main body

---

编辑带句柄的更新版本：

def threaded(fn):
    def wrapper(*args, **kwargs):
        thread = threading.Thread(target=fn, args=args, kwargs=kwargs)
        thread.start()
        return thread
    return wrapper

class MyClass:
    somevar = 'someval'

    @threaded
    def func_to_be_threaded(self):
        print 'xyz'

这可以如下使用：

>>> my_obj = MyClass()
>>> handle = my_obj.func_to_be_threaded()
>>> handle.join()

---

现在，如果希望从函数返回一个值，则可以进一步扩展它。考虑一下：

from threading import Thread
from concurrent.futures import Future

def call_with_future(fn, future, args, kwargs):
    try:
        result = fn(*args, **kwargs)
        future.set_result(result)
    except Exception as exc:
        future.set_exception(exc)

def threaded(fn):
    def wrapper(*args, **kwargs):
        future = Future()
        Thread(target=call_with_future, args=(fn, future, args, kwargs)).start()
        return future
    return wrapper


class MyClass:
    @threaded
    def get_my_value(self):
        return 1

>>> my_obj = MyClass()
>>> fut = my_obj.get_my_value()  # this will run in a separate thread
>>> fut.result()  # will block until result is computed
1

如果您没有concurrent.futures.Future class（例如，因为您使用的是Python2.7或更早版本），那么可以使用这个简化的实现：

from threading import Event

class Future(object):
    def __init__(self):
        self._ev = Event()

    def set_result(self, result):
        self._result = result
        self._ev.set()

    def set_exception(self, exc):
        self._exc = exc
        self._ev.set()

    def result(self):
        self._ev.wait()
        if hasattr(self, '_exc'):
            raise self._exc
        return self._result

我建议阅读concurrent.futures模块，因为它有很多整洁的工具。例如，Thread类应该替换为一个ThreadPoolExecutor实例来限制并发性（例如，您不想发送10k线程的垃圾邮件）。同样使用ThreadPoolExecutor代码更加简单（而且不易出错）：

from concurrent.futures import ThreadPoolExecutor

tp = ThreadPoolExecutor(10)  # max 10 threads

def threaded(fn):
    def wrapper(*args, **kwargs):
        return tp.submit(fn, *args, **kwargs)  # returns Future object
    return wrapper

记住，在完成所有并行工作之后，您必须tp.shutdown()。

Answer 3

我敢肯定你不能把一个函数变成线程。

整个类将被线程化。当您实例化对象时，它的__init__将在另一个线程上调用，然后当您在该对象上调用start()时，它的run()将在另一个线程上调用一次。

所以，如果你有一个任务需要在它自己的线程上（disc IO，socket listening，等等），那么你需要一个类来处理这个任务。

@ndpu的答案解决了您的范围/访问问题。