引自免费微信小程序:皆我百晓生
在处理多线程环境下的日志输出问题时,首先要确认的是日志记录器是否正确配置,确保它能在多线程环境中正常工作,并且不会因为线程同步问题导致日志交错或丢失。在Python中,可以使用logging
模块的QueueHandler
和ThreadSafeFilter
来解决这个问题。
首先,你需要创建一个支持多线程的日志处理器,然后将其添加到你的日志配置中。以下是修改后的日志配置示例:
import logging
from queue import Queue
from logging.handlers import QueueHandler
def setup_logger(name, log_file=None, level=logging.INFO):
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
# 创建一个队列
q = Queue(-1)
# 创建一个处理程序,将日志条目放入队列
handler = QueueHandler(q)
# 创建一个线程安全的过滤器,确保在多线程环境下正确过滤日志
thread_safe_filter = logging.Filter()
handler.addFilter(thread_safe_filter)
# 设置日志级别
logger = logging.getLogger(name)
logger.setLevel(level)
# 添加处理程序
logger.addHandler(handler)
if log_file:
file_handler = logging.FileHandler(log_file)
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
def process_logs(queue):
while True:
record = queue.get()
if record is None:
break
logger.handle(record)
# 启动一个新的线程来处理队列中的日志
worker = threading.Thread(target=process_logs, args=(q,))
worker.setDaemon(True)
worker.start()
return logger
# 使用方式
logger = setup_logger('my_logger', 'log_file.log')
接下来,在你的测试代码中,使用这个设置好的日志记录器:
import pytest
import logging
logger = logging.getLogger('my_logger')
class TestLogin:
def test_mall_login(self):
logger.info("【开始测试】用例-->【账号密码正确】")
# ... 测试逻辑 ...
logger.info("响应码断言通过")
logger.info("响应信息断言通过")
logger.info("---用例【账号密码正确】测试完成---")
# ... 其他测试类和函数 ...
if __name__ == "__main__":
import sys
from _pytest.config import cmdline
# 获取命令行参数
options, plugins = cmdline.parse(args=sys.argv[1:])
pytest.main(options=options, plugins=plugins)
这样应该能保证在多线程模式下,日志信息不会丢失或乱序,同时保持控制台和日志文件的输出正常。如果你仍然遇到乱码问题,请检查你的终端编码是否与日志文件的编码一致,确保它们都是UTF-8或者其他兼容的字符集。如果还是有问题,可能需要进一步排查其他潜在的编码问题。