Python-logging详解（彩色日志扩展，多进程安全等）

曾经终败给现在 2022-09-15 14:57 106阅读 0赞

**目录**

简介

日志级别

记录器

处理器

格式器

多线程与多进程安全

代码

导入及全局变量

函数

实验及结果

参考

--------------------

# 简介 #

日志是工程中不可缺少的一部分，国家等保2.0也规定，至少保留日志180天。对于程序员来说，日志也方便进行记录及排错。

logging是Python自带的日志模块，主要有以下几个部分：

*  记录器暴露了应用程序代码直接使用的接口。
 *  处理器将日志记录（由记录器创建）发送到适当的目标。
 *  过滤器提供了更细粒度的功能，用于确定要输出的日志记录。
 *  格式器指定最终输出中日志记录的样式。

# 日志级别 #

<table style="width:600px;"> 日志级别 
 <thead> 
 <tr> 
 <th> 级别 </th> 
 <th> 数值 </th> 
 </tr> 
 </thead> 
 <tbody> 
 <tr> 
 <td> <code>CRITICAL</code> </td> 
 <td> 50 </td> 
 </tr> 
 <tr> 
 <td> <code>ERROR</code> </td> 
 <td> 40 </td> 
 </tr> 
 <tr> 
 <td> <code>WARNING</code> </td> 
 <td> 30 </td> 
 </tr> 
 <tr> 
 <td> <code>INFO</code> </td> 
 <td> 20 </td> 
 </tr> 
 <tr> 
 <td> <code>DEBUG</code> </td> 
 <td> 10 </td> 
 </tr> 
 <tr> 
 <td> <code>NOTSET</code> </td> 
 <td> 0 </td> 
 </tr> 
 </tbody> 
</table>

import logging之后可以从里面找到，用于区分日志记录的级别，通过级别可以确定是否输出。

# 记录器 #

`setLevel`(*level*)

给记录器设置阈值为 *level* 。日志等级小于 *level* 会被忽略。严重性为 *level* 或更高的日志消息将由该记录器的任何一个或多个处理器发出，除非将处理器的级别设置为比 *level* 更高的级别。

创建记录器时，级别默认设置为 `NOTSET。`

`addHandler`(*hdlr*)

将指定的处理器 *hdlr* 添加到此记录器。

`addFilter`(*filter*)

将指定的过滤器 *filter* 添加到此记录器。

`debug`(*msg*, *\*args*, *\*\*kwargs*)

在此记录器上记录 `DEBUG` 级别的消息。

`info`(*msg*, *\*args*, *\*\*kwargs*)

在此记录器上记录 `INFO` 级别的消息。

`warning`(*msg*, *\*args*, *\*\*kwargs*)

在此记录器上记录 `WARNING` 级别的消息。

`error`(*msg*, *\*args*, *\*\*kwargs*)

在此记录器上记录 `ERROR` 级别的消息。

`critical`(*msg*, *\*args*, *\*\*kwargs*)

在此记录器上记录 `CRITICAL` 级别的消息。

`log`(*level*, *msg*, *\*args*, *\*\*kwargs*)

在此记录器上记录 *level* 整数代表的级别的消息。

# 处理器 #

StreamHandler

它可将日志记录输出发送到数据流例如 *sys.stdout*, *sys.stderr* 或任何文件类对象（或者更精确地说，任何支持 `write()` 和 `flush()` 方法的对象）。

FileHandler

它可将日志记录输出到磁盘文件中。

# 格式器 #

`Formatter`(*fmt=None*, *datefmt=None*, *style='%'*, *validate=True*, *\**, *defaults=None*)

