<返回更多

python日志包从头到尾讲清楚

2022-11-26  今日头条  SuperOps
加入收藏


 

大纲

今天我们就来一起挖一挖Python/ target=_blank class=infotextkey>Python里面的各种日志的用法,废话不多说,日志包我为了讲清楚,涉及到的内容大纲如下:


 

基本打印


 

如上图看到的那么简单,在python中打印日志只要import logging包即可。直接执行效果如下图:


 

日志配置

如果没有日志配置,日志只能打印到stdout,那就会很糟糕,因此我们需要有一些配置可以输出到文件

import logging # add filemode="w" to overwrite logging.basicConfig(filename="sample.log", level=logging.INFO) logging.debug("This is a debug message") # This one won't log logging.info("Informational message") logging.error("An error has hAppened!")

增加basicconfig后日志就可以打印到指定文件,但是日志我们还需要设置日志格式

import logging logger = logging.getLogger('stream_logger') logger.setLevel(logging.INFO) console = logging.StreamHandler() formatter = logging.Formatter('%(asctime)s - %(name)s - %(message)s') console.setFormatter(formatter) logger.addHandler(console) logger.info("This is an informational message")

在python中还可以从配置文件读取日志规则

import logging import logging.config logging.config.fileConfig('logging.conf') logger = logging.getLogger("exampleApp") logger.info("Program started") logger.info("Done!")

logging.conf配置如下:

[loggers] keys=root,exampleApp [handlers] keys=fileHandler, consoleHandler [formatters] keys=myFormatter [logger_root] level=CRITICAL handlers=consoleHandler [logger_exampleApp] level=INFO handlers=fileHandler qualname=exampleApp [handler_consoleHandler] class=StreamHandler level=DEBUG formatter=myFormatter args=(sys.stdout,) [handler_fileHandler] class=FileHandler formatter=myFormatter args=("config.log",) [formatter_myFormatter] format=%(asctime)s - %(name)s - %(levelname)s - %(message)s datefmt=日志过滤

通过Formatter就可以搞定日志格式问题,如果生产环境日志太多,又想有一些日志过滤。

import logging import sys class MyFilter(logging.Filter): def filter(self, record): if record.funcName == 'a': return False return True logger = logging.getLogger('filter_test') logger.addFilter(MyFilter()) def a(): """ Ignore this function's log messages """ logger.debug('Message from function a') def b(): logger.debug('Message from B') if __name__ == '__main__': logging.basicConfig(stream=sys.stderr, level=logging.DEBUG) a() b()

这样你的日志就可以实现过滤效果。

日志切割

可是生产环境真正运行的服务,又不仅仅是运行一天,日志一直打印到同一个文件非常不方便debug和定位,有时候一天几个G的文件,如果运行30天去查日志时会非常耗时,我们需要一个日志切割的逻辑来保障日志可以按照时间动态切换文件。

 

import logging import time from logging.handlers import TimedRotatingFileHandler def create_timed_rotating_log(path): """""" logger = logging.getLogger("Rotating Log") logger.setLevel(logging.INFO) handler = TimedRotatingFileHandler(path, when="s", interval=5, backupCount=5) logger.addHandler(handler) for i in range(6): logger.info("This is a test!") time.sleep(75) if __name__ == "__main__": log_file = "timed_rotation.log" create_timed_rotating_log(log_file) import logging import time from logging.handlers import RotatingFileHandler def create_rotating_log(path): """ Creates a rotating log """ logger = logging.getLogger("Rotating Log") logger.setLevel(logging.INFO) # add a rotating handler handler = RotatingFileHandler(path, maxBytes=20, backupCount=5) logger.addHandler(handler) for i in range(10): logger.info("This is test log line %s" % i) time.sleep(1.5) if __name__ == "__main__": log_file = "rotated.log" create_rotating_log(log_file)多进程日志

 

在python中因为受GIL的影响,涉及到并发性能问题时经常使用多进程方式来解决,此时多进程日志打印会受到文件锁的影响,为了避免日志打印错乱,日志异常,我们针对这种场景需要定制化处理下日志:

import logging from logging.handlers import RotatingFileHandler, QueueHandler from multiprocessing import Process class MultiProcessQueueLoggingListner(Process): def __init__(self, name, queue): super().__init__() self.name = name self.queue = queue self.logger = logging.getLogger(name) self.file_handler = RotatingFileHandler(name, maxBytes=536870912, backupCount=2) self.formatter = logging.Formatter('%(asctime)s %(processName)-10s %(name)s %(levelname)-8s %(message)s') self.file_handler.setFormatter(self.formatter) self.logger.addHandler(self.file_handler) def run(self): while True: try: record = self.queue.get() if record is None: break self.logger.handle(record) except Exception: import sys, traceback print('Whoops! Problem:', file=sys.stderr) traceback.print_exc(file=sys.stderr) class MulitProcessQueueLogger(object): def __init__(self, name, queue): self.name = name self.queue = queue self.queue_handler = QueueHandler(queue) self.logger = logging.getLogger(name) self.logger.addHandler(self.queue_handler) self.logger.setLevel(logging.DEBUG)

测试代码如下:

import multi_process_logging import multiprocessing from time import sleep def worker(po): name = multiprocessing.current_process().name po = multi_process_logging.MulitProcessQueueLogger('test.log', q) print("In worker") for i in range(10): po.logger.info(f"Logging from {name} line {i}") po.queue.put(None) def main(): q = multiprocessing.Queue() lp = multi_process_logging.MultiProcessQueueLoggingListner('test.log', q) lp.start() p = multiprocessing.Process(target=worker, args=(q,)) p.start() p.join() lp.join() if __name__ == '__main__': main()总结

至此,大部分python打印日志的场景使用方式都讲解到了,希望对读者有所帮助。

声明:本站部分内容来自互联网,如有版权侵犯或其他问题请与我们联系,我们将立即删除或处理。
▍相关推荐
更多资讯 >>>