python - 使用多处理写入文件

Question

我在 python 中遇到以下问题。

我需要并行进行一些计算，我需要将其结果按顺序写入文件中。所以我创建了一个接收 multiprocessing.Queue 和文件句柄的函数，进行计算并将结果打印到文件中:

import multiprocessing
from multiprocessing import Process, Queue
from mySimulation import doCalculation   

# doCalculation(pars) is a function I must run for many different sets of parameters and collect the results in a file

def work(queue, fh):
while True:
    try:
        parameter = queue.get(block = False)
        result = doCalculation(parameter) 
        print >>fh, string
    except:
        break


if __name__ == "__main__":
    nthreads = multiprocessing.cpu_count()
    fh = open("foo", "w")
    workQueue = Queue()
    parList = # list of conditions for which I want to run doCalculation()
    for x in parList:
        workQueue.put(x)
    processes = [Process(target = writefh, args = (workQueue, fh)) for i in range(nthreads)]
    for p in processes:
       p.start()
    for p in processes:
       p.join()
    fh.close()

但是脚本运行后文件最终为空。我试图将 worker() 函数更改为:

def work(queue, filename):
while True:
    try:
        fh = open(filename, "a")
        parameter = queue.get(block = False)
        result = doCalculation(parameter) 
        print >>fh, string
        fh.close()
    except:
        break

并将文件名作为参数传递。然后它按我的预期工作。当我尝试按顺序做同样的事情时，没有多处理，它也能正常工作。

为什么它在第一个版本中不起作用？我看不出问题所在。

另外:我能保证两个进程不会同时尝试写入文件吗？

---

编辑:

谢谢。我现在明白了。这是工作版本:

import multiprocessing
from multiprocessing import Process, Queue
from time import sleep
from random import uniform

def doCalculation(par):
    t = uniform(0,2)
    sleep(t)
    return par * par  # just to simulate some calculation

def feed(queue, parlist):
    for par in parlist:
            queue.put(par)

def calc(queueIn, queueOut):
    while True:
        try:
            par = queueIn.get(block = False)
            print "dealing with ", par, "" 
            res = doCalculation(par)
            queueOut.put((par,res))
        except:
            break

def write(queue, fname):
    fhandle = open(fname, "w")
    while True:
        try:
            par, res = queue.get(block = False)
            print >>fhandle, par, res
        except:
            break
    fhandle.close()

if __name__ == "__main__":
    nthreads = multiprocessing.cpu_count()
    fname = "foo"
    workerQueue = Queue()
    writerQueue = Queue()
    parlist = [1,2,3,4,5,6,7,8,9,10]
    feedProc = Process(target = feed , args = (workerQueue, parlist))
    calcProc = [Process(target = calc , args = (workerQueue, writerQueue)) for i in range(nthreads)]
    writProc = Process(target = write, args = (writerQueue, fname))


    feedProc.start()
    for p in calcProc:
        p.start()
    writProc.start()

    feedProc.join ()
    for p in calcProc:
        p.join()
    writProc.join ()

Answer 1

您确实应该使用两个队列和三种不同的处理方式。

1. 将内容放入队列 #1。

2. 从队列 #1 中取出内容并进行计算，将内容放入队列 #2。你可以有很多这样的东西，因为它们来自一个队列并安全地放入另一个队列。

3. 从队列 #2 中取出内容并将其写入文件。您必须恰好拥有其中的 1 个，不能更多。它“拥有”该文件，保证原子访问，并绝对确保该文件被干净且一致地写入。