<table> 
 <thead> 
 <tr> 
 <th> 属性名称 </th> 
 <th> 格式 </th> 
 <th> 描述 </th> 
 </tr> 
 </thead> 
 <tbody> 
 <tr> 
 <td> args </td> 
 <td> 此属性不需要用户进行格式化。 </td> 
 <td> 合并到&nbsp;<code>msg</code>&nbsp;以产生&nbsp;<code>message</code>&nbsp;的包含参数的元组，或是其中的值将被用于合并的字典（当只有一个参数且其类型为字典时）。 </td> 
 </tr> 
 <tr> 
 <td> asctime </td> 
 <td> <code>%(asctime)s</code> </td> 
 <td> 被创建的供人查看时间值。 默认形式为 '2003-07-08 16:49:45,896' （逗号之后的数字为时间的毫秒部分）。 </td> 
 </tr> 
 <tr> 
 <td> created </td> 
 <td> <code>%(created)f</code> </td> 
 <td> 被创建的时间。 </td> 
 </tr> 
 <tr> 
 <td> exc_info </td> 
 <td> 此属性不需要用户进行格式化。 </td> 
 <td> 异常元组（例如&nbsp;<code>sys.exc_info</code>）或者如未发生异常则为&nbsp;<code>None</code>。 </td> 
 </tr> 
 <tr> 
 <td> filename </td> 
 <td> <code>%(filename)s</code> </td> 
 <td> <code>pathname</code>&nbsp;的文件名部分。 </td> 
 </tr> 
 <tr> 
 <td> funcName </td> 
 <td> <code>%(funcName)s</code> </td> 
 <td> 函数名包括调用日志记录. </td> 
 </tr> 
 <tr> 
 <td> levelname </td> 
 <td> <code>%(levelname)s</code> </td> 
 <td> 消息文本记录级别（<code>'DEBUG'</code>，<code>'INFO'</code>，<code>'WARNING'</code>，<code>'ERROR'</code>，<code>'CRITICAL'</code>）。 </td> 
 </tr> 
 <tr> 
 <td> levelno </td> 
 <td> <code>%(levelno)s</code> </td> 
 <td> 消息数字的记录级别 (<code>DEBUG</code>,&nbsp;<code>INFO</code>,&nbsp;<code>WARNING</code>,&nbsp;<code>ERROR</code>,&nbsp;<code>CRITICAL</code>). </td> 
 </tr> 
 <tr> 
 <td> lineno </td> 
 <td> <code>%(lineno)d</code> </td> 
 <td> 发出日志记录调用所在的源行号（如果可用）。 </td> 
 </tr> 
 <tr> 
 <td> message </td> 
 <td> <code>%(message)s</code> </td> 
 <td> 记入日志的消息，即&nbsp;<code>msg&nbsp;%&nbsp;args</code>&nbsp;的结果。 </td> 
 </tr> 
 <tr> 
 <td> module -- 模块 </td> 
 <td> <code>%(module)s</code> </td> 
 <td> 模块 (<code>filename</code>&nbsp;的名称部分)。 </td> 
 </tr> 
 <tr> 
 <td> msecs </td> 
 <td> <code>%(msecs)d</code> </td> 
 <td> 被创建的时间的毫秒部分。 </td> 
 </tr> 
 <tr> 
 <td> msg </td> 
 <td> 此属性不需要用户进行格式化。 </td> 
 <td> 在原始日志记录调用中传入的格式字符串。 与&nbsp;<code>args</code>&nbsp;合并以产生&nbsp;<code>message</code>，或是一个任意对象。 </td> 
 </tr> 
 <tr> 
 <td> name </td> 
 <td> <code>%(name)s</code> </td> 
 <td> 用于记录调用的日志记录器名称。 </td> 
 </tr> 
 <tr> 
 <td> pathname </td> 
 <td> <code>%(pathname)s</code> </td> 
 <td> 发出日志记录调用的源文件的完整路径名（如果可用）。 </td> 
 </tr> 
 <tr> 
 <td> process </td> 
 <td> <code>%(process)d</code> </td> 
 <td> 进程ID（如果可用） </td> 
 </tr> 
 <tr> 
 <td> processName </td> 
 <td> <code>%(processName)s</code> </td> 
 <td> 进程名（如果可用） </td> 
 </tr> 
 <tr> 
 <td> relativeCreated </td> 
 <td> <code>%(relativeCreated)d</code> </td> 
 <td> 以毫秒数表示的 LogRecord 被创建的时间，即相对于 logging 模块被加载时间的差值。 </td> 
 </tr> 
 <tr> 
 <td> stack_info </td> 
 <td> 此属性不需要用户进行格式化。 </td> 
 <td> 当前线程中从堆栈底部起向上直到包括日志记录调用并引发创建当前记录堆栈帧创建的堆栈帧信息（如果可用）。 </td> 
 </tr> 
 <tr> 
 <td> thread </td> 
 <td> <code>%(thread)d</code> </td> 
 <td> 线程ID（如果可用） </td> 
 </tr> 
 <tr> 
 <td> threadName </td> 
 <td> <code>%(threadName)s</code> </td> 
 <td> 线程名（如果可用） </td> 
 </tr> 
 </tbody> 
