CMDB02/单例模式、资产采集参考命令、日志处理
目录
1. 单例模式
1.1 多例模式
-
代码示例
class Foo(object): def __init__(self,name,age): self.name = name self.age = age def func(self): msg = "%s-%s" %(self.name,self.age) obj1 = Foo("张三",18) # Foo的一个对象/Foo类的一个实例 obj1.func() obj2 = Foo("李四",19) obj2.func()
1.2 单例模式
-
代码示例/以前:
class Foo(object): instance = None def __init__(self,name,age): self.name = name self.age = age def __new__(cls,*arg,**kawrgs): if not cls.instance: cls.instance = object.__new__(cls) return cls.instance
class Singleton(object): instance = None def __init__(self): self.name = None def __new__(cls, *arg, **kawrgs): if not cls.instance: cls.instance = object.__new__(cls) return cls.instance
-
应用场景:
-
django配置文件,只要加载一次,以后使用都用同一份值。
class Singleton(object): instance = None def __init__(self): self.k0 = 0 self.k1 = 1 self.k2 = 2 ... def __new__(cls, *arg, **kawrgs): if not cls.instance: cls.instance = object.__new__(cls) return cls.instance obj1 = Singleton() obj2 = Singleton() obj3 = Singleton()
-
django的admin,在注册models中用,希望所有的model类注册到同一个列表中。
-
1.2.1 单例模式/错误
-
代码示例:
class Singleton(object): instance = None def __init__(self): self.name = None def __new__(cls, *arg, **kawrgs): if not cls.instance: cls.instance = object.__new__(cls) return cls.instance
1.2.2 单例模式 new/正确
-
代码示例:
import time import threading class Singleton(object): instance = None lock = threading.RLock() def __new__(cls, *arg, **kwargs): if cls.instance: return cls.instance with cls.lock: if not cls.instance: cls.instance = object.__new__(cls) return cls.instance obj1 = Singleton() obj2 = Singleton()
1.2.3 单例模式之文件导入/正确,在源码中的应用
-
代码示例1:
# xx.py class Site(object): def __init__(self): self.names = [] def xx(self): pass site = Site()
import xx print(xx.site)
-
代码示例2:
# xx.py class Singleton(object): def __init__(self): self._registry = [] def register(self,model_class): self._registry.append(model_class) site = Singleton()
import xx xx.site
1.3 单例模式补充
-
基于
__new__
实现单例模式,在init中不设置值。 -
是单例,但数据会被覆盖
class Singleton(object): instance = None def __init__(self): self.registry = [] def __new__(cls, *arg, **kawrgs): if not cls.instance: cls.instance = object.__new__(cls) return cls.instance def register(self, val): self.registry.append(val) # {registry:[]} obj1 = Singleton() # {registry:[x1]} obj1.register('x1') # instance = {registry:[x1]} # instance = {registry:[]} obj2 = Singleton() # instance = {registry:[x2]} obj2.register('x2') print(obj2.registry) print(obj1.registry)
-
单例模式/数据不会覆盖
class Singleton(object): instance = None registry = [] def __new__(cls, *arg, **kawrgs): if not cls.instance: cls.instance = object.__new__(cls) return cls.instance def register(self, val): self.registry.append(val) obj1 = Singleton() obj1.register('x1') obj2 = Singleton() obj2.register('x2') print(obj2.registry) print(obj1.registry)
2. 日志
-
代码示例:
import logging import settings class AutoLogger(object): def __init__(self,log_path,log_name): file_handler = logging.FileHandler(log_path, 'a', encoding='utf-8') fmt = logging.Formatter(fmt="%(asctime)s - %(name)s - %(levelname)s -%(module)s: %(message)s") file_handler.setFormatter(fmt) self.logger = logging.Logger(log_name, level=logging.DEBUG) self.logger.addHandler(file_handler) def log(self,msg): self.logger.error(msg) logger = AutoLogger(settings.LOG_FILE_PATH,'cmdb')
-
获取日志堆栈信息
import traceback def run(): try: int('lbzhk') except Exception as e: print(traceback.format_exc()) if __name__ == '__main__': run()
3. 获取资产信息的命令
-
内存信息
sudo dmidecode -q -t 17 2>/dev/null # 注意:linux上要提前安装 yum install dmidecode
-
硬盘(安装MegaCli)
sudo MegaCli -PDList -aALL
-
网卡
sudo ip link show sudo ip addr show
-
主板
sudo dmidecode -t1
-
CPU
cat /proc/cpuinfo
4. 数据封装
-
代码示例:
class BaseResponse(object): def __init__(self): self.status = True self.data = None self.error = None @property def dict(self): return self.__dict__ def process(): info = BaseResponse() try: info.status = True info.data = "lbzhk" except Exception: pass return info.dict result = process() print(result)
5.表关系
总结:
-
cmdb资产采集后,为什么不直接放到数据库?
- 单独编写api,为了给其他系统提供数据支持(接口)
- 维护的数据库连接比较多,修改不方便。
-
手写单例模式
- new+锁
-
其他单例模式
-
使用模块
class Singleton(object): def foo(self): pass singleton = Singleton()
from a import singleton
-
使用装饰器
def Singleton(cls): _instance = {} def _singleton(*args, **kargs): if cls not in _instance: _instance[cls] = cls(*args, **kargs) return _instance[cls] return _singleton @Singleton class A(object): a = 1 def __init__(self, m=0): self.m = m a1 = A(1) a2 = A(2)
-
使用类 -- 可以防止多线程的时候出问题
import time import threading class Singleton(object): _instance_lock = threading.Lock() def __init__(self): time.sleep(1) @classmethod def instance(cls, *args, **kwargs): with Singleton._instance_lock: if not hasattr(Singleton, "_instance"): Singleton._instance = Singleton(*args, **kwargs) return Singleton._instance def task(arg): obj = Singleton.instance() print(obj) for i in range(10): t = threading.Thread(target=task,args=[i,]) t.start() time.sleep(20) obj = Singleton.instance() print(obj)
-
基于
__new__
实现 ,代码如上
-
-
__new__
方法返回的是什么- 新创建的对象,内部没有数据,需要经过init来进行初始化。
-
单利模式应用场景:
- django的配置文件
- django的admin
- 数据库链接池(单例模式)
- 写日志