博客

python清理子进程机制剖析

  • Post category:Python

Python清理子进程机制剖析

当使用multiprocessing模块来启动子进程时，有时候需要确保子进程在运行结束后能够清理干净。Python提供了一个清理子进程的机制来保证这点。

子进程清理机制简介

子进程清理机制包括两个部分：在主进程退出时，Python能够自动清理所有运行的子进程；而在创建子进程时，也可以设置子进程在退出时自动清理。

主进程自动清理子进程

当主进程执行结束时，Python会自动清理并终止未结束的子进程。这个机制确保了主进程退出后所有子进程都会被及时清理，避免了悬挂子进程。

使用该机制非常简单，只需要在主进程的最后调用multiprocessing模块的join()方法即可。该方法会阻塞主进程，直到所有子进程都运行结束，然后再清理所有的子进程。

import multiprocessing
import time

def worker():
    print("子进程开始执行")
    time.sleep(3)
    print("子进程执行结束")

if __name__ == '__main__':
    p = multiprocessing.Process(target=worker)
    p.start()
    p.join()
    print("主进程执行结束")

上述代码中，主进程先创建一个子进程，并调用start()方法启动子进程。然后调用join()方法阻塞主进程，直到子进程运行结束。最后主进程执行完毕，Python自动清理子进程。

子进程自动清理

为了确保子进程退出时能够自动清理，需要在创建子进程时设置daemon属性，将其设置为True。当子进程运行结束时，Python会自动将其清理。

import multiprocessing
import time

def worker():
    print("子进程开始执行")
    time.sleep(3)
    print("子进程执行结束")

if __name__ == '__main__':
    p = multiprocessing.Process(target=worker)
    p.daemon = True
    p.start()
    time.sleep(1)

上述代码中，在创建子进程后，将其daemon属性设置为True，然后启动子进程。在执行完毕后，子进程会自动清理。注意，此时主进程不能调用join()方法，否则会抛出异常。

示例说明

示例1

下面是一个基于multiprocessing模块的简单进程池示例。该进程池使用Process类动态分配子进程来执行其中的任务。

import multiprocessing
import time

def worker(task_id):
    print("任务%s开始执行" % task_id)
    time.sleep(3)
    print("任务%s执行结束" % task_id)

class ProcessPool(object):

    def __init__(self, max_process_count):
        self.max_process_count = max_process_count
        self.processes = []

    def add_task(self, task_id):
        if len(self.processes) >= self.max_process_count:
            self.wait_for_process()
        p = multiprocessing.Process(target=worker, args=(task_id,))
        self.processes.append(p)
        p.start()

    def wait_for_process(self):
        for p in self.processes:
            p.join()
            print("子进程%s结束" % p.pid)
            self.processes.remove(p)

    def wait_for_complete(self):
        for p in self.processes:
            p.join()
            print("子进程%s结束" % p.pid)

if __name__ == '__main__':
    pool = ProcessPool(2)
    for i in range(5):
        pool.add_task(i)
    pool.wait_for_complete()
    print("主进程执行结束")

上述代码是一个简单的进程池实现。在添加任务时，如果进程池中的进程数已经达到max_process_count上限，会调用wait_for_process()方法等待已经运行的子进程结束。

在该示例代码中，需要确保所有的子进程能够在运行结束后被清理干净。为了实现该功能，只需要在创建子进程时将其daemon属性设置为True即可。

示例2

下面是一个基于multiprocessing的守护进程示例。启动守护进程后，该进程每隔一段时间检查日志文件是否存在，如果不存在则创建一个新的日志文件。

import multiprocessing
import time

def daemon_process():
    while True:
        if not os.path.isfile("test.log"):
            with open("test.log", "w") as f:
                pass
        time.sleep(5)

if __name__ == '__main__':
    p = multiprocessing.Process(target=daemon_process)
    p.daemon = True
    p.start()
    while True:
        time.sleep(1)

在该示例代码中，守护进程是一个无限循环的进程，需要确保它在运行结束后能够被正确清理。因此，在创建子进程时，将其daemon属性设置为True。

Tags: ,