令人尴尬的并行问题通常由三个基本部分组成：
- 读取输入数据（从文件、数据库、tcp 连接等）。
- 对输入数据运行计算，其中每个计算独立于任何其他计算。
- 将计算结果写入（到文件、数据库、tcp 连接等）。
我们可以在两个维度上并行化程序：
- 第 2 部分可以在多个内核上运行，因为每个计算都是独立的；处理顺序无关紧要。
- 每个部分都可以独立运行。第 1 部分可以将数据放入输入队列，第 2 部分可以从输入队列中拉出数据并将结果放入输出队列，第 3 部分可以从输出队列中拉出结果并将其写出。
这似乎是并发编程中最基本的模式，但我仍然在尝试解决它时迷失了方向，所以让我们编写一个规范的示例来说明如何使用 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:])
让我们使用这个程序并重写它以使用多处理来并行化上述三个部分。下面是这个新的并行程序的骨架，需要充实以解决注释中的部分：
#!/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:])
可以在 github 上找到这些代码，以及可以生成示例 CSV 文件以进行测试的另一段代码。
我将不胜感激您对并发专家如何解决此问题的任何见解。
以下是我在思考这个问题时遇到的一些问题。解决任何/所有问题的奖励积分：
- 我应该有子进程来读取数据并将其放入队列中，还是主进程可以在读取所有输入之前不阻塞地执行此操作？
- 同样，我应该有一个子进程来将结果从已处理队列中写出来，还是主进程可以这样做而不必等待所有结果？
- 我应该使用进程池进行求和操作吗？
- 如果是，我应该在池上调用什么方法来让它开始处理进入输入队列的结果，而不阻塞输入和输出进程？应用异步（）？地图异步（）？地图（）？imap_unordered()？
- 假设我们不需要在数据进入时从输入和输出队列中抽出，而是可以等到所有输入都被解析并计算出所有结果（例如，因为我们知道所有输入和输出都将适合系统内存）。我们是否应该以任何方式更改算法（例如，不要在 I/O 的同时运行任何进程）？