QueueHandler日志需调用future.result()才显示，如何实现实时输出？

问题分析 #

你的核心问题有两个：

1. 子进程日志未正确路由到队列：func中使用的logger是主进程定义的全局变量，多进程环境下该变量是主进程logger的副本，并未使用子进程init_job配置的QueueHandler。
2. 终止信号发送时机不当：提前向队列发送None会导致消费线程直接退出，无法处理后续日志；原代码等待future.result()再处理日志，会造成日志延迟输出。

修复方案 #

修改点1：确保子进程日志使用QueueHandler

将func中的logger.info改为直接调用logging.getLogger().info，这样会使用子进程中配置的根logger及其QueueHandler，确保日志被放入队列。

修改点2：让消费线程持续运行直到任务完成

消费线程在后台异步运行，只要队列有日志就实时处理，等任务完成后再发送终止信号，无需提前阻塞在future.result()。

修复后的代码 #

import concurrent.futures
import logging
import logging.handlers
import multiprocessing
import threading

# force=True确保覆盖默认日志配置，避免子进程继承冗余handler
logging.basicConfig(level=logging.INFO, force=True)

def init_job(log_queue):
    # 重置子进程根logger的handler，仅保留QueueHandler
    root_logger = logging.getLogger()
    root_logger.handlers = [logging.handlers.QueueHandler(log_queue)]
    root_logger.setLevel(logging.INFO)

def func():
    # 使用子进程的根logger，确保日志进入队列
    logging.getLogger().info('Here')

def thread_func(log_queue):
    while True:
        record = log_queue.get()
        if record is None:
            break
        logging.info('Handling record')
        logging.getLogger().handle(record)

def main():
    log_queue = multiprocessing.Queue()
    # 启动日志消费线程
    thread = threading.Thread(target=thread_func, args=(log_queue,))
    thread.start()

    with concurrent.futures.ProcessPoolExecutor(initializer=init_job, initargs=(log_queue,)) as executor:
        future = executor.submit(func)
        # 等待任务完成，此时消费线程已在后台实时处理日志
        future.result()
        # 任务完成后发送终止信号
        log_queue.put(None)
    
    # 等待消费线程处理完剩余日志并退出
    thread.join()

if __name__ == '__main__':
    main()

额外优化建议 #

可以用threading.Event更优雅地控制消费线程终止，避免依赖队列中的None信号，同时防止线程长期阻塞：

def thread_func(log_queue, stop_event):
    while not stop_event.is_set():
        try:
            # 设置超时，避免线程一直阻塞在get()
            record = log_queue.get(timeout=0.1)
            logging.info('Handling record')
            logging.getLogger().handle(record)
        except multiprocessing.queues.Empty:
            continue

# 在main函数中：
stop_event = threading.Event()
thread = threading.Thread(target=thread_func, args=(log_queue, stop_event))
thread.start()

# 任务完成后触发终止信号
stop_event.set()
log_queue.put(None)  # 唤醒可能阻塞的线程
thread.join()

内容的提问来源于stack exchange，提问作者Daniel Walker