Python怎么实现日志实时监测

这篇文章主要讲解了“Python怎么实现日志实时监测”，文中的讲解内容简单清晰，易于学习与理解，下面请大家跟着小编的思路慢慢深入，一起来研究和学习“Python怎么实现日志实时监测”吧！

介绍

观察者模式：是一种行为型设计模式。主要关注的是对象的责任，允许你定义一种订阅机制，可在对象事件发生时通知多个"观察"该对象的其他对象。用来处理对象之间彼此交互。

观察者模式也叫发布-订阅模式，定义了对象之间一对多依赖，当一个对象改变状态时，这个对象的所有依赖者都会收到通知并按照自己的方式进行更新。

观察者设计模式是最简单的行为模式之一。在观察者设计模式中，对象维护了一个依赖（观察者）列表，以便主题可以使用观察者定义的任何方法通知所有观察者它所发生的变化。

可使用观察者模式应用场景

在广播或者发布订阅系统的情形中，你会看到观察者设计模式的用法，它的主要使用场景如下：

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:', '地图加载失败！')

感谢各位的阅读，以上就是“Python怎么实现日志实时监测”的内容了，经过本文的学习后，相信大家对Python怎么实现日志实时监测这一问题有了更深刻的体会，具体使用情况还需要大家实践验证。这里是亿速云，小编将为大家推送更多相关知识点的文章，欢迎关注！