问题标签 [concurrent.futures]

我有一个直播concurrent.futures.ThreadPoolExecutor。我想检查它的状态。我想知道有多少线程，有多少正在处理任务，有哪些任务，有多少是空闲的，哪些任务在队列中。我怎样才能找到这些东西？

我刚开始使用concurrent.futuresPython 3 中的库将一些函数应用于图像列表，以便处理这些图像并重塑它们。函数是resize(height, width)和opacity(number)。

另一方面，我有images()产生类似文件的对象的功能，所以我尝试了这段代码来并行处理我的图像：

有人能告诉我我是否走在正确的道路上来实现这一目标吗？

我正在尝试使用 Yahoo Finance 的多个线程下载大量数据。我concurrent.futures.ThreadPoolExecutor用来加快速度。一切都很顺利，直到我使用了所有可用的文件描述符（默认为 1024）。

当urllib.request.urlopen()引发异常时，文件描述符保持打开状态（无论我使用的套接字超时）。通常，如果我仅从单个（主）线程运行内容，则会重用此文件描述符，因此不会发生此问题。但是，当这些异常urlopen()调用从ThreadPoolExecutor线程进行时，这些文件描述符保持打开状态。到目前为止，我想出的唯一解决方案是使用ProcessPoolExecutor非常繁琐且效率低下的进程（ ）。必须有更聪明的方法来处理这个问题。

而且我想知道这是否是 Python 库中的一个错误，或者我只是做错了什么......

我在 Debian 上运行 Python 3.4.1（测试，内核 3.10-3-amd64）。

这是演示此行为的示例代码：

我正在使用 Python 2.7。

我目前正在使用这样的 ThreadPoolExecuter：

问题是f有时运行时间过长。每当我运行f时，我想将其运行时间限制为 100 秒，然后将其杀死。

最终，对于 中的每个元素x，param我想知道是否f必须被杀死，如果不是 - 返回值是多少。即使f一个参数超时，我仍然想用下一个参数运行它。

该executer.map方法确实有一个timeout参数，但它为整个运行设置了超时，从调用到 的时间executer.map，而不是单独为每个线程设置超时。

获得所需行为的最简单方法是什么？

在 ThreadPoolExecutor (TPE) 中，回调是否总是保证与提交的函数在同一个线程中运行？

例如，我使用以下代码对此进行了测试。我运行了很多次，它似乎func并且callback总是在同一个线程中运行。

但是，当我删除语句时，它似乎失败了time.sleep(random.random())，即至少有几个func函数并且callbacks 没有在同一个线程中运行。

对于我正在处理的项目，回调必须始终与提交的函数在同一线程上运行，因此我想确保 TPE 能保证这一点。（而且没有随机睡眠的测试结果也令人费解）。

我查看了执行程序的源代码，似乎我们在运行回调之前没有将线程切换到主线程。但只是想确定一下。

我有一些利用 ctypes.CDLL 的 Python 代码，根据文档，这不涉及 gil。话虽如此，我在分析时遇到了一些我不清楚的瓶颈。如果我使用 time.sleep 甚至 ctypes.windll.kernel32.Sleep 运行一些琐碎的代码，我可以看到时间比例，因为线程数与任务数相匹配，换句话说，如果任务要休眠 1 秒并且我在 1 个线程中提交 1 个任务或在 20 个线程中提交 20 个任务，它们都需要大约 1 秒才能完成。

切换回我的代码，它没有按预期扩展，而是线性扩展。分析表明 _thread.lock 中的 acquire() 等待。

有哪些技术可以进一步深入研究以了解问题在哪里体现？ThreadPoolExecutor 不是这里的最佳选择吗？我知道它实现了一个基本的线程池，与 multiprocessing.pool 中的 ThreadPool 没有什么不同？

我正在尝试运行多个进程，同时使用concurrent.futures.ProcessPoolExecutor来运行 CPU 密集型作业。前几个请求得到了愉快的服务，但随后KeyError从 引发了a concurrent.futures.process，并且服务器挂起。

这是 Tornado 中的错误吗？

这是我剥离代码的最简单形式。

服务器：

客户：

错误回溯

我有一个 python 脚本，在我的文件顶部：

我了解到队列是过程安全的数据结构。我有一个方法：

这里的perform_scan.delay()函数实际上调用了一个 celery worker，但我认为不相关（它是一个异步进程调用）。

我也有：

最后我有一个main()功能：

print 语句的结果是 0。但是，如果我包含一个大小为result_queuein的 print 语句enqueue_tasks，当程序运行时，我可以看到大小正在增加，并且正在将内容添加到队列中。

对正在发生的事情的想法？

我正在为几组迭代运行一个函数，一旦所有进程完成，就会返回所有结果的列表。

如何在计算结果后立即打印中间结果，例如变量 1 = 1、变量 2 = 7？

在 Python文档中，我看到：

concurrent.futures.Future... ...除测试外，不应直接创建。

我想在我的代码中使用它作为一个承诺，我很惊讶不建议这样使用它。

我的用例：
我有一个线程来读取来自套接字的数据包，并且我有许多回调，这些回调取决于数据包中包含的一些信息。数据包是对消费者请求的响应，所有消费者使用单一连接。每个消费者都会收到一个 Promise 并向其添加一些处理程序，这些处理程序在响应到达时被调用。

所以我不能Executor在这里使用子类，因为我只有一个线程，但我需要创建许多 Futures（承诺）。

Promise 是一种非常普遍的编程技术，我认为这Future是 Python 的 Promise 实现。但是如果不建议像promise那样使用，那么pythonistas常用的有哪些呢？

笔记

我使用 Python 2.7向后移植concurrent.futures到 2.7