蟒蛇 - 附加到同一個文件從多個線程

Question

我正在寫一個應用程序，將行附加到來自多個線程的同一個文件。

我有一個問題，其中一些行被添加而沒有新行。

任何解決方案？

class PathThread(threading.Thread): 
    def __init__(self, queue): 
     threading.Thread.__init__(self) 
     self.queue = queue 

    def printfiles(self, p): 
     for path, dirs, files in os.walk(p): 
      for f in files: 
       print(f, file=output) 

    def run(self): 
     while True: 
      path = self.queue.get() 
      self.printfiles(path) 
      self.queue.task_done() 


pathqueue = Queue.Queue() 
paths = getThisFromSomeWhere() 

output = codecs.open('file', 'a') 

# spawn threads 
for i in range(0, 5): 
    t = PathThread(pathqueue) 
    t.setDaemon(True) 
    t.start() 

# add paths to queue 
for path in paths: 
    pathqueue.put(path) 

# wait for queue to get empty 
pathqueue.join() 

Answer 1

解決方法是僅在一個線程中寫入文件。

import Queue # or queue in Python 3 
import threading 

class PrintThread(threading.Thread): 
    def __init__(self, queue): 
     threading.Thread.__init__(self) 
     self.queue = queue 

    def printfiles(self, p): 
     for path, dirs, files in os.walk(p): 
      for f in files: 
       print(f, file=output) 

    def run(self): 
     while True: 
      result = self.queue.get() 
      self.printfiles(result) 
      self.queue.task_done() 

class ProcessThread(threading.Thread): 
    def __init__(self, in_queue, out_queue): 
     threading.Thread.__init__(self) 
     self.in_queue = in_queue 
     self.out_queue = out_queue 

    def run(self): 
     while True: 
      path = self.in_queue.get() 
      result = self.process(path) 
      self.out_queue.put(result) 
      self.in_queue.task_done() 

    def process(self, path): 
     # Do the processing job here 

pathqueue = Queue.Queue() 
resultqueue = Queue.Queue() 
paths = getThisFromSomeWhere() 

output = codecs.open('file', 'a') 

# spawn threads to process 
for i in range(0, 5): 
    t = ProcessThread(pathqueue, resultqueue) 
    t.setDaemon(True) 
    t.start() 

# spawn threads to print 
t = PrintThread(resultqueue) 
t.setDaemon(True) 
t.start() 

# add paths to queue 
for path in paths: 
    pathqueue.put(path) 

# wait for queue to get empty 
pathqueue.join() 
resultqueue.join() 

Answer 2

也許一些更多的換行符，他們不應該是？ 您應該記住共享資源不應該一次被多個線程訪問，否則不可預知的後果可能會發生。 （它在使用線程時使用'原子操作'） 看一看這個頁面有一點直覺。
Thread-Synchronization

Answer 3

的事實，你永遠不會看到在一條線上的中間同一線路或新線路雜亂的文字，你實際上不需要到syncronize附加到文件的線索。問題是您使用print來寫入單個文件句柄。我懷疑print實際上是在一次調用中對文件句柄進行2次操作，並且這些操作正在線程之間競爭。基本上print正在做這樣的事情：

file_handle.write('whatever_text_you_pass_it') 
file_handle.write(os.linesep) 

而且由於不同線程對同一文件句柄同時這樣做有時一個線程將在第一時間拿到寫，然後其他線程將獲得在其第一次寫，然後你將連續獲得兩個回車。或者真的是這些的任何排列。

解決此問題的最簡單方法是停止使用print並直接使用write。嘗試這樣的：

output.write(f + os.linesep) 

這對我仍然是危險的。我不知道你可以期待什麼gaurantees與所有線程使用相同的文件句柄對象和競爭其內部緩衝區。個人身份證方面的步驟整個問題，並讓每個線程獲得自己的文件句柄。還要注意，這是可行的，因爲寫入緩衝區刷新的默認值是行緩衝的，所以當它對文件進行刷新時，它會以os.linesep結束。強制它使用行緩衝發送1作爲open的第三個參數。你可以測試出來是這樣的：

#!/usr/bin/env python 
import os 
import sys 
import threading 

def hello(file_name, message, count): 
    with open(file_name, 'a', 1) as f: 
    for i in range(0, count): 
     f.write(message + os.linesep) 

if __name__ == '__main__': 
    #start a file 
    with open('some.txt', 'w') as f: 
    f.write('this is the beginning' + os.linesep) 
    #make 10 threads write a million lines to the same file at the same time 
    threads = [] 
    for i in range(0, 10): 
    threads.append(threading.Thread(target=hello, args=('some.txt', 'hey im thread %d' % i, 1000000))) 
    threads[-1].start() 
    for t in threads: 
    t.join() 
    #check what the heck the file had 
    uniq_lines = set() 
    with open('some.txt', 'r') as f: 
    for l in f: 
     uniq_lines.add(l) 
    for u in uniq_lines: 
    sys.stdout.write(u) 

輸出看起來是這樣的：

hey im thread 6 
hey im thread 7 
hey im thread 9 
hey im thread 8 
hey im thread 3 
this is the beginning 
hey im thread 5 
hey im thread 4 
hey im thread 1 
hey im thread 0 
hey im thread 2