</table>

一般常用的就是时间，级别，内容，可以在团队内自行规定msg的格式作为补充。

# 多线程与多进程安全 #

logging是多线程安全的，所以添加多进程安全即可

# 代码 #

"""
    -*- coding:utf-8 -*-
    @File: log.py
    @Author:frank yu
    @DateTime: 2021.10.12 20:56
    @Contact: frankyu112058@gmail.com
    @Description:
    """
    import logging
    import os
    import random
    import sys
    import multiprocessing
    import time
    from shutil import copyfile
    from colorlog import ColoredFormatter
    
    # log directory
    LOG_DIR = "./logs"
    # level fot stdout
    STD_LEVEL = logging.WARN
    # level for file
    FILE_LEVEL = logging.DEBUG
    
    CRITICAL = 50
    FATAL = CRITICAL
    ERROR = 40
    WARNING = 30
    WARN = WARNING
    INFO = 20
    DEBUG = 10
    NOTSET = 0
    
    
    class Log:
        def __init__(self, log_dir=LOG_DIR):
            """
            function:init create log file, judge if log file is too bigger
            :param log_dir: directory of logs
            """
            log_path = log_dir + "/log"
            if not os.path.exists(log_dir):
                os.mkdir(log_dir)
                os.chmod(log_dir, 0o777)
            if not os.path.exists(log_path):
                f = open(log_path, mode='w', encoding='utf-8')
                f.close()
                os.chmod(log_path, 0o777)
            # if log file is more than 1MB, copy to a file and clear log file
            if os.path.getsize(log_path) / 1048576 > 1:
                # print(os.path.getsize(log_path))
                copyfile(log_path, log_dir + "/log" + str(time.time()).replace(".", ""))
                with open(log_path, 'w') as f:
                    f.truncate()
                    f.close()
            self.logger_format = logging.Formatter('%(asctime)s - %(levelname)s: %(message)s')
            self.c_logger_format = ColoredFormatter(fmt='%(log_color)s%(asctime)s'
                                                        ' - %(log_color)s%(levelname)s: %(log_color)s%(message)s',
                                                    reset=True,
                                                    secondary_log_colors={},
                                                    style='%'
                                                    )
            self.logger = logging.getLogger(str(random.random()))
            self.logger.handlers.clear()
            self.logger.setLevel(logging.DEBUG)
    
            self.filehandler = logging.FileHandler(log_path, mode='a')
            self.filehandler.setLevel(FILE_LEVEL)
            self.filehandler.setFormatter(self.logger_format)
    
            self.stdouthandler = logging.StreamHandler(sys.stdout)
            self.stdouthandler.setLevel(STD_LEVEL)
            self.stdouthandler.setFormatter(self.c_logger_format)
    
            self.logger.addHandler(self.stdouthandler)
            self.logger.addHandler(self.filehandler)
    
            self.__lock = multiprocessing.Lock()
    
        def __log(self, msg=None, level=logging.INFO):
            if level == logging.DEBUG:
                self.logger.debug(msg)
            elif level == logging.INFO:
                self.logger.info(msg)
            elif level == logging.WARNING:
                self.logger.warning(msg)
            elif level == logging.ERROR:
                self.logger.error(msg)
            elif level == logging.CRITICAL:
                self.logger.critical(msg)
    
        def log_show_store(self, msg, level):
            """
            function: show logs on screen and store logs into log file
            :param msg:message of log
            :param level:level of log
            return: None
            """
            self.__lock.acquire()
            self.logger.addHandler(self.stdouthandler)
            self.logger.addHandler(self.filehandler)
            self.__log(msg=msg, level=level)
            self.__lock.release()
    
        def log_show(self, msg, level):
            """
            function:show msg on the screen
            :param msg:message of log
            :param level: level of log
            """
            self.__lock.acquire()
            self.logger.removeHandler(self.filehandler)
            self.logger.addHandler(self.stdouthandler)
            self.__log(msg, level=level)
            self.logger.removeHandler(self.stdouthandler)
            self.logger.addHandler(self.filehandler)
            self.__lock.release()
    
        def log_store(self, msg, level):
            """
            function:store msg into log file
            :param msg:message of log
            :param level: level of log
            """
            self.__lock.acquire()
            self.logger.removeHandler(self.stdouthandler)
            self.logger.addHandler(self.filehandler)
            self.__log(msg, level=level)
            self.logger.removeHandler(self.filehandler)
            self.logger.addHandler(self.stdouthandler)
            self.__lock.release()
    
        def log_wrapper(self, *args):
            """
            function: record arg and result of functions
            :param args:
            :return: wrapper
            """
            self.log_store(' '.join(args), FILE_LEVEL)
    
            def wrapper(func):
                def inner(*args, **kwargs):
                    self.log_store(getattr(func, "__name__") + " call " + str(args) + str(kwargs), 10)
                    ret = func(*args, **kwargs)
                    self.log_store(getattr(func, "__name__") + " return " + str(ret), 10)
                    return ret
    
                return inner
    
            return wrapper

