python多进程日志以及分布式日志的实现方式

本文涉及的产品
日志服务 SLS,月写入数据量 50GB 1个月
简介: python日志在多进程环境下的问题python日志模块logging支持多线程,但是在多进程下写入日志文件容易出现下面的问题:PermissionError: [WinError 32] 另一个程序正在使用此文件,进程无法访问。也就是日志文件被占用的情况,原因是多个进程的文件handler对日志文件进行操作产生的。

python日志在多进程环境下的问题

python日志模块logging支持多线程,但是在多进程下写入日志文件容易出现下面的问题:


PermissionError: [WinError 32] 另一个程序正在使用此文件,进程无法访问。


也就是日志文件被占用的情况,原因是多个进程的文件handler对日志文件进行操作产生的。


这个问题经常在TimedRotatingFileHandler、RotatingFileHandler中出现。

解决办法

题主在网上搜集了各种解决上面问题的办法,基本以下面三个方向为主:

  • 安装第三方库提供的handler
  • 重写filehandler加全局锁
  • 使用队列将消息传递


但是三种方法各有小缺陷:

  • 第三方库很久无人维护,且支持的功能比较单一,无法满足生产环境的需求。
  • 轮转日志的时候由于全局锁的存在,其他子进程无法记录日志,有丢失日志的风险。
  • 使用多进程消息队列的缺点在于使用困难,如果是多模块编程,需要将全局队列传来传去,在大型项目中显得很麻烦。


经过对官网的研究 ,题主无意中找到了一种非常方便且高效的方法,并且经过一定的修改使这种方法可用于分布式日志,且支持多语言日志的处理。

唯一的不足是需要新学习一个zmq通信协议,但是这并不是问题,如果只是想要一个解决方案并立即投入使用,只需要按照下面的方法编写,无需关注zmq的相关知识。

基于zmq的分布式日志

实现思路

  • 通过zmq的多对一通信,将多个地方的日志发送到一个地方集中处理,从而实现分布式日志。
  • 这个方法不仅可以解决python分布日志的问题,还可以很好的兼容其他语言,比如项目中还有C、java,那么可以将它们中的日志也发送过来,一并处理。

看到这很多人可能明白了,这个方法类似官网提供的SocketHandler,但本方法其实是基于QueueHandler实现的,有利于发挥zmq易用性、可插拔、并发性能好的优点。

代码实现

首先是集中处理日志的程序,也就是上面所说"多对一"中的一

import zmq
import logging
from logging import handlers
 
class ZeroMQSocketListener(handlers.QueueListener):
    def __init__(self, uri="tcp://127.0.0.1:5555", *handlers,**kwargs):
        self.respect_handler_level = True     # handler日志等级启用,允许对handler设置setLevel,False则忽视级别
        self.ctx = kwargs.get('ctx') or zmq.Context()
        socket = self.ctx.socket(zmq.SUB)
        socket.bind(uri)
        socket.setsockopt_string(zmq.SUBSCRIBE, '')     # 订阅所有主题
        super().__init__(socket, *handlers, respect_handler_level=self.respect_handler_level)
 
    def dequeue(self,block):
        msg = self.queue.recv_json()
        # print('111',msg)    # 测试用
        return logging.makeLogRecord(msg)
 
def main_logger():
    # 日志集中处理区,在主程序中调用一次
    
    # handlers配置区,filter可选
    formatter = logging.Formatter("%(name)s - %(asctime)s - %(levelname)s - %(module)s - %(funcName)s - %(message)s")
    console = logging.StreamHandler()
    console.setLevel(logging.ERROR)
    ch = handlers.TimedRotatingFileHandler(r'logs\face.log',when='M',
                                           # backupCount=180,
                                           encoding='utf-8')
    ch.setLevel(logging.INFO)
    ch.setFormatter(formatter)  # add formatter to ch
    
    # 设置监听的端口,并传递handlers
    loggerListener = ZeroMQSocketListener("tcp://127.0.0.1:5555",*(ch,console))
    loggerListener.start()   # 开启一个子线程处理记录器监听
    
# 主进程调用一次,非阻塞
main_logger()

自此,日志集中处理就结束了,是不是很简单,而且需要注意,我们这里不需要用到root logger,因为ZeroMQSocketListener会自动调用各种handlers将日志内容进行处理,想当于替代了logger的工作,所以也就没必要声明一个logger出来了。


更新:

这里的main_logger()是非阻塞,也就是下面还可以写其他代码,但是如果什么代码都没有,那么主进程就会直接退出,日志就收不到了。

如果接下来不需要做其他工作,那么请在main_logger()下方使用while True:time.sleep(0.5) 将主进程阻塞。


  • 需要重点关注通信地址"tcp://127.0.0.1:5555",因为其他地方的日志都会发送到这里来。


接下来是子进程中或者是你想记录日志的任何地方,比如在其他同事的电脑里

  • subprocess.py
import logging,zmq
from logging import handlers
 
# 我们需要的handler
class ZeroMQSocketHandler(handlers.QueueHandler):
    def __init__(self, uri="tcp://127.0.0.1:5555", socktype=zmq.PUB, ctx=None):
        self.ctx = ctx or zmq.Context()
        socket = self.ctx.socket(socktype)
        socket.connect(uri)
        super().__init__(socket)
    def enqueue(self, record):
        self.queue.send_json(record.__dict__)
    def close(self):
        self.queue.close()
        
