Xử lý Python bằng cách sử dụng Hàng đợi để ghi vào cùng một tệp

Question

Tôi biết có nhiều bài đăng trên Stack Exchange liên quan đến việc viết kết quả từ đa xử lý tới tệp đơn và tôi đã phát triển mã của mình sau khi chỉ đọc những bài đăng đó. Những gì tôi đang cố gắng để đạt được là chạy 'RevMapCoord' chức năng song song và viết kết quả của nó trong một tập tin duy nhất bằng cách sử dụng multiprocess.queue. Nhưng tôi đang gặp vấn đề trong khi xếp hàng công việc của mình. Mã của tôi:

def RevMapCoord(list): 
    "Read a file, Find String and Do something" 

def feed(queue, parlist): 
    for par in parlist: 
     print ('Echo from Feeder: %s' % (par)) 
     queue.put(par) 
    print ('**Feeder finished queing**') 

def calc(queueIn, queueOut): 
    print ('Worker function started') 
    while True: 
     try: 
      par = queueIn.get(block = False) 
      res = RevMapCoord(final_res) 
      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() 


feedProc = Process(target = feed , args = (workerQueue, final_res)) 
calcProc = [Process(target = calc , args = (workerQueue, writerQueue)) for i in range(nproc)] 
writProc = Process(target = write, args = (writerQueue, sco_inp_extend_geno)) 

feedProc.start() 
print ('Feeder is joining') 
feedProc.join() 
for p in calcProc: 
    p.start() 
for p in calcProc: 
    p.join() 
writProc.start() 
writProc.join() 

Khi tôi chạy tập lệnh mã này sẽ dừng ở bước "feedProc.start()". Một vài dòng đầu ra cuối cùng từ màn hình hiển thị tuyên bố in từ cuối "feedProc.start()":

Echo from Feeder: >AK779,AT61680,50948-50968,50959,6,0.406808,Ashley,Dayne 
Echo from Feeder: >AK832,AT30210,1091-1111,1102,7,0.178616,John,Caine 
**Feeder finished queing** 

Nhưng treo trước khi thực hiện dòng tiếp theo "feedProc.join()". Mã không có lỗi và tiếp tục chạy nhưng không làm gì cả (bị treo). Vui lòng cho tôi biết tôi đang mắc lỗi gì.

Answer 1

tôi đã đạt được bằng văn bản kết quả.. từ đa xử vào một tập tin duy nhất bằng cách uing chức năng 'map_async' trong Python3 Dưới đây là chức năng tôi đã viết:.

def PPResults(module,alist):##Parallel processing 
    npool = Pool(int(nproc))  
    res = npool.map_async(module, alist) 
    results = (res.get())###results returned in form of a list 
    return results 

Vì vậy, tôi cung cấp chức năng này với một danh sách các thông số trong 'a_list' và 'module' là một chức năng mà không xử lý và trả về kết quả.Các chức năng trên giữ trên thu thập các kết quả dưới dạng danh sách và trả về khi tất cả các đoạn mét từ 'a_list' đã được xử lý. Kết quả có thể không đúng thứ tự nhưng theo thứ tự không quan trọng đối với tôi, điều này có hiệu quả. Danh sách 'kết quả' có thể được lặp và kết quả cá nhân bằng văn bản trong tập tin như:

fh_out = open('./TestResults', 'w') 
for i in results:##Write Results from list to file 
    fh_out.write(i) 

Để giữ thứ tự của các kết quả chúng ta có thể cần phải sử dụng 'hàng đợi' tương tự như tôi đã đề cập trong câu hỏi của tôi (ở trên). Mặc dù tôi có thể sửa mã nhưng tôi tin rằng nó không bắt buộc phải được đề cập ở đây.

Cảm ơn

AK

Answer 2

Tôi nghĩ bạn nên làm mỏng ví dụ của mình về cơ bản. Ví dụ:

from multiprocessing import Process, Queue 

def f(q): 
    q.put('Hello') 
    q.put('Bye') 
    q.put(None) 

if __name__ == '__main__': 
    q = Queue() 
    p = Process(target=f, args=(q,)) 
    p.start() 
    with open('file.txt', 'w') as fp: 
     while True: 
      item = q.get() 
      print(item) 
      if item is None: 
       break 
      fp.write(item) 
    p.join() 

Ở đây tôi có hai quy trình (quy trình chính, p). p đặt chuỗi trong một hàng đợi được truy xuất bởi quá trình chính. Khi quá trình chính thấy None (một trọng điểm mà tôi đang sử dụng để chỉ ra: "Tôi đang thực hiện" nó phá vỡ vòng lặp

Mở rộng này để nhiều quá trình (hoặc chủ đề) là tầm thường