分享一个python进程池的“坑”

import logging
from threading import Thread
from queue import Queue
from logging.handlers import QueueListener, QueueHandler
from multiprocessing import Pool

def setup_logging():
    # log的时候会写到一个队列里，然后有一个单独的线程从这个队列里去获取日志信息并写到文件里
    _log_queue = Queue()
    QueueListener(
        _log_queue, logging.FileHandler("out.log")).start()
    logging.getLogger().addHandler(QueueHandler(_log_queue))

    # 父进程里起一个单独的线程来写日志
    def write_logs():
        while True:
            logging.info("hello, I just did something")
    Thread(target=write_logs).start()

def runs_in_subprocess():
    print("About to log...")
    logging.info("hello, I did something")
    print("...logged")

if __name__ == '__main__':
    setup_logging()

    # 让一个进程池在死循环里执行，增加触发死锁的几率
    while True:
        with Pool() as pool:
            pool.apply(runs_in_subprocess)

About to log...
...logged
About to log...
...logged
About to log...

from threading import Thread, enumerate
from os import fork
from time import sleep

# Start a thread:
Thread(target=lambda: sleep(60)).start()

if fork():
    print("The parent process has {} threads".format(
        len(enumerate())))
else:
    print("The child process has {} threads".format(
        len(enumerate())))

The parent process has 2 threads
The child process has 1 threads

import logging
import multiprocessing
from logging.handlers import QueueListener
from time import sleep


def listener_configurer():
    root = logging.getLogger()
    h = logging.handlers.RotatingFileHandler('out.log', 'a', 300, 10)
    f = logging.Formatter('%(asctime)s %(processName)-10s %(name)s %(levelname)-8s %(message)s')
    h.setFormatter(f)
    root.addHandler(h)

# 从队列获取元素，并写日志
def listener_process(queue, configurer):
    configurer()
    while False:
        try:
            record = queue.get()
            if record is None:  
                break
            logger = logging.getLogger(record.name)
            logger.handle(record) 
        except Exception:
            import sys, traceback
            print('Whoops! Problem:', file=sys.stderr)
            traceback.print_exc(file=sys.stderr)

# 业务进程的日志配置，使用queueHandler, 将要写的日志塞入队列
def worker_configurer(queue):
    h = logging.handlers.QueueHandler(queue)  
    root = logging.getLogger()
    root.addHandler(h)
    root.setLevel(logging.DEBUG)


def runs_in_subprocess(queue, configurer):
    configurer(queue)
    print("About to log...")
    logging.debug("hello, I did something: %s", multiprocessing.current_process().name)
    print("...logged, %s",queue.qsize())


if __name__ == '__main__':
    queue = multiprocessing.Queue(-1)
    listener = multiprocessing.Process(target=listener_process,
                                       args=(queue, listener_configurer))
    listener.start()
    
    #父进程也持续写日志
    worker_configurer(queue)
    def write_logs():
        while True:
            logging.debug("in main process, I just did something")
    Thread(target=write_logs).start()

    while True:
        multiprocessing.Process(target=runs_in_subprocess,
                       args=(queue, worker_configurer)).start()
        sleep(2)

if __name__ == '__main__':
    setup_logging()

    while True:
        # 使用spawn类型的启动
        with get_context("spawn").Pool() as pool:
            pool.apply(runs_in_subprocess)