我想使用 concurrent.futures 启用我的程序的并行处理/线程化模块。

不幸的是，我似乎找不到任何使用 concurrent.futures 模块的漂亮、简单、防白痴的例子。他们通常需要更高级的 Python 知识或处理/线程概念和行话。

下面是一个基于我的程序的简化的、独立的示例:有一个纯粹的 CPU 绑定(bind)任务非常适合多进程，还有一个单独的 IO 绑定(bind)任务插入数据库 (SQLite)。 在我的程序中，我已经将其转换为使用多处理池类，但由于 CPU 绑定(bind)任务的结果全部收集起来等待任务完成，因此它使用了大量内存。 因此，我希望结合使用线程/处理，我相信 concurrent.futures 可以为我做的相当简单。

那么我如何将下面的内容转换成使用这个模块的东西呢？

import sqlite3

#Stand in CPU intensive task
def calculate(value):
    return value * 10

#Stand in Thread I/O intensive task
def output(value):
    global db

    if (value % 1000) == 0:
        db.execute('delete from test_table')

    db.execute('insert into test_table (result) values (?)', (value,))

def main():
    global db
    results = []

    db  = sqlite3.connect('e:\\z_dev\\test.sqlite')
    db.cursor()

    #=========
    #Perform CPU intensive task
    for i in range(1000):
        results.append( calculate(i))

    #Perform Threading intensive task
    for a in results:
        output(a)
    #=========

    db.commit()
    db.close()

if __name__ == '__main__':
    main()

我正在寻找一个不使用任何花哨/复杂 python 的答案。或者一个很好的清晰简单的解释，或者最好是两者兼而有之!

谢谢

编辑:我当前的“多处理器”实现。可能是错误的，但它似乎有效。没有任何线程。这进入上面的“#=========”部分。

#Multiprocessing
pool = multiprocessing.Pool(None)
for i in range(1000):
    results.append( pool.apply_async(calculate(i)))
pool.close()
pool.join()

for i in results:
    results[i] = results[i].get()

#Complete lack of threading; but if I had it, it'd be here:     
for a in results:
    output(a)

最佳答案

concurrent.futures 具有简约的 API。它很容易用于非常简单的问题，但你没有一个非常简单的问题。如果你这样做了，你就已经解决了它 ;-)

您没有显示您编写的任何 multiprocessing.Pool 代码，但那将是一个更有希望的起点 - 假设您更想解决问题而不是确认您的问题希望如果你只切换到一个较弱的 API，它一定很容易做到；-)

继续使用多处理 的“明显”方法是使用Pool.apply_async() 方法，将异步结果对象放在有界队列中。 Queue，并让主程序中的线程将它们从 Queue 中拉出并等待结果显示。这很容易，但这并不神奇。它解决了您的问题，因为有界 队列 是在以不同速度运行的生产者和消费者之间进行调解的 规范方式。 concurrent.futures 中没有任何内容直接解决那个问题，它是“大量内存”问题的核心。

# Define global result_queue only in the main program.
import Queue
result_queue = Queue.Queue(100)  # pick a reasonable max size based on your problem

# Run this in as many threads as you like.
def consume_results():
    while True:
        a = result_queue.get()
        if a is None:
            break
        output(a.get())  # `output()` is your function

...
# main program passes out work, after starting threads
for i in range(1000):
    # the .put() will block so long as the queue is at its max size
    result_queue.put(pool.apply_async(calculate, args=(i,)))
# add sentinels to let threads know they're done
for i in range(number_of_threads_you_started):
    result_queue.put(None)

这就是您需要让生产者和消费者大致保持平衡的事情，而且任何标准库中都没有任何东西可以神奇地为您做到这一点。

编辑 - 充实它

这是一个完整的可执行示例，任何使用 Python3 的人都可以运行。备注:

• 它不使用您的代码片段，因为那些依赖于外部数据库模块的代码并不是每个人都能运行的。
• 它坚持使用 concurrent.futures 来管理进程和线程。使用 multiprocessing 和 threading 并不难，实际上这里使用了线程的方式，使用 会更容易一些直接线程。但是这种方式已经足够清楚了。
• concurrent.futures Future 对象基本上与 multiprocessing 异步结果对象相同 - API 功能只是拼写不同。
• 你的问题并不简单，因为它有多个阶段，可以以不同的速度运行。同样，任何标准库中的任何内容都无法神奇地隐藏其潜在的不良后果。创建自己的有界队列仍然是最好的解决方案。对于任何合理的 MAX_QUEUE_SIZE 值，此处的内存使用将保持适度。
• 您通常不希望创建比您可用的内核数少一个的受 CPU 限制的工作进程。主程序也需要循环运行，操作系统也是如此。
• 一旦你习惯了这些东西，这段代码中的所有注释都会很烦人，就像在代码行 i += 1 上看到注释“increment by 1” ;-)

代码如下:

import concurrent.futures as cf
import threading
import queue

NUM_CPUS = 3
NUM_THREADS = 4
MAX_QUEUE_SIZE = 20

# Runs in worker processes.
def producer(i):
    return i + 10

def consumer(i):
    global total
    # We need to protect this with a lock because
    # multiple threads in the main program can
    # execute this function simultaneously.
    with sumlock:
        total += i

# Runs in threads in main program.
def consume_results(q):
    while True:
        future = q.get()
        if future is None:
            break
        else:
            consumer(future.result())

if __name__ == "__main__":
    sumlock = threading.Lock()
    result_queue = queue.Queue(MAX_QUEUE_SIZE)
    total = 0
    NUM_TO_DO = 1000
    with cf.ThreadPoolExecutor(NUM_THREADS) as tp:
        # start the threads running `consume_results`
        for _ in range(NUM_THREADS):
            tp.submit(consume_results, result_queue)
        # start the worker processes
        with cf.ProcessPoolExecutor(NUM_CPUS) as pp:
            for i in range(NUM_TO_DO):
                # blocks until the queue size <= MAX_QUEUE_SIZE
                result_queue.put(pp.submit(producer, i))
        # tell threads we're done
        for _ in range(NUM_THREADS):
            result_queue.put(None)
    print("got", total, "expected", (10 + NUM_TO_DO + 9) * NUM_TO_DO // 2)

如果一切顺利，这是预期的输出:

got 509500 expected 509500

关于python - Concurrent.futures 使用指南——同时使用线程和处理的简单示例，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/20783337/