使用Python多进程解决尴尬的并行问题

Question

如何使用 multiprocessing 来解决容易并行的问题呢？

容易并行的问题通常包含三个基本部分：

读取输入数据（可以是文件、数据库、TCP连接等）。
对输入数据进行计算，每个计算都是 相互独立的。
写入计算结果（到文件、数据库、TCP连接等）。

我们可以从两个方面来实现程序的并行：

第二部分可以在多个核心上运行，因为每个计算都是独立的；处理的顺序并不重要。
每个部分可以独立运行。第一部分可以把数据放到输入队列，第二部分可以从输入队列中取出数据并把结果放到输出队列，第三部分可以从输出队列中取出结果并写出。

这看起来是并发编程中最基本的模式，但我还是有点迷茫，所以让我们写一个经典的例子来说明如何使用 multiprocessing 来实现这个。

这里是一个例子问题：给定一个 CSV文件，里面有整数的行，计算它们的总和。将这个问题分成三个部分，这三部分都可以并行运行：

处理输入文件，得到原始数据（整数的列表/可迭代对象）
并行计算数据的总和
输出总和

下面是一个传统的单进程Python程序，它解决了这三个任务：

#!/usr/bin/env python
# -*- coding: UTF-8 -*-
# basicsums.py
"""A program that reads integer values from a CSV file and writes out their
sums to another CSV file.
"""

import csv
import optparse
import sys

def make_cli_parser():
    """Make the command line interface parser."""
    usage = "\n\n".join(["python %prog INPUT_CSV OUTPUT_CSV",
            __doc__,
            """
ARGUMENTS:
    INPUT_CSV: an input CSV file with rows of numbers
    OUTPUT_CSV: an output file that will contain the sums\
"""])
    cli_parser = optparse.OptionParser(usage)
    return cli_parser


def parse_input_csv(csvfile):
    """Parses the input CSV and yields tuples with the index of the row
    as the first element, and the integers of the row as the second
    element.

    The index is zero-index based.

    :Parameters:
    - `csvfile`: a `csv.reader` instance

    """
    for i, row in enumerate(csvfile):
        row = [int(entry) for entry in row]
        yield i, row


def sum_rows(rows):
    """Yields a tuple with the index of each input list of integers
    as the first element, and the sum of the list of integers as the
    second element.

    The index is zero-index based.

    :Parameters:
    - `rows`: an iterable of tuples, with the index of the original row
      as the first element, and a list of integers as the second element

    """
    for i, row in rows:
        yield i, sum(row)


def write_results(csvfile, results):
    """Writes a series of results to an outfile, where the first column
    is the index of the original row of data, and the second column is
    the result of the calculation.

    The index is zero-index based.

    :Parameters:
    - `csvfile`: a `csv.writer` instance to which to write results
    - `results`: an iterable of tuples, with the index (zero-based) of
      the original row as the first element, and the calculated result
      from that row as the second element

    """
    for result_row in results:
        csvfile.writerow(result_row)


def main(argv):
    cli_parser = make_cli_parser()
    opts, args = cli_parser.parse_args(argv)
    if len(args) != 2:
        cli_parser.error("Please provide an input file and output file.")
    infile = open(args[0])
    in_csvfile = csv.reader(infile)
    outfile = open(args[1], 'w')
    out_csvfile = csv.writer(outfile)
    # gets an iterable of rows that's not yet evaluated
    input_rows = parse_input_csv(in_csvfile)
    # sends the rows iterable to sum_rows() for results iterable, but
    # still not evaluated
    result_rows = sum_rows(input_rows)
    # finally evaluation takes place as a chain in write_results()
    write_results(out_csvfile, result_rows)
    infile.close()
    outfile.close()


if __name__ == '__main__':
    main(sys.argv[1:])

现在我们来把这个程序重写，使用 multiprocessing 来并行处理上面提到的三个部分。下面是这个新并行程序的框架，需要根据注释来完善：

#!/usr/bin/env python
# -*- coding: UTF-8 -*-
# multiproc_sums.py
"""A program that reads integer values from a CSV file and writes out their
sums to another CSV file, using multiple processes if desired.
"""

import csv
import multiprocessing
import optparse
import sys

NUM_PROCS = multiprocessing.cpu_count()

def make_cli_parser():
    """Make the command line interface parser."""
    usage = "\n\n".join(["python %prog INPUT_CSV OUTPUT_CSV",
            __doc__,
            """
ARGUMENTS:
    INPUT_CSV: an input CSV file with rows of numbers
    OUTPUT_CSV: an output file that will contain the sums\
"""])
    cli_parser = optparse.OptionParser(usage)
    cli_parser.add_option('-n', '--numprocs', type='int',
            default=NUM_PROCS,
            help="Number of processes to launch [DEFAULT: %default]")
    return cli_parser


def main(argv):
    cli_parser = make_cli_parser()
    opts, args = cli_parser.parse_args(argv)
    if len(args) != 2:
        cli_parser.error("Please provide an input file and output file.")
    infile = open(args[0])
    in_csvfile = csv.reader(infile)
    outfile = open(args[1], 'w')
    out_csvfile = csv.writer(outfile)

    # Parse the input file and add the parsed data to a queue for
    # processing, possibly chunking to decrease communication between
    # processes.

    # Process the parsed data as soon as any (chunks) appear on the
    # queue, using as many processes as allotted by the user
    # (opts.numprocs); place results on a queue for output.
    #
    # Terminate processes when the parser stops putting data in the
    # input queue.

    # Write the results to disk as soon as they appear on the output
    # queue.

    # Ensure all child processes have terminated.

    # Clean up files.
    infile.close()
    outfile.close()


if __name__ == '__main__':
    main(sys.argv[1:])

这些代码片段，以及另一段可以生成示例CSV文件的代码，可以在 github上找到。

如果你们这些并发编程的高手能给我一些建议，我会非常感激。

在思考这个问题时，我有一些疑问。 如果能解决任何一个或多个问题，我会很感激：

我应该为读取数据并放入队列创建子进程，还是主进程可以在不阻塞的情况下完成这项工作，直到所有输入都读取完？
同样，我应该为从处理队列中写出结果创建子进程，还是主进程可以在不等待所有结果的情况下完成这项工作？
我应该为求和操作使用进程池吗？
- 如果是，我该调用进程池的哪个方法来开始处理输入队列中的结果，而不阻塞输入和输出进程？ apply_async()? map_async()? imap()? imap_unordered()?
假设我们不需要在数据进入输入和输出队列时就将其分流，而是可以等到所有输入都解析完、所有结果都计算完（例如，因为我们知道所有输入和输出都能适应系统内存）。我们应该以任何方式改变算法吗（例如，不与I/O并发运行任何进程）？

并发编程数据处理并行计算多进程进程池输入输出队列计算总和易并行问题

使用Python多进程解决尴尬的并行问题

5 个回答

撰写回答