# 创建远端日志
rmtlogger = logging.getLogger('sub_root_name')    ##
rmtlogger.setLevel(logging.INFO)     # 建议设置一下,有时候默认是WARNING级别
rmtlogger.propagate=False    # 不允许传递,日志传递到这里就发送到主进程中
 
# 配置handler
zmqhandler = ZeroMQSocketHandler()
zmqhandler.setLevel(logging.INFO)
rmtlogger.addHandler(zmqhandler)
 
# if you have submodule
# import submodule  
 
# 记录日志
rmtlogger.info("这是一条遥远的日志")
  • 如果是多进程环境下,您大可直接将上面的代码直接开启到多个子进程中,并不会出现网络问题。


logger可以通过python日志的name系统进行传递,也就是说如果子进程中还有其他模块,可以通过日志传递系统将其他模块产生的日志传递过来,最后一并发送给监听器,就像下面:

  • submodule.py
# subprocess.py的子模块,如需测试注意调用
import logging
subMolduleLogger = logging.getLogger(f'sub_root_name.modulename')
 
subMolduleLogger.info("这是一条子模块日志")
# 这部分内容需要logging基础知识
  • 上面这条日志会传递给rmtlogger,通过rmtlogger发送到主进程。


在主进程中,设置了logging.Formatter对象,可以将产生日志的名字打印出来,用于区分日志产生的位置。

多语言支持

由于zmq本身就支持多语言,比如你使用c语言或其他语言,只需要在代码中使用zmq将日志通过json发送过来,python日志可以通过dict方法重建logger对象,具体可以打印上面代码中ZeroMQSocketListener.dequeue中的msg进行摸索,实现起来还是比较简单的。

相关实践学习
日志服务之使用Nginx模式采集日志
本文介绍如何通过日志服务控制台创建Nginx模式的Logtail配置快速采集Nginx日志并进行多维度分析。
相关文章
|
7天前
|
分布式计算 并行计算 算法
探索排序的宇宙奥秘:Python中归并排序的并行处理与分布式应用!
【7月更文挑战第11天】归并排序是一种分治算法,适用于并行和分布式处理。在Python中,利用`concurrent.futures`可实现并行归并排序,但因GIL限制,可能需借助`multiprocessing`或GPU库。分布式归并排序则通过分布式框架如Apache Spark处理大规模数据,每个节点独立排序后进行网络合并。并行与分布式技术提升了处理大数据的速度和效率。**
17 9
|
4天前
|
消息中间件 安全 数据处理
Python中的并发编程:理解多线程与多进程的区别与应用
在Python编程中,理解并发编程是提高程序性能和响应速度的关键。本文将深入探讨多线程和多进程的区别、适用场景及实际应用,帮助开发者更好地利用Python进行并发编程。
|
5天前
|
Unix Linux Python
`subprocess`模块是Python中用于生成新进程、连接到它们的输入/输出/错误管道,并获取它们的返回(退出)代码的模块。
`subprocess`模块是Python中用于生成新进程、连接到它们的输入/输出/错误管道,并获取它们的返回(退出)代码的模块。
|
5天前
|
Python
在Python中,`multiprocessing`模块提供了一种在多个进程之间共享数据和同步的机制。
在Python中,`multiprocessing`模块提供了一种在多个进程之间共享数据和同步的机制。
|
5天前
|
安全 API Python
`multiprocessing`是Python的一个标准库,用于支持生成进程,并通过管道和队列、信号量、锁和条件变量等同步原语进行进程间通信(IPC)。
`multiprocessing`是Python的一个标准库,用于支持生成进程,并通过管道和队列、信号量、锁和条件变量等同步原语进行进程间通信(IPC)。
|
5天前
|
Unix Python
Python代码示例:使用`syslog`模块进行日志记录
Python代码示例:使用`syslog`模块进行日志记录
|
5天前
|
Python
Python的`signal`模块提供了访问底层操作系统提供的信号机制的方式。信号是操作系统用来通知进程发生了某种情况(如用户按下Ctrl+C)的一种机制。
Python的`signal`模块提供了访问底层操作系统提供的信号机制的方式。信号是操作系统用来通知进程发生了某种情况(如用户按下Ctrl+C)的一种机制。
|
10天前
|
NoSQL Java Redis
实现基于Redis的分布式锁机制
实现基于Redis的分布式锁机制
|
23天前
|
NoSQL Redis
redis分布式锁redisson
底层会尝试去加锁,如果加锁失败,会睡眠,自旋加锁,直到获取到锁为止。
22 1
|
20天前
|
消息中间件 NoSQL Java
Redis系列学习文章分享---第六篇(Redis实战篇--Redis分布式锁+实现思路+误删问题+原子性+lua脚本+Redisson功能介绍+可重入锁+WatchDog机制+multiLock)
Redis系列学习文章分享---第六篇(Redis实战篇--Redis分布式锁+实现思路+误删问题+原子性+lua脚本+Redisson功能介绍+可重入锁+WatchDog机制+multiLock)
50 0