python 线程安全的 单例 实现 日志分级

简介: python 线程安全的 单例 实现 日志分级

什么叫 单例 模式

多次初始化 只返回 同一个对象 叫做单例模式

为什么 要使用单例模式


  1. 代码 中 有些 对象 比如 日志对象 为了防止 多次初始化 可以使用单例模式
  2. 有些 模型 对象 体积庞大, 加载缓慢, 必须使用 单例模式

python 单例 线程安全的 实现方式

import threading
class SingletonType(type):
    _instance_lock = threading.Lock()
    def __call__(cls, *args, **kwargs):
        if not hasattr(cls, "_instance"):
            with SingletonType._instance_lock:
                if not hasattr(cls, "_instance"):
                    cls._instance = super(SingletonType,cls).__call__(*args, **kwargs)
        return cls._instance
class Foo(metaclass=SingletonType):
    def __init__(self,name):
        self.name = name
        print(name)
obj1 = Foo('name2')
obj2 = Foo('name1')
print(obj1,obj2)
# 测试结果 init 函数 只被执行了一次
# python3 singlon.py 
# name2
# <__main__.Foo object at 0x7f7cd9b04860> <__main__.Foo object at 0x7f7cd9b04860>

使用 单例 封装 日志对象

  1. 实现日志单例
  2. 实现日志 分级 输出文件
  3. 实现日志 按照 每周 分文件保存

Singleton.py 单例基类

#!/usr/bin/python
# -*- coding:utf-8 -*-
'''
@File    :   Singleton.py
@Time    :   2020/12/14 14:17:11
@Author  :   lmk
@Version :   1.0
'''
import threading
class SingletonType(type):
    _instance_lock = threading.Lock()
    def __call__(cls, *args, **kwargs):
        if not hasattr(cls, "_instance"):
            with SingletonType._instance_lock:
                if not hasattr(cls, "_instance"):
                    cls._instance = super(SingletonType,cls).__call__(*args, **kwargs)
        return cls._instance
if __name__ == '__main__':
    pass


singleton_log.py 每天 midnight 更换日志文件

#!/usr/bin/python
# -*- coding:utf-8 -*-
'''
@File    :   singleton_log.py
@Time    :   2020/12/14 14:26:05
@Author  :   lmk
@Version :   1.0
'''
from scripts.Singleton import SingletonType
from logging import Logger
import logging  # 引入logging模块
import logging.handlers
import os
import time
from gen_log.a_log_config import info_log_path,err_log_path
class singleLogger(metaclass=SingletonType):
    def __init__(self):
        # 第一步,创建一个logger
        self.logger = logging.getLogger()
        self.logger.setLevel(logging.INFO)  # Log等级总开关
        # 第二步,创建一个handler,用于写入日志文件
        # fh = logging.FileHandler(info_log_path,mode="w")
        info_fh = logging.handlers.TimedRotatingFileHandler(info_log_path,when="midnight")
        # fh.setLevel(logging.DEBUG)  # 输出到file的log等级的开关
        # 设置日志过滤器
        info_filter = logging.Filter()
        info_filter.filter = lambda record: record.levelno <= logging.WARNING # 设置过滤等级
        info_fh.setLevel(logging.INFO)
        info_fh.addFilter(info_filter)
        # fh = logging.FileHandler(info_log_path,mode="w")
        err_fh = logging.handlers.TimedRotatingFileHandler(err_log_path,when="midnight")
        # fh.setLevel(logging.DEBUG)  # 输出到file的log等级的开关
        err_filter = logging.Filter()
        err_filter.filter = lambda record: record.levelno > logging.WARNING
        err_fh.setLevel(logging.ERROR)
        err_fh.addFilter(err_filter)
        # 第三步,定义handler的输出格式
        formatter = logging.Formatter("%(asctime)s - %(filename)s[line:%(lineno)d] - %(levelname)s: \n%(message)s\n")
        info_fh.setFormatter(formatter)
        err_fh.setFormatter(formatter)
        # 第四步,将logger添加到handler里面
        self.logger.addHandler(info_fh)
        self.logger.addHandler(err_fh)
logger = singleLogger().logger

a_log_config.py 日志配置

