如何将一些变量绑定到具有conCurent.futures.ThreadPoolExecutor或MultiProcing.pool.ThreadPool的线程?

How to bind some variable to thread with concurrent.futures.ThreadPoolExecutor or multiprocessing.pool.ThreadPool?(如何将一些变量绑定到具有conCurent.futures.ThreadPoolExecutor或MultiProcing.pool.ThreadPool的线程?)
本文介绍了如何将一些变量绑定到具有conCurent.futures.ThreadPoolExecutor或MultiProcing.pool.ThreadPool的线程?的处理方法,对大家解决问题具有一定的参考价值,需要的朋友们下面随着小编来一起学习吧!

问题描述

我想做的事情是这样的:

class MyThread(threading.Thread):
    def __init__(self, host, port):
        threading.Thread.__init__(self)
        # self._sock = self.initsocket(host, port)
        self._id = random.randint(0, 100)

    def run(self):
        for i in range(3):
            print("current id: {}".format(self._id))

def main():
    ts = []
    for i in range(5):
        t = MyThread("localhost", 3001)
        t.start()
        ts.append(t)

    for t in ts:
        t.join()

我得到以下输出:

current id: 10
current id: 10
current id: 13
current id: 43
current id: 13
current id: 10
current id: 83
current id: 83
current id: 83
current id: 13
current id: 98
current id: 43
current id: 98
current id: 43
current id: 98

这就是我想要的输出。正如您所看到的,我的_id在不同的线程中是不同的,但在单个线程中,我共享相同的_id。(_id只是这些变量之一,我还有许多其他类似的变量)。

现在,我想对multiprocessing.pool.ThreadPool

执行相同的操作
class MyProcessor():
    def __init__(self, host, port):
        # self._sock = self.initsocket(host, port)
        self._id = random.randint(0, 100)

    def __call__(self, i):
        print("current id: {}".format(self._id))
        return self._id * i

def main():
    with ThreadPool(5) as p:
        p.map(MyProcessor("localhost", 3001), range(15))

但现在_id将由所有线程共享:

current id: 58
current id: 58
current id: 58
current id: 58
current id: 58
current id: 58
current id: 58
current id: 58
current id: 58
current id: 58
current id: 58
current id: 58
current id: 58
current id: 58
current id: 58

concurrent.futures.ThreadPoolExecutor,我也尝试做同样的事情:

class MyProcessor():
    def __init__(self, host, port):
        # self.initsocket(host, port)
        self._id = random.randint(0, 100)

    def __call__(self, i):
        print("current id: {}".format(self._id))
        return self._id * i

def main():
    with ThreadPoolExecutor(max_workers=5) as executor:
        func = MyProcessor("localhost", 3001)
        futures = [executor.submit(func, i) for i in range(15)]
        for f in as_completed(futures):
            pass

输出如下:

current id: 94
current id: 94
current id: 94
current id: 94
current id: 94
current id: 94
current id: 94
current id: 94
current id: 94
current id: 94
current id: 94
current id: 94
current id: 94
current id: 94
current id: 94

当然,我得到这个结果并不奇怪,因为我只调用了__init__一次。但我想问的是:

如何使用concurrent.futures.ThreadPoolExecutormultiprocessing.pool.ThreadPool(也请不再使用全局变量)执行相同的操作。

推荐答案

这里有几个问题,我将尽最大努力解决所有这些问题。

在您给出的第一个示例中,您可以完全控制您创建的所有,因此每个线程在初始化式中都有一个唯一的ID。当然,问题是您一次启动所有线程,这对于大量线程来说可能是非常低效的。

在问题中的两个线程池示例中,您只为可调用对象初始化了一次ID,因此每个线程当然不会有单独的ID。正确的方法是在__call__方法中为每个线程初始化一个ID:

class MyProcessor():
    def __init__(self, host, port):
        self.initsocket(host, port)

    def __call__(self, i):
        id_ = random.randint(0, 100)
        print("current id: {}".format(id_))
        return id_ * i

def main():
    func = MyProcessor("localhost", 3001)
    with ThreadPoolExecutor(max_workers=5) as executor:
        collections.deque(executor.map(MyProcessor, range(15)), maxlen=0)
请注意,如果您只关心最终结果而不是中间的Future对象,那么您也可以通过使用map方法来缩短concurrent.futures.ThreadPoolExecutor示例。deque(..., maxlen=0)调用是使用迭代器的标准用法。

考虑到您在注释中链接的要点,我理解您为什么希望拥有线程本地数据。然而,您当然不需要全局变量来实现该结果。以下是几个替代方案:

  1. 只需将您的thread-local数据添加到初始化器中的self,所有调用都可以访问它,而不是全局的:

    def __init__(self, host, port):
        self.thread_local = threading.local()
    
    def __call__(self, i):
        try:
            id_ = self.thread_local.id_
        except AttributeError:
            id_ = random.randint(0, 100)
        ...
    
  2. 使用函数本地数据而不是线程本地数据。您使用的是线程本地数据,以避免将连接(在要点中)传递给一些私有函数。这并不是一种真正的需要,而是一种审美选择。您可以始终使用def _send_data(self, conn, **kwargs)def _recv_data(self, conn),因为连接实际来自的唯一位置是__call__

虽然可能存在选项#1的情况,但我强烈建议您不要将其用于任何类型的线程池管理器。线程池可以重复使用相同的线程,以便从任务提交到的队列中按顺序运行任务。这意味着您将在本应打开其自身的任务中获得相同的连接。在您最初的示例中,您可以独立创建所有线程,但当您在回收的池线程上有多个MyProcessor调用时,情况可能就不一样了。

这篇关于如何将一些变量绑定到具有conCurent.futures.ThreadPoolExecutor或MultiProcing.pool.ThreadPool的线程?的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持编程学习网!

本站部分内容来源互联网,如果有图片或者内容侵犯您的权益请联系我们删除!

相关文档推荐

Leetcode 234: Palindrome LinkedList(Leetcode 234:回文链接列表)
How do I read an Excel file directly from Dropbox#39;s API using pandas.read_excel()?(如何使用PANDAS.READ_EXCEL()直接从Dropbox的API读取Excel文件?)
subprocess.Popen tries to write to nonexistent pipe(子进程。打开尝试写入不存在的管道)
I want to realize Popen-code from Windows to Linux:(我想实现从Windows到Linux的POpen-code:)
Reading stdout from a subprocess in real time(实时读取子进程中的标准输出)
How to call type safely on a random file in Python?(如何在Python中安全地调用随机文件上的类型?)