Làm thế nào để bạn chuyển một tham chiếu Queue đến một hàm được quản lý bởi pool.map_async()?

Question

Tôi muốn một quy trình chạy dài để trả lại tiến trình của nó trên một Hàng đợi (hoặc một thứ gì đó tương tự) mà tôi sẽ chuyển đến hộp thoại thanh tiến trình. Tôi cũng cần kết quả khi quá trình hoàn tất. Ví dụ kiểm tra ở đây không thành công với số RuntimeError: Queue objects should only be shared between processes through inheritance.

import multiprocessing, time 

def task(args): 
    count = args[0] 
    queue = args[1] 
    for i in xrange(count): 
     queue.put("%d mississippi" % i) 
    return "Done" 

def main(): 
    q = multiprocessing.Queue() 
    pool = multiprocessing.Pool() 
    result = pool.map_async(task, [(x, q) for x in range(10)]) 
    time.sleep(1) 
    while not q.empty(): 
     print q.get() 
    print result.get() 

if __name__ == "__main__": 
    main() 

tôi đã có thể để có được điều này để làm việc sử dụng đối tượng Process cá nhân (nơi tôi đang alowed để vượt qua một tham chiếu Queue) nhưng sau đó tôi không có một hồ bơi để quản lý các quá trình nhiều Tôi muốn để khởi chạy. Bất kỳ tư vấn về một mô hình tốt hơn cho điều này?

Answer 1

Các mã sau đây dường như làm việc:

import multiprocessing, time 

def task(args): 
    count = args[0] 
    queue = args[1] 
    for i in xrange(count): 
     queue.put("%d mississippi" % i) 
    return "Done" 


def main(): 
    manager = multiprocessing.Manager() 
    q = manager.Queue() 
    pool = multiprocessing.Pool() 
    result = pool.map_async(task, [(x, q) for x in range(10)]) 
    time.sleep(1) 
    while not q.empty(): 
     print q.get() 
    print result.get() 

if __name__ == "__main__": 
    main() 

Lưu ý rằng Queue được nhận được từ một manager.Queue() chứ không phải là multiprocessing.Queue(). Cảm ơn Alex đã chỉ cho tôi theo hướng này.

Answer 2

Làm qtoàn cầu công trình ...:

import multiprocessing, time 

q = multiprocessing.Queue() 

def task(count): 
    for i in xrange(count): 
     q.put("%d mississippi" % i) 
    return "Done" 

def main(): 
    pool = multiprocessing.Pool() 
    result = pool.map_async(task, range(10)) 
    time.sleep(1) 
    while not q.empty(): 
     print q.get() 
    print result.get() 

if __name__ == "__main__": 
    main() 

Nếu bạn cần nhiều hàng đợi, ví dụ để tránh trộn lẫn tiến độ của các quy trình nhóm khác nhau, danh sách hàng đợi toàn cầu sẽ hoạt động (tất nhiên, mỗi quá trình sẽ cần biết chỉ số chỉ số trong danh sách để sử dụng, nhưng đó là OK để vượt qua đối số;).