本篇內(nèi)容介紹了“怎么用Python實(shí)現(xiàn)強(qiáng)大的 logging 模塊”的有關(guān)知識(shí)，在實(shí)際案例的操作過(guò)程中，不少人都會(huì)遇到這樣的困境，接下來(lái)就讓小編帶領(lǐng)大家學(xué)習(xí)一下如何處理這些情況吧！希望大家仔細(xì)閱讀，能夠?qū)W有所成！

目前成都創(chuàng)新互聯(lián)已為千余家的企業(yè)提供了網(wǎng)站建設(shè)、域名、雅安服務(wù)器托管、成都網(wǎng)站托管、企業(yè)網(wǎng)站設(shè)計(jì)、云浮網(wǎng)站維護(hù)等服務(wù)，公司將堅(jiān)持客戶導(dǎo)向、應(yīng)用為本的策略，正道將秉承"和諧、參與、激情"的文化，與客戶和合作伙伴齊心協(xié)力一起成長(zhǎng)，共同發(fā)展。

Python 的 logging 模塊實(shí)現(xiàn)了靈活的日志系統(tǒng)。整個(gè)模塊僅僅 3 個(gè)類，不到 5000 行代碼的樣子，學(xué)習(xí)它可以加深對(duì)程序日志的了解，本文分下面幾個(gè)部分:

• logging 簡(jiǎn)介

• logging API 設(shè)計(jì)

• 記錄器對(duì)象 Logger

• 日志記錄對(duì)象 LogRecord

• 處理器對(duì)象 Hander

• 格式器對(duì)象 Formatter

• 滾動(dòng)日志文件處理器

• 小結(jié)

• 小技巧

logging 簡(jiǎn)介

本次代碼使用的是 python 3.8.5 的版本，官方中文文檔 3.8.8 。參考鏈接中官方中文文檔非常詳細(xì)，建議先看一遍了解日志使用。

我們主要研究日志如何輸出到標(biāo)準(zhǔn)窗口這一主線；日志的配置，日志的線程安全及各種特別的Handler等支線可以先忽略。

logging API 設(shè)計(jì)

先看看日志使用：

import logging  logging.basicConfig(level=logging.INFO, format='%(levelname)-8s %(name)-10s %(asctime)s %(message)s') lang = {"name": "python", "age":20} logging.info('This is a info message %s', lang) logging.debug('This is a debug message') logging.warning('This is a warning message')  logger = logging.getLogger(__name__) logger.warning('This is a warning')

輸出內(nèi)容如下:

INFO     root       2021-03-04 00:03:53,473 This is a info message {'name': 'python', 'age': 20} WARNING  root       2021-03-04 00:03:53,473 This is a warning message WARNING  __main__   2021-03-04 00:03:53,473 This is a warning

可以看到 logging 的使用非常方便，模塊直接提供了一組API。

root = RootLogger(WARNING)  # 默認(rèn)提供的logger Logger.root = root Logger.manager = Manager(Logger.root)  def debug(msg, *args, **kwargs): # info，warning等api類似     if len(root.handlers) == 0:         basicConfig()  # 默認(rèn)配置     root.debug(msg, *args, **kwargs)  def getLogger(name=None):     if name:         return Logger.manager.getLogger(name)  # 創(chuàng)建特定的logger     else:         return root  # 返回默認(rèn)的logger

這種API的提供方式，我們?cè)?nbsp;requests 中也有看到。api中很重要的設(shè)置config的方式:

def basicConfig(**kwargs):     ...     if handlers is None:         filename = kwargs.pop("filename", None)         mode = kwargs.pop("filemode", 'a')         if filename:             h = FileHandler(filename, mode)         else:             stream = kwargs.pop("stream", None)             h = StreamHandler(stream)  # 默認(rèn)的handler         handlers = [h]     dfs = kwargs.pop("datefmt", None)     style = kwargs.pop("style", '%')     fs = kwargs.pop("format", _STYLES[style][1])     fmt = Formatter(fs, dfs, style)  # 生成formatter     for h in handlers:         if h.formatter is None:             h.setFormatter(fmt)         root.addHandler(h)  # 設(shè)置root的handler     level = kwargs.pop("level", None)     if level is not None:         root.setLevel(level)  # 設(shè)置日志級(jí)別

可以看到，日志的配置主要包括下面幾項(xiàng):

• level 日志級(jí)別

• format 信息格式化模版

• filename 輸出到文件

• datefmt %Y-%m-%d %H:%M:%S,uuu 時(shí)間的格式模版

• style [ % , { ,$] 格式樣板

演示代碼輸出中，可以看到debug日志沒(méi)有顯示，是因?yàn)?nbsp;debug < info :

