网上的方案来回就寥寥几种,但是我尝试下来都有问题,求助下社区
换 loguru 不就行了
既然是写日志可以直接往文本里面写吗,或者调第三方的包
换 loguru 不就行了
在 Python 中,由于logging
模块在多进程环境中存在一些限制,直接在多进程中使用logging
模块写入同一个文件可能会导致日志错乱或死锁问题。为了解决这个问题,可以采用以下几种策略:
使用QueueHandler
和QueueListener
:创建一个日志队列,让主进程负责收集日志并写入文件。这种方式可以避免多进程直接写入文件,从而解决日志错乱和死锁的问题。
使用进程安全的日志库:例如concurrent-log-handler
模块,它提供了线程和进程安全的日志处理方式。通过继承BaseRotatingHandler
并实现文件锁机制,可以确保日志写入的准确性,尽管这可能会牺牲一些效率。
避免在多进程中使用文件锁:在多进程环境中,文件锁可能会导致死锁。可以通过设计日志系统来避免使用文件锁,例如,让一个单独的进程或线程负责日志的收集和写入。
使用独立的日志文件:为每个进程分配一个独立的日志文件,这样可以避免多进程写入同一个文件时的冲突。
使用网络日志:通过SocketHandler
,可以让不同的进程通过网络发送日志消息到一个中心日志服务器,由服务器来处理日志存储。
使用第三方日志管理工具:例如使用Sentry
、Logstash
等工具,它们可以更好地处理多进程和多线程环境下的日志收集和存储。
在multiprocessing
的target
函数之外定义 logger:确保在创建进程之前已经配置好 logger,这样可以减少多进程中由于 logger 初始化导致的潜在问题。
使用multiprocessing-logging
库:这是一个专门为多进程日志设计的库,可以简化多进程日志的处理。
综上所述,要解决logging
模块在多进程测试时写入文件报错的问题,需要根据具体的应用场景选择合适的日志处理策略。在设计日志系统时,应该考虑到多进程环境下的特殊性,采用合适的方法来确保日志的正确性和完整性。
在 Python 中,如果你在多进程环境中使用 logging 模块并尝试将日志写入同一个文件,可能会遇到竞争条件和其他并发问题,因为默认的 logging 配置并不是为并发写操作设计的。
为了解决这个问题,你可以采取以下几种策略:
为每个进程使用单独的日志文件:
这是最简单的方法。你可以在创建每个进程时,给它们一个唯一的日志文件路径。
使用队列和单个日志处理进程:
你可以设置一个单独的进程来处理日志记录,其他所有进程都将日志消息发送到该进程的队列中。这个处理进程可以从队列中读取消息,并将它们写入日志文件。
使用线程安全的日志处理:
虽然 logging 模块本身不是线程安全的,但你可以使用线程安全的队列或其他同步机制来确保对日志文件的访问是顺序的。但是,在多进程环境中,这通常比使用单独的日志文件更复杂且效率更低。
使用日志轮转和延迟打开:
如果日志文件不是实时需要的,你可以考虑使用日志轮转,并在每个进程结束时才将日志写入文件。这可以通过延迟打开文件句柄直到需要写入日志来实现。
使用第三方库:
有些第三方库,如 loguru,提供了更高级的多进程日志记录功能。
下面是一个简单的示例,展示如何为每个进程使用单独的日志文件:
import logging
import multiprocessing
def worker_with_logging(process_id, log_file):
# 配置每个进程的日志记录器
logging.basicConfig(filename=log_file, level=logging.INFO,
format='%(asctime)s process %d: %(message)s' % process_id)
# 记录一些日志
logging.info('Starting work')
# ... 执行一些工作 ...
logging.info('Finishing work')
if __name__ == '__main__':
processes = []
for i in range(5): # 假设我们创建5个进程
log_file = f'log_{i}.txt'
p = multiprocessing.Process(target=worker_with_logging, args=(i, log_file))
p.start()
processes.append(p)
# 等待所有进程完成
for p in processes:
p.join()
在这个示例中,每个进程都有自己的日志文件,因此不存在并发写入同一个文件的问题。
我昨天就是按类似的这种 AI 生成的答案改,改来改去都不对,concurrent 使用不来,multiprocessing 那个多进程的时候还不能自动判断开了多少个进程😵
logging 换用 loguru 导致 pytest 报告收集不到日志,简洁的解决方案:
# 用于将loguru的日志传播到logging系统中
class PropagateHandler(logging.Handler):
# 被调用来处理日志记录
def emit(self, record):
logging.getLogger(record.name).handle(record)
logger.add(PropagateHandler())