#!/usr/bin/python
# -*- coding:utf-8 -*-
'''
@File    :   a_log_config.py
@Time    :   2020/12/14 13:28:02
@Author  :   lmk
@Version :   1.0
'''
import os 
import time
opdn = os.path.dirname
dirpath = opdn(opdn(__file__))
logs_dir_path = os.path.join(dirpath,"Logs")
# 消息日志路径
# info_log_name = time.strftime('info_%Y%m%d.log', time.localtime(time.time()))
info_log_name = time.strftime('info.log', time.localtime(time.time()))
info_log_path = os.path.join(logs_dir_path,info_log_name)
# 错误日志路径
# err_log_name = time.strftime('err_%Y%m%d.log', time.localtime(time.time()))
err_log_name = time.strftime('err.log', time.localtime(time.time()))
err_log_path = os.path.join(logs_dir_path,err_log_name)
if __name__ == '__main__':
    pass


最后可以项目 入口 将异常写入 日志

main.py

s = traceback.format_exc()

logger.error(s)


将异常调用栈 信息 取出并 记录到日志

# 开始 定时 处理授权文件
@my_decorator.while_do
def grant_auth():
    try:
        global count
        count+=1
        deal_temp_grant()
        raise(Exception("err {}".format(count)))
    except:
        s = traceback.format_exc()
        logger.error(s)


相关实践学习
【涂鸦即艺术】基于云应用开发平台CAP部署AI实时生图绘板
【涂鸦即艺术】基于云应用开发平台CAP部署AI实时生图绘板
相关文章
|
2月前
|
监控 安全 程序员
Python日志模块配置:从print到logging的优雅升级指南
从 `print` 到 `logging` 是 Python 开发的必经之路。`print` 调试简单却难维护,日志混乱、无法分级、缺乏上下文;而 `logging` 支持级别控制、多输出、结构化记录,助力项目可维护性升级。本文详解痛点、优势、迁移方案与最佳实践,助你构建专业日志系统,让程序“有记忆”。
256 0
|
5月前
|
数据采集 存储 JSON
Python爬取知乎评论:多线程与异步爬虫的性能优化
Python爬取知乎评论:多线程与异步爬虫的性能优化
|
5月前
|
人工智能 安全 调度
Python并发编程之线程同步详解
并发编程在Python中至关重要,线程同步确保多线程程序正确运行。本文详解线程同步机制,包括互斥锁、信号量、事件、条件变量和队列,探讨全局解释器锁(GIL)的影响及解决线程同步问题的最佳实践,如避免全局变量、使用线程安全数据结构、精细化锁的使用等。通过示例代码帮助开发者理解并提升多线程程序的性能与可靠性。
200 0
|
2月前
|
Java 调度 数据库
Python threading模块:多线程编程的实战指南
本文深入讲解Python多线程编程,涵盖threading模块的核心用法:线程创建、生命周期、同步机制(锁、信号量、条件变量)、线程通信(队列)、守护线程与线程池应用。结合实战案例,如多线程下载器,帮助开发者提升程序并发性能,适用于I/O密集型任务处理。
290 0
|
4月前
|
数据采集 消息中间件 并行计算
Python多线程与多进程性能对比:从原理到实战的深度解析
在Python编程中,多线程与多进程是提升并发性能的关键手段。本文通过实验数据、代码示例和通俗比喻,深入解析两者在不同任务类型下的性能表现,帮助开发者科学选择并发策略,优化程序效率。
304 1
|
4月前
|
安全 JavaScript Java
Python中None与NoneType的真相:从单例对象到类型系统的深度解析
本文通过10个真实场景,深入解析Python中表示“空值”的None与NoneType。从单例模式、函数返回值,到类型注解、性能优化,全面揭示None在语言设计与实际编程中的核心作用,帮助开发者正确高效地处理“无值”状态,写出更健壮、清晰的Python代码。
452 3
|
5月前
|
数据采集 监控 调度
干货分享“用 多线程 爬取数据”:单线程 + 协程的效率反超 3 倍,这才是 Python 异步的正确打开方式
在 Python 爬虫中,多线程因 GIL 和切换开销效率低下,而协程通过用户态调度实现高并发,大幅提升爬取效率。本文详解协程原理、实战对比多线程性能,并提供最佳实践,助你掌握异步爬虫核心技术。
|
6月前
|
JSON 算法 Java
打造终端里的下载利器:Python实现可恢复式多线程下载器
在数字时代,大文件下载已成为日常需求。本文教你用Python打造专业级下载器,支持断点续传、多线程加速、速度限制等功能,显著提升终端下载体验。内容涵盖智能续传、多线程分块下载、限速控制及Rich库构建现代终端界面,助你从零构建高效下载工具。
401 1
|
5月前
|
数据采集 存储 Java
多线程Python爬虫:加速大规模学术文献采集
多线程Python爬虫:加速大规模学术文献采集

热门文章

最新文章

推荐镜像

更多