CRITICAL = 50 FATAL = CRITICAL ERROR = 40 WARNING = 30 WARN = WARNING INFO = 20 DEBUG = 10 NOTSET = 0

記錄器對(duì)象 Logger

查看Logger之前，先看logger對(duì)象的管理類Manager

_loggerClass = Logger  class Manager(object):     def __init__(self, rootnode):         self.root = rootnode         self.disable = 0         self.loggerDict = {}  # 所有日志記錄對(duì)象的字典     ...     def getLogger(self, name):         rv = None         if name in self.loggerDict:             rv = self.loggerDict[name]  # 獲取已經(jīng)創(chuàng)建過(guò)的同名logger             ...         else:             rv = (self.loggerClass or _loggerClass)(name)  # 創(chuàng)建新的logger             rv.manager = self             self.loggerDict[name] = rv             ...         return rv

日志過(guò)濾器

class Filterer(object):      def __init__(self):         self.filters = []      def addFilter(self, filter):         self.filters.append(filter)      def removeFilter(self, filter):         self.filters.remove(filter)      def filter(self, record):         rv = True         for f in self.filters:  # 過(guò)濾日志             if hasattr(f, 'filter'):                 result = f.filter(record)             else:                 result = f(record) # assume callable - will raise if not             if not result:                 rv = False                 break         return r

核心的 Logger 實(shí)際上只是一個(gè)控制中心：

class Logger(Filterer):  # logger可以過(guò)濾日志     def __init__(self, name, level=NOTSET):         Filterer.__init__(self)         self.name = name         self.level = _checkLevel(level)         self.parent = None  # 日志可以有層級(jí)         self.propagate = True         self.handlers = []  # 可以輸出到多個(gè)handler         self.disabled = False  # 可以關(guān)閉         self._cache = {}          def debug(self, msg, *args, **kwargs):  # 輸出debug日志         if self.isEnabledFor(DEBUG):             self._log(DEBUG, msg, args, **kwargs)

logger可以判斷日志級(jí)別:

def isEnabledFor(self, level):     if self.disabled:         return False      try:         return self._cache[level]     except KeyError:         try:             if self.manager.disable >= level:                 is_enabled = self._cache[level] = False             else:                 is_enabled = self._cache[level] = (                     level >= self.getEffectiveLevel()                 )         return is_enabled  def getEffectiveLevel(self):     logger = self     while logger:         if logger.level:             return logger.level         logger = logger.parent     return NOTSET

日志輸出:

def _log(self, level, msg, args, exc_info=None, extra=None, stack_info=False,          stacklevel=1):     ...     fn, lno, func = "(unknown file)", 0, "(unknown function)"     ...     # 生成日志記錄     record = self.makeRecord(self.name, level, fn, lno, msg, args,                              exc_info, func, extra, sinfo)     # 使用handler處理日志     self.handle(record)

日志記錄的生產(chǎn)，就是創(chuàng)建一個(gè)LogRecord對(duì)象:

_logRecordFactory = LogRecord  def makeRecord(self, name, level, fn, lno, msg, args, exc_info,                func=None, extra=None, sinfo=None):     ...     rv = _logRecordFactory(name, level, fn, lno, msg, args, exc_info, func,                          sinfo)     ...     return rv

使用logger對(duì)象的所有handler處理日志：

def handle(self, record):     c = self     found = 0     while c:         for hdlr in c.handlers:  # 使用所有的handler處理日志             found = found + 1             if record.levelno >= hdlr.level:                 hdlr.handle(record)

root-logger的handler是在config中配置的:

def basicConfig(**kwargs):     ...     root.addHandler(h)  # 設(shè)置root的handler

日志記錄對(duì)象 LogRecord

日志記錄對(duì)象非常簡(jiǎn)單:

class LogRecord(object):     def __init__(self, name, level, pathname, lineno,                  msg, args, exc_info, func=None, sinfo=None, **kwargs):         ct = time.time()         self.name = name  # logger名稱         self.msg = msg  # 日志標(biāo)識(shí)信息         ...         self.args = args  # 變量         self.levelname = getLevelName(level)         ...          def getMessage(self):         msg = str(self.msg)         if self.args:             msg = msg % self.args  # 格式化消息         return msg

處理器對(duì)象 Hander

頂級(jí)Handler定義了Handler的模版方法

class Handler(Filterer):  # 處理器也可以過(guò)濾日志     def __init__(self, level=NOTSET):         Filterer.__init__(self)         self._name = None         self.level = _checkLevel(level)  # handler也有日志級(jí)別         self.formatter = None         _addHandlerRef(self)         self.createLock()              def handle(self, record):  # 處理日志         rv = self.filter(record)  # 過(guò)濾日志         if rv:             self.acquire()  # 申請(qǐng)鎖             try:                 self.emit(record)  # 提交記錄，由不同子類實(shí)現(xiàn)              finally:                 self.release()  # 釋放鎖         return rv

