0

我有一个脚本来读取一个可能有 10 多个演出的文件，我想使用多处理来处理它。

这是一种压缩算法，我希望用户定义一个缓冲区，然后将启动 3 个进程，一个从文件中读取缓冲区行数，将行传递给处理进程，然后将处理后的行传递给一个进程将行写入新文件。我希望所有这一切同时发生，并且每个进程都等待下一组行。

我已经有了脚本，但是当我运行它时，它并没有结束。我认为流程有问题。我认为这与我的 read 函数中的 islice 有关，但我不知道如何更好地编写它。

import multiprocessing as mp
import time
from itertools import islice

def read(from_filename, buffer, process_queue):
  file = open(from_filename, 'r')
  slice = islice(file, buffer)
  while slice:
    to_process = []
    for line in slice:
      to_process.append(line)
    process_queue.put(to_process)
  process_queue.put('kill')

def write(to_filename, write_queue):
  to_file = open(to_filename, 'a+')
  while 1:
    to_write = write_queue.get()
    if to_write == 'kill':
      break
    to_file.write(to_write + '\n')

def compress(process_queue, write_queue):
  while 1:
    to_process = process_queue.get()
    if to_process == 'kill':
      write_queue.put('kill')
      break
    # process, put output in to_write
    write_queue.put(to_write)

def decompress(process_queue, write_queue):
  while 1:
    to_process = process_queue.get()
    if to_process == 'kill':
      write_queue.put('kill')
      break
    # process, put output in to_write
    write_queue.put(to_write)

def main():
  option = raw_input("C for Compress OR D for Decompress: ")
  from_file = raw_input("Enter input filename: ")
  buf = int(raw_input("Enter line buffer: "))
  to_file = raw_input("Enter output filename: ")
  start = time.time()
  write_queue = mp.Queue()
  process_queue = mp.Queue()
  reader = mp.Process(target=read, args=(from_file, buf, process_queue))
  writer = mp.Process(target=write, args=(to_file, write_queue))
  if option == 'c' or option == 'C':
    processor = mp.Process(target=compress, args=(process_queue, write_queue))
  elif option == 'd' or option == 'D':
    processor = mp.Process(target=decompress, args=(process_queue, write_queue))
  else:
    print "Invalid Options..."
  writer.start()
  processor.start()
  reader.start()
  reader.join()
  processor.join()
  writer.join()
  end = time.time()
  elapsed = (end - start)
  print "\n\nTotal Time Elapsed: " + str(elapsed) + " secs"

if __name__=='__main__':
  main()

这是我第一次尝试多处理。当我运行它时，它并没有结束。我认为一个过程卡在某个地方。

4

1 回答 1

1

这段代码是错误的：

def read(from_filename, buffer, process_queue):
  file = open(from_filename, 'r')
  slice = islice(file, buffer)
  while slice:
    to_process = []
    for line in slice:
      to_process.append(line)
    process_queue.put(to_process)
  process_queue.put('kill')

由于slice是一个islice对象，因此条件while slice将始终为真，因此就像有一个while True存在。您应该每次都重新创建切片对象。

def read(from_filename, buffer, process_queue):
  file = open(from_filename, 'r')

  while True:
    slice = islice(file, buffer)
    to_process = []
    for line in slice:
      to_process.append(line)
    process_queue.put(to_process)
    if not to_process:
        # input ended
        break
  process_queue.put('kill')

或者你可以这样做：

def read_chunk(file, buffer):
    return [file.readline() for _ in xrange(buffer)]
    # or, "more" equivalent to using islice
    #return [line for i,line in itertools.izip(xrange(buffer), file)]

def read(from_filename, buffer, process_queue):
  file = open(from_filename, 'r')

  for to_process in iter(lambda: read_chunk(file, buffer), []):
    process_queue.put(to_process)
  process_queue.put('kill')

itertools.islice请注意，如果您无论如何都必须构建列表，那么使用它是没有意义的。

于 2013-07-16T06:04:56.927 回答