## 导入及全局变量 ##

colorlog需要安装一下，可以控制台输出彩色的

multiprocessing用于进程安全，加锁

设置了几个全局变量，方便修改

LOG\_DIR是日志保存的目录

STD\_LEVEL是控制台输出的级别，只输出WARN及以上的

FILE\_LEVEL是保存日志文件的级别，一般调试时使用DEBUG，生产环境时使用INFO

## 函数 ##

\_\_init\_\_

1.  创建目录和日志文件，如果不存在。若大于1M，复制为当前日期时间的日志文件
2.  设置了控制台和文件的格式器，控制台是彩色的
3.  设置了控制台和文件的处理器

log\_show\_store

保存到日志文件并打印到控制台

log\_show

仅打印到控制台

log\_store

仅保存到日志文件

log\_wrapper

日志装饰器，用于添加到重要函数上，记录函数调用参数及返回值

# 实验及结果 #

"""
    -*- coding:utf-8 -*-
    @File: main.py
    @Author:frank yu
    @DateTime: 2021.10.12 20:56
    @Contact: frankyu112058@gmail.com
    @Description:
    """
    from utils.log import Log, DEBUG, WARN, INFO
    from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
    
    log = Log()
    
    
    @log.log_wrapper("wrapper for test")
    def test(a, b, c=5):
        log.log_show('main test 1 ' + str(a), level=DEBUG)
        log.log_show('main test 2 ' + str(b), level=WARN)
        log.log_store('main test 3 ' + str(a), level=DEBUG)
        log.log_store('main test 4 ' + str(b), level=WARN)
        log.log_show_store('main test 5 ' + str(a + b), level=INFO)
        return a + b + c
    
    
    def even(num):
        log.log_store('main even 1 ' + str(num), level=INFO)
    
    
    def odd(num):
        log.log_store('main odd 1 ' + str(num), level=INFO)
    
    
    # multi thread test
    def multi_thread():
        exe = ThreadPoolExecutor(4)
        for i in range(10):
            if i & 1 == 0:
                exe.submit(even, i)
            else:
                exe.submit(odd, i)
    
    
    # multi process test
    def multi_process():
        exe = ProcessPoolExecutor(4)
        for i in range(10):
            if i & 1 == 0:
                exe.submit(even, i)
            else:
                exe.submit(odd, i)
    
    
    if __name__ == '__main__':
        test(3, 4, 6)
        multi_thread()
        multi_process()

msg这里传的是 **模块 函数 函数行数 消息** ，程序出错方便定位。

test测试了日志级别和日志装饰器

multi\_thread测试了多线程

multi\_process测试了多进程

结果如下：

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAbGFkeV9raWxsZXI5_size_20_color_FFFFFF_t_70_g_se_x_16][] 测试结果

可以看到控制台仅输出了WARNING级别，当然，如果有更高级别也会输出。

可以看到test函数的调用及参数和返回结果

可以看到多线程和多进程状态下结果没有丢失

# 参考 #

[logging-Python 的日志记录工具][logging-Python]

更多python相关内容：[【python总结】python学习框架梳理][python_python]

有问题请下方评论，转载请注明出处，并附有原文链接，谢谢！如有侵权，请及时联系。如果您感觉有所收获，自愿打赏，可选择支付宝18833895206（小于），您的支持是我不断更新的动力。

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAbGFkeV9raWxsZXI5_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/741e56ed60d64d47b108a405d3e03c83.png
[logging-Python]: https://docs.python.org/zh-cn/3/library/logging.html
[python_python]: https://blog.csdn.net/lady_killer9/article/details/108763489