默認(rèn)的console流 StreamHandler

class StreamHandler(Handler):      terminator = '\n'  # 自動(dòng)換行      def __init__(self, stream=None):         Handler.__init__(self)         if stream is None:             stream = sys.stderr  # 默認(rèn)使用stderr輸出         self.stream = stream          def emit(self, record):         try:             msg = self.format(record)  # 格式化日志記錄             stream = self.stream             stream.write(msg + self.terminator)  # 寫(xiě)日志             self.flush()  # 刷新寫(xiě)緩存         except Exception:             ...          def format(self, record):         if self.formatter:             fmt = self.formatter         else:             fmt = _defaultFormatter         return fmt.format(record)  # 使用格式化器格式化日志記錄

為什么使用stderr，可以看下面的測(cè)試中的輸出都是到console:

print("haha") print("fatal error", file=sys.stderr) sys.stderr.write("fatal error\n")

格式器對(duì)象 Formatter

格式化器主要使用Formatter和Style實(shí)現(xiàn)

class Formatter(object):     def __init__(self, fmt=None, datefmt=None, style='%', validate=True):         self._style = _STYLES[style][0](fmt)         self._fmt = self._style._fmt         self.datefmt = datefmt          def format(self, record):         record.message = record.getMessage()         s = self.formatMessage(record)         return s              def formatMessage(self, record):         return self._style.format(record)  # 格式化

Style類

class PercentStyle(object):      default_format = '%(message)s'     asctime_format = '%(asctime)s'     asctime_search = '%(asctime)'     validation_pattern = re.compile(r'%\(\w+\)[#0+ -]*(\*|\d+)?(\.(\*|\d+))?[diouxefgcrsa%]', re.I)      def __init__(self, fmt):         self._fmt = fmt or self.default_format      def usesTime(self):         return self._fmt.find(self.asctime_search) >= 0      def validate(self):         """Validate the input format, ensure it matches the correct style"""         if not self.validation_pattern.search(self._fmt):             raise ValueError("Invalid format '%s' for '%s' style" % (self._fmt, self.default_format[0]))      def _format(self, record):         return self._fmt % record.__dict__  # 格式化日志記錄對(duì)象      def format(self, record):         try:             return self._format(record)         except KeyError as e:             raise ValueError('Formatting field not found in record: %s' % e)

滾動(dòng)日志文件處理器

線上的日志持續(xù)輸出到一個(gè)文件的話，會(huì)讓文件巨大，即有加劇了丟失的風(fēng)險(xiǎn)，也難以處理。通常有按照大小滾動(dòng)或者按照日期滾動(dòng)的方法，這個(gè)功能非常重要。先看滾動(dòng)日志處理器模版:

class BaseRotatingHandler(logging.FileHandler):     def emit(self, record):         try:             if self.shouldRollover(record): # 判斷是否需要滾動(dòng)                 self.doRollover()  # 滾動(dòng)日志             logging.FileHandler.emit(self, record)  # 輸出日志         except Exception:             self.handleError(record)          def rotate(self, source, dest):         if not callable(self.rotator):             if os.path.exists(source):                 os.rename(source, dest)  # 重命名日志文件         else:             self.rotator(source, dest)

按大小滾動(dòng) RotatingFileHandler

按照文件大小滾動(dòng)的處理器:

class RotatingFileHandler(BaseRotatingHandler):      def __init__(self, filename, mode='a', maxBytes=0, backupCount=0, encoding=None, delay=False):         if maxBytes > 0:             mode = 'a'         BaseRotatingHandler.__init__(self, filename, mode, encoding, delay)         self.maxBytes = maxBytes  # 單個(gè)文件大小上限         self.backupCount = backupCount  # 日志備份數(shù)量              def doRollover(self):  # 執(zhí)行滾動(dòng)         if self.stream:             self.stream.close()  # 關(guān)閉當(dāng)前的流             self.stream = None         if self.backupCount > 0:             for i in range(self.backupCount - 1, 0, -1):                 sfn = self.rotation_filename("%s.%d" % (self.baseFilename, i))                 dfn = self.rotation_filename("%s.%d" % (self.baseFilename,                                                         i + 1))                 if os.path.exists(sfn):                     if os.path.exists(dfn):                         os.remove(dfn)                     os.rename(sfn, dfn)             dfn = self.rotation_filename(self.baseFilename + ".1")             if os.path.exists(dfn):                 os.remove(dfn)             self.rotate(self.baseFilename, dfn)  # 重命名文件         if not self.delay:             self.stream = self._open()  # 如果shouldRollover延遲，可以打開(kāi)新的流      def shouldRollover(self, record):  # 判斷是否需要滾動(dòng)         if self.stream is None:  # 立即打開(kāi)流             self.stream = self._open()         if self.maxBytes > 0:                msg = "%s\n" % self.format(record)             self.stream.seek(0, 2)  #due to non-posix-compliant Windows feature             if self.stream.tell() + len(msg) >= self.maxBytes:  # 判斷大小                 return 1         return 0

