Python实现日志实时监测的示例详解
作者:Brook 发布时间:2023-01-06 15:48:29
介绍
观察者模式:是一种行为型设计模式。主要关注的是对象的责任,允许你定义一种订阅机制,可在对象事件发生时通知多个"观察"该对象的其他对象。用来处理对象之间彼此交互。
观察者模式也叫发布-订阅模式,定义了对象之间一对多依赖,当一个对象改变状态时,这个对象的所有依赖者都会收到通知并按照自己的方式进行更新。
观察者设计模式是最简单的行为模式之一。在观察者设计模式中,对象维护了一个依赖(观察者)列表,以便主题可以使用观察者定义的任何方法通知所有观察者它所发生的变化。
可使用观察者模式应用场景
在广播或者发布订阅系统的情形中,你会看到观察者设计模式的用法,它的主要使用场景如下:
1、分布式系统中实现事件服务。
2、广播或发布/阅系统情形中。
2、用作新闻机器的框架。
3、股票监测机器人。
观察者模式类图
观察者模式类图
1、发布者Publisher:向其他对象发送值得关注的事件。事件会在发布者自身状态改变或执行特定行为后发生。发布者中包含一个允许新订阅者加入和当前订阅者离开列表的订阅机制。
2、订阅者Subscriber:定义通知接口。一般情况下,该接口仅包含一个update()更新方法。方法中可以有多个参数,使发布者能在更新时传递事件详细信息。
3、客户端Client:分别创建发布者和订阅者对象,然后为订阅者注册,发布者更新。
观察者模式示例
假如我们对应用函数运行状态进行监测,当发生异常时报警记录,可通过观察者模式进行信息订阅:1、短信 2、日志 3、邮件
代码实现---subscription_model.py
1、创建订阅者类
Subscriber订阅者:所有希望关注发布者状态变化的其他对象。
这里提供了三个主要的订阅者(观察者)接口,跟踪着同一个发布者类的事件。主要包括:
1)、每个具体订阅者__init()方法使用attach()方法向发布者进行注册以获取信息更新。
2)、具体订阅者的update()更新消息。
#抽象订阅者
from abc import ABCMeta,abstractmethod
class Subscriber(metaclass=ABCMeta):
#向具体订阅者发送消息的方法
@abstractmethod
def update(self):
pass
#具体订阅者
#1、短信订阅者
class SMSSubscriber(Subscriber):
def __init__(self,publisher):
self.publisher = publisher
self.publisher.attach(self)
def update(self):
print(type(self).__name__,self.publisher.getNews())
#2、邮件订阅者
class EmailSubscriber(Subscriber):
def __init__(self, publisher):
self.publisher = publisher
self.publisher.attach(self)
def update(self):
print(type(self).__name__,self.publisher.getNews())
info = self.publisher.getNews()
# 发送邮件
Sender_mail(info).sender_mail()
#3、日志订阅(文件存储)
class LoggerSubscriber(Subscriber):
def __init__(self, publisher):
log_dir = os.path.expanduser(r".\apps\Mapview\logs")
log_file = os.path.join(log_dir, "file_{time}.log")
logger.add(log_file, rotation="100KB", retention=2)
self.publisher = publisher
self.publisher.attach(self)
def update(self):
print(type(self).__name__,self.publisher.getNews())
info=self.publisher.getNews()
logger.info(f"{info}")
2、创建发布者类
Publisher发布者:将自身的状态改变通知其他对象,为发布者添加订阅机制,每个对象都能订阅或取消订阅者事件流。
主要包括:
1)self.__subscribers = []:一个用于存储订阅对象列表
2)供订阅者来注册NewsPublisher或删除订阅用户。
3)几个用于添加、删除或查看列表中订阅者的公有方法。
4)notifySubscribers(self):用于通知所有订阅者出现新的信息,发送者会遍历订阅列表并通过内部调用具体订阅者实现的update()方法来实现。
5)创建新消息和返回最新消息。
#创建发布者
class NewsPublisher:
def __init__(self):
self.__subscribers = []
self.__latestNews = None
#将订阅者添加到队列中
def attach(self,subscriber):
self.__subscribers.append(subscriber)
#从订阅的主题里面移除
def detach(self):
return self.__subscribers.pop()
#生成观察者列表
def subscribers(self):
return [type(x).__name__ for x in self.__subscribers]
#发送通知给相关的主题订阅者
def notifySubscribers(self):
for sub in self.__subscribers:
#update()方法由具体的观察者或订阅者实现的
sub.update() #推送更新
#创建新消息
def addNews(self,news):
self.__latestNews = news
#返回最新消息,并通知观察者
def getNews(self):
return "Got News:",self.__latestNews
3、应用客户端-Map_server_client.py
订阅者通常需要一些上下文信息正确处理更新。因此,发布者通常会将一些上下文数据作为通知方法的参数传递。
这里给第一篇文章留下的尾巴补充一下,客户端实例化get_Map_model方法添加带参数装饰器,@fail_data(msg='地图加载失败')添加接口调用失败处理机制,追加日志记录。这里可以进一步将更多细节参数添加到日志中,装饰器传参并在接口中声明通知方法及参数,这样发布者在发出通知时传递一些上下文数据。
from apps.tools.subscription_model import NewsPublisher,LoggerSubscriber,EmailSubscriber
import functools
#如果加载失败,调用订阅者
def publisher(info):
news_publisher = NewsPublisher()
# for Subscribers in [EmailSubscriber, LoggerSubscriber]:
for Subscribers in [LoggerSubscriber]:
# eval(LoggerSubscriber)(news_publisher)
Subscribers(news_publisher)
print("\nSubscribers", news_publisher.subscribers())
news_publisher.addNews(f"{info}")
news_publisher.notifySubscribers()
#处理异常的装饰器
def fail_data(msg='地图加载失败'):
def catch_exception(origin_func):
@functools.wraps(origin_func)
def wrapper(*args, **kwargs):
try:
u = origin_func(*args, **kwargs)
print("这个函数正常执行:%s" % origin_func.__name__)
return u
except Exception as e:
info = f"{msg}:{e.__doc__}"
"""
接口调用失败处理机制,追加日志
"""
print(info)
publisher(info)
# news_publisher = NewsPublisher()
# LoggerSubscriber(news_publisher)
# print("\nSubscribers", news_publisher.subscribers())
# news_publisher.addNews(f"{info}")
# news_publisher.notifySubscribers()
return wrapper
return catch_exception
4、测试
if __name__ == '__main__':
from loguru import logger
from apps.tools.Sender_Email import Sender_mail
news_publisher =NewsPublisher()
for Subscribers in [LoggerSubscriber]:
print(Subscribers)
Subscribers(news_publisher)
print("\nSubscribers",news_publisher.subscribers())
news_publisher.addNews("地图加载失败!")
news_publisher.notifySubscribers()
结果
class '__main__.LoggerSubscriber';
Subscribers ['LoggerSubscriber']
LoggerSubscriber ('Got News:', '地图加载失败!')
2022-04-05 16:38:00.667 | INFO | __main__:update:81 - ('Got News:', '地图加载失败!')
来源:https://mp.weixin.qq.com/s/RZAz2_2rWIwfFYGSB_MmJQ


