Python多线程日志记录方案 Python多线程安全输出日志方法

import logging
import threading
import time
<h1>配置日志格式和输出方式</h1><p>logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(threadName)s] %(levelname)s: %(message)s',
handlers=[
logging.FileHandler("app.log", encoding='utf-8'),
logging.StreamHandler()
]
)</p><p>def worker():
for i in range(3):
logging.info(f"正在处理任务 {i}")
time.sleep(0.1)</p><h1>创建多个线程</h1><p>threads = []
for i in range(3):
t = threading.Thread(target=worker, name=f"线程-{i}")
threads.append(t)
t.start()</p><p>for t in threads:
t.join()</p>

# 错误示例：手动打开文件并写入
log_file = open("bad.log", "w")
def bad_log(msg):
    log_file.write(msg + "\n")  # 多线程同时写会冲突

import logging
from logging.handlers import QueueHandler, QueueListener
import queue
<h1>创建队列</h1><p>log_queue = queue.Queue()</p><h1>配置主日志器</h1><p>logger = logging.getLogger()
logger.setLevel(logging.INFO)</p><h1>添加队列处理器</h1><p>queue_handler = QueueHandler(log_queue)
logger.addHandler(queue_handler)</p><h1>配置实际写入的处理器</h1><p>file_handler = logging.FileHandler("async.log", encoding='utf-8')
stream_handler = logging.StreamHandler()</p><h1>启动监听器（单独线程）</h1><p>listener = QueueListener(log_queue, file_handler, stream_handler)
listener.start()</p><h1>使用日志</h1><p>def task(name):
for i in range(2):
logging.info(f"{name} 执行任务 {i}")</p><p>import threading
t1 = threading.Thread(target=task, args=("任务A",))
t2 = threading.Thread(target=task, args=("任务B",))
t1.start(); t2.start()
t1.join(); t2.join()</p><p>listener.stop()  # 清理资源</p>

format='%(asctime)s [%(threadName)s/%(thread)d] %(levelname)s: %(message)s'