0

我有一个包含数十万行的文件，每一行都需要经历相同的过程（计算协方差）。我打算使用多线程，因为它需要很长时间。然而，我看到的所有示例/教程对于我想做的事情来说都相当复杂。如果有人能给我指出一个很好的教程来解释如何一起使用这两个模块，那就太好了。

4

1 回答 1

0

每当我必须并行处理某些东西时，我都会使用类似的东西（我只是从现有脚本中删除了它）：

#!/usr/bin/env python2
# This Python file uses the following encoding: utf-8

import os, sys, time
from multiprocessing import Queue, Manager, Process, Value, Event, cpu_count

class ThreadedProcessor(object):
  def __init__(self, parser, input_file, output_file, threads=cpu_count()):
    self.parser = parser

    self.num_processes = threads
    self.input_file = input_file
    self.output_file = output_file

    self.shared_proxy = Manager()

    self.input_queue = Queue()
    self.output_queue = Queue()

    self.input_process = Process(target=self.parse_input)
    self.output_process = Process(target=self.write_output)

    self.processes = [Process(target=self.process_row) for i in range(self.num_processes)]

    self.input_process.start()
    self.output_process.start()

    for process in self.processes:
      process.start()

    self.input_process.join()

    for process in self.processes:
      process.join()

    self.output_process.join()

  def parse_input(self):
    for index, row in enumerate(self.input_file):
      self.input_queue.put([index, row])

    for i in range(self.num_processes):
      self.input_queue.put('STOP')

  def process_row(self):
    for index, row in iter(self.input_queue.get, 'STOP'):
      self.output_queue.put([index, row[0], self.parser.parse(row[1])])

    self.output_queue.put('STOP')

  def write_output(self):
    current = 0
    buffer = {}

    for works in range(self.num_processes):
      for index, id, row in iter(self.output_queue.get, 'STOP'):
        if index != current:
          buffer[index] = [id] + row
        else:
          self.output_file.writerow([id] + row)
          current += 1

          while current in buffer:
            self.output_file.writerow(buffer[current])
            del buffer[current]
            current += 1

基本上，您有两个进程管理文件的读/写。一个读取并解析输入，另一个从“完成”队列中读取并写入您的输出文件。产生了其他进程（在这种情况下，数量等于您的 CPU 拥有的总处理器内核数）并且它们都处理来自输入队列的元素。

于 2012-05-29T07:16:16.560 回答