猜你喜欢
- 在刚进公司的时候,要写一个需求,使用django的admin站点管理,实现一个二级联动的功能,因为要用到django自带的页面,因为不是自定
- 需求是这样的,我从本科到现在硬盘里存了好多照片,本来是按类别分的,有一天,我突然想,要是能按照时间来分类可能会更好。可以右键查看照片的属性,
- 问题描述有时候,产品让我们做的表格,会有合并列的功能,但是官方的demo略有不清晰,本文举个例子简述之。我们先看下效果图:假设产品的需求是这
- demo中的p中的script改大或改小看看。。。移动到top的时间始终是一定的,你也可以设置一个阀值,在页面高度到达这个阀值之前,移动的总
- 用程序来处理图像和办公文档经常出现在实际开发中,Python的标准库中虽然没有直接支持这些操作的模块,但我们可以通过Python生态圈中的第
- 前言:HTML5和CSS3的时代到来了,新版2011版淘宝网首页已全部使用HTML5,拥抱变化才是王道。为之漫笔翻译的很好,看了一遍后,感觉
- 一组常用的弹出窗口用法,以下代码集合常用的弹出窗口用法。1、最基本的弹出窗口代码<SCRIPT LANGUAGE="
- 一个普通WEB站点的页面常常需要查询N条SQL语句后才能得出页面结果,当网站访问速度慢而前端做了大量优化工作以后,数据库瓶颈的查找也是WEB
- 本文介绍了多个 Python IDE,并评价其优缺点。读者可以参考此文列举的 Python IDE 列表,选择适合自己的编辑器。写 Pyth
- 目的:删除文件之后,如果目录为空,递归删除为空的目录。svr.dataDir指目录的最外层,递归删除目录的时候判断到这一层即可。eg:/Us
- 1.创建一个项目django-admin.py startproject HelloWorld2.进入HelloWorld项目,在manag
- 1.相关基础python3中有两种字符串类型:str和bytespython编码问题可以参考文章str以unicode编码格式保存在内存所以
- 今天把Ext.js源码又读了一遍,不过这次比较认真。看完代码,有了不少收获也遇到不少问题。主要总结如下:1、document.execCom
- 前言本博客默认读者对神经网络与Tensorflow有一定了解,对其中的一些术语不再做具体解释。并且本博客主要以图片数据为例进行介绍,如有错误
- 自从python2.2提供了yield关键字之后,python的生成器的很大一部分用途就是可以用来构建协同程序,能够将函数挂起返回中间值并能
- 上篇文章讲了js中的传值和传址 和 函数的作用域.这章我们来探讨js中的变量,表达式,和运算符 还有一些 js 语句。升级中……1, 表达式
- 本文实例为大家分享了python实现简单五子棋游戏的具体代码,供大家参考,具体内容如下from graphics import *from
- vscode简介什么是vscode简单来说,vscode是一款文本编辑器,而不是ide。这就是说,vscode只提供编辑的环境而不提供编译的
- 本文实例讲述了Python同时向控制台和文件输出日志logging的方法。分享给大家供大家参考。具体如下:python提供了非常方便的日志模
- 前言pytest是一款强大的python自动化测试工具,可以胜任各种类型或者级别的软件测试工作。pytest提供了丰富的功能,包括asser