文件大小滾動(dòng)就是在記錄日志時(shí)候判斷文檔是否超過(guò)上限，超過(guò)則重命名舊日志，生成新日志。

按照日期滾動(dòng) TimedRotatingFileHandler

按照日期滾動(dòng)的處理器:

class TimedRotatingFileHandler(BaseRotatingHandler):     def __init__(self, filename, when='h', interval=1, backupCount=0, encoding=None, delay=False, utc=False, atTime=None):         BaseRotatingHandler.__init__(self, filename, 'a', encoding, delay)         self.when = when.upper()         self.backupCount = backupCount         self.utc = utc         self.atTime = atTime         # 日期設(shè)置,支持多種方式         if self.when == 'S':             self.interval = 1 # one second             self.suffix = "%Y-%m-%d_%H-%M-%S"             self.extMatch = r"^\d{4}-\d{2}-\d{2}_\d{2}-\d{2}-\d{2}(\.\w+)?$"         ...          self.extMatch = re.compile(self.extMatch, re.ASCII)         self.interval = self.interval * interval # multiply by units requested         filename = self.baseFilename         if os.path.exists(filename):             t = os.stat(filename)[ST_MTIME]  # 最后修改時(shí)間         else:             t = int(time.time())         self.rolloverAt = self.computeRollover(t)  # 提前計(jì)算終止時(shí)間      def computeRollover(self, currentTime):         # 判斷的方法還是很長(zhǎng)很復(fù)雜的，先pass      def shouldRollover(self, record):         t = int(time.time())         if t >= self.rolloverAt:  # 判斷是否到期             return 1         return 0      def doRollover(self):         ...         dfn = self.rotation_filename(self.baseFilename + "." +                                      time.strftime(self.suffix, timeTuple))         #  滾動(dòng)日志文件         if os.path.exists(dfn):             os.remove(dfn)         self.rotate(self.baseFilename, dfn)         if self.backupCount > 0:             for s in self.getFilesToDelete():                 os.remove(s)         ...         # 計(jì)算下一個(gè)時(shí)間點(diǎn)         newRolloverAt = self.computeRollover(currentTime)         ...         self.rolloverAt = newRolloverAt

日期滾動(dòng)就是計(jì)算最后時(shí)間點(diǎn)，超過(guò)時(shí)間點(diǎn)則重新生成新的日志文件。

小結(jié)

logging的處理邏輯大概是這樣的:

• 創(chuàng)建Logger對(duì)象，提供API，用來(lái)接收應(yīng)用程序日志

• Logger對(duì)象包括多個(gè)Handler

• 每個(gè)Handler有一個(gè)Formatter對(duì)象

• 每條日志都會(huì)生成一個(gè)LogRecord對(duì)象

• 使用不同的Handler對(duì)象將LogRecored對(duì)象提交到不同的流

• 每個(gè)日志對(duì)象通過(guò)Formatter格式化輸出

• 可以使用按日期/文件大小的方式進(jìn)行日志文件的滾動(dòng)記錄

小技巧

覆蓋對(duì)象的 __reduce__ 方法，讓對(duì)象支持 reduce 函數(shù):

class RootLogger(Logger):     def __init__(self, level):         Logger.__init__(self, "root", level)      def __reduce__(self):         return getLogger, ()

線程鎖的創(chuàng)建和釋放:

_lock = threading.RLock()  def _acquireLock():     if _lock:         _lock.acquire()  def _releaseLock():     if _lock:         _lock.release()

線程鎖的使用:

def addHandler(self, hdlr):     _acquireLock()     try:         self.handlers.append(hdlr)     finally:         _releaseLock()  def removeHandler(self, hdlr):     _acquireLock()     try:         self.handlers.remove(hdlr)     finally:         _releaseLock()

參考鏈接

• Logging in Python https://realpython.com/python-logging/

• 日志操作手冊(cè) https://docs.python.org/zh-cn/3.8/howto/logging-cookbook.html#cookbook-rotator-namer

• Python 的日志記錄工具 https://docs.python.org/zh-cn/3.8/library/logging.html

“怎么用Python實(shí)現(xiàn)強(qiáng)大的 logging 模塊”的內(nèi)容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業(yè)相關(guān)的知識(shí)可以關(guān)注創(chuàng)新互聯(lián)網(wǎng)站，小編將為大家輸出更多高質(zhì)量的實(shí)用文章！