在Python中通过队列传输文件对象

Question

我有一个多处理任务，用于处理输入数据并将结果写入临时文件（以备后用）。但是，当我尝试通过队列将文件句柄传输到父进程时，它会失败（不会引发异常，但队列仍为空）。

import multiprocessing, tempfile 

def worker(i): 
    my_data_object = [] 
    my_tmp_file = tempfile.NamedTemporaryFile('wb') 
    my_tmp_file.write(bytes('Hello world #{}'.format(i), 'utf-8')) 
    my_tmp_file.seek(0) 
    queue.put(my_tmp_file) 

queue = multiprocessing.Queue() 

print('Writing...') 
proc = [] 
for i in range(16): 
    proc.append(multiprocessing.Process(target = worker, args = (i,))) 
    proc[i].start() 
for p in proc: 
    p.join() 

print('Reading...') 
my_strings = [] 
while True: 
    try: 
     tmp_file = queue.get_nowait() 
    except: 
     print('All data are read. Queue is now empty') 
     break 
    my_strings.append(tmp_file.read()) 
    tmp_file.close() 

print('Files content: ', my_strings) 
print('Successful termination') 

有没有人知道解决方案？

Answer 1

保持开放的文件似乎会造成问题，如果你打电话给你的工人函数读取和关闭它的工作原理后：

from multiprocessing import Process, Queue 

def worker(i,queue): 
    my_tmp_file = tempfile.NamedTemporaryFile() 
    my_tmp_file.write(bytes('Hello world #{}'.format(i), 'utf-8')) 
    my_tmp_file.seek(0) 
    queue.put(my_tmp_file.read()) 
    my_tmp_file.close() 

q = Queue() 

processes = [Process(target=worker, args=(i, q)) for i in range(16)] 

for p in processes: 
    p.start() 

for p in processes: 
    p.join() 

while q.qsize(): 
    out = q.get() 
    print(out) 

如果你试图关闭文件对象不读，你会得到一个TypeError: cannot serialize '_io.FileIO' object作为不可打开的_io.FileIO对象。

什么可能取决于你想要做的就是把.NAME队列和删除设置为False，并重新打开文件有什么帮助：

import multiprocessing, tempfile 

def worker(i): 
    with tempfile.NamedTemporaryFile(delete=False) as my_tmp_file: 
     my_tmp_file.write(bytes('Hello world #{}'.format(i), 'utf-8')) 
     my_tmp_file.seek(0) 
     queue.put(my_tmp_file.name) 

queue = multiprocessing.Queue() 

print('Writing...') 
proc = [] 
for i in range(16): 
    proc.append(multiprocessing.Process(target = worker, args = (i,))) 
    proc[i].start() 
for p in proc: 
    p.join() 

print('Reading...') 
my_strings = [] 
while True: 
    try: 
     tmp_file = queue.get_nowait() 
    except Exception as e: 
     print('All data are read. Queue is now empty') 
     break 
    with open(tmp_file) as f: 
     my_strings.append(f) 

但你仍然需要重新打开该文件，因此不能确定如果有任何好处将会发生什么。