Hướng dẫn python multiprocessing process does not terminate - quá trình đa xử lý python không chấm dứt

Tôi đang cố gắng chạy nhiều yêu cầu API song song với Multiprocessing.Process và yêu cầu. Tôi đặt URL để phân tích cú pháp vào ví dụ có thể tham gia và đặt lại nội dung vào ví dụ hàng đợi. Tôi đã nhận thấy rằng việc đưa phản hồi. Incontent vào hàng đợi bằng cách nào đó ngăn không cho quá trình chấm dứt.

Đây là ví dụ đơn giản hóa chỉ với 1 quy trình (Python 3.5):

import multiprocessing as mp
import queue
import requests
import time


class ChildProcess(mp.Process):
    def __init__(self, q, qout):
        super().__init__()
        self.qin = qin
        self.qout = qout
        self.daemon = True

    def run(self):
        while True:
            try:
                url = self.qin.get(block=False)
                r = requests.get(url, verify=False)
                self.qout.put(r.content)
                self.qin.task_done()
            except queue.Empty:
                break
            except requests.exceptions.RequestException as e:
                print(self.name, e)
                self.qin.task_done()
        print("Infinite loop terminates")


if __name__ == '__main__':
    qin = mp.JoinableQueue()
    qout = mp.Queue()
    for _ in range(5):
        qin.put('http://en.wikipedia.org')
    w = ChildProcess(qin, qout)
    w.start()
    qin.join()
    time.sleep(1)
    print(w.name, w.is_alive())

Sau khi chạy mã tôi nhận được:

Vòng lặp vô hạn chấm dứt

Trẻ em-1 đúng

Vui lòng giúp hiểu lý do tại sao quá trình không chấm dứt sau khi chạy chức năng thoát.

CẬP NHẬT: Đã thêm câu lệnh in để hiển thị vòng lặp chấm dứt

Đã hỏi ngày 23 tháng 5 năm 2018 lúc 8:34May 23, 2018 at 8:34

1

Như đã lưu ý trong tài liệu đường ống và hàng đợi

Nếu một quá trình trẻ em đã đặt các mục lên hàng đợi (và nó chưa sử dụng tham gia

Điều này có nghĩa là nếu bạn cố gắng tham gia vào quá trình đó, bạn có thể nhận được bế tắc trừ khi bạn chắc chắn rằng tất cả các mục đã được đặt vào hàng đợi đã được tiêu thụ.

...

Lưu ý rằng một hàng đợi được tạo bằng cách sử dụng Trình quản lý không có vấn đề này.

Nếu bạn chuyển sang hàng đợi người quản lý, thì quá trình sẽ chấm dứt thành công:

import multiprocessing as mp
import queue
import requests
import time


class ChildProcess(mp.Process):
    def __init__(self, q, qout):
        super().__init__()
        self.qin = qin
        self.qout = qout
        self.daemon = True

    def run(self):
        while True:
            try:
                url = self.qin.get(block=False)
                r = requests.get(url, verify=False)
                self.qout.put(r.content)
                self.qin.task_done()
            except queue.Empty:
                break
            except requests.exceptions.RequestException as e:
                print(self.name, e)
                self.qin.task_done()
        print("Infinite loop terminates")


if __name__ == '__main__':
    manager = mp.Manager()
    qin = mp.JoinableQueue()
    qout = manager.Queue()
    for _ in range(5):
        qin.put('http://en.wikipedia.org')
    w = ChildProcess(qin, qout)
    w.start()
    qin.join()
    time.sleep(1)
    print(w.name, w.is_alive())

Đã trả lời ngày 12 tháng 11 năm 2019 lúc 9:49Nov 12, 2019 at 9:49

KimkimKim

6989 Huy hiệu bạc20 Huy hiệu Đồng9 silver badges20 bronze badges

Một chút khó để tìm ra điều này dựa trên tài liệu Queue - Tôi đã đấu tranh với cùng một vấn đề.

Khái niệm chính ở đây là trước khi một chuỗi nhà sản xuất chấm dứt, nó tham gia bất kỳ hàng đợi nào mà nó có dữ liệu put vào; Việc tham gia sau đó chặn cho đến khi luồng nền của hàng đợi chấm dứt, điều này chỉ xảy ra khi hàng đợi trống. Vì vậy, về cơ bản, trước khi ChildProcess của bạn có thể thoát ra, ai đó phải tiêu thụ tất cả những thứ nó put vào hàng đợi!which only happens when the queue is empty. So basically, before your ChildProcess can exit, someone has to consume all the stuff it put into the queue!

Có một số tài liệu về hàm Queue.cancel_join_thread, được cho là sẽ phá vỡ vấn đề này, nhưng tôi không thể có được bất kỳ hiệu ứng nào - có lẽ tôi không sử dụng nó một cách chính xác.

Đây là một sửa đổi ví dụ, bạn có thể thực hiện điều đó sẽ khắc phục sự cố:

if __name__ == '__main__':
    qin = mp.JoinableQueue()
    qout = mp.Queue()
    for _ in range(5):
        qin.put('http://en.wikipedia.org')
    w = ChildProcess(qin, qout)
    w.start()
    qin.join()
    while True:
        try:
            qout.get(True, 0.1)     # Throw away remaining stuff in qout (or process it or whatever,
                                    # just get it out of the queue so the queue background process
                                    # can terminate, so your ChildProcess can terminate.
        except queue.Empty:
            break
    w.join()                # Wait for your ChildProcess to finish up.
    # time.sleep(1)         # Not necessary since we've joined the ChildProcess
    print(w.name, w.is_alive())

Đã trả lời ngày 5 tháng 9 năm 2019 lúc 18:48Sep 5, 2019 at 18:48

BrioniusbrioniusBrionius

13.6K3 Huy hiệu vàng36 Huy hiệu bạc48 Huy hiệu đồng3 gold badges36 silver badges48 bronze badges

Thêm một cuộc gọi đến w.terminate () trên tin nhắn in.

Liên quan đến lý do tại sao quá trình không chấm dứt chính nó; Mã chức năng của bạn là một vòng lặp vô hạn, vì vậy nó không bao giờ quay lại. Gọi chấm dứt tín hiệu quá trình để tự sát.

Đã trả lời ngày 23 tháng 5 năm 2018 lúc 8:53May 23, 2018 at 8:53

Chen A.CHEN A.Chen A.

9.1473 Huy hiệu vàng37 Huy hiệu bạc54 Huy hiệu đồng3 gold badges37 silver badges54 bronze badges

2

Làm thế nào để khóa đa xử lý Python hoạt động?

Pool trong đa xử lý Python là gì?

Chủ đề hàng đợi đa xử lý Python có an toàn không?

Làm thế nào để xếp hàng đa xử lý Python hoạt động?