python多進(jìn)程日志以及分布式日志的實(shí)現(xiàn)方式

更新時(shí)間：2024年06月28日 17:28:29 作者：brandon_l

這篇文章主要介紹了python多進(jìn)程日志以及分布式日志的實(shí)現(xiàn)方式,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教

python日志在多進(jìn)程環(huán)境下的問題

python日志模塊logging支持多線程，但是在多進(jìn)程下寫入日志文件容易出現(xiàn)下面的問題：

PermissionError: [WinError 32] 另一個(gè)程序正在使用此文件，進(jìn)程無法訪問。

也就是日志文件被占用的情況，原因是多個(gè)進(jìn)程的文件handler對(duì)日志文件進(jìn)行操作產(chǎn)生的。

這個(gè)問題經(jīng)常在TimedRotatingFileHandler、RotatingFileHandler中出現(xiàn)。

解決辦法

題主在網(wǎng)上搜集了各種解決上面問題的辦法，基本以下面三個(gè)方向?yàn)橹鳎?/p>

安裝第三方庫提供的handler
重寫filehandler加全局鎖
使用隊(duì)列將消息傳遞

但是三種方法各有小缺陷：

第三方庫很久無人維護(hù)，且支持的功能比較單一，無法滿足生產(chǎn)環(huán)境的需求。
輪轉(zhuǎn)日志的時(shí)候由于全局鎖的存在，其他子進(jìn)程無法記錄日志，有丟失日志的風(fēng)險(xiǎn)。
使用多進(jìn)程消息隊(duì)列的缺點(diǎn)在于使用困難，如果是多模塊編程，需要將全局隊(duì)列傳來傳去，在大型項(xiàng)目中顯得很麻煩。

經(jīng)過對(duì)官網(wǎng)的研究¹，題主無意中找到了一種非常方便且高效的方法，并且經(jīng)過一定的修改使這種方法可用于分布式日志，且支持多語言日志的處理。

唯一的不足是需要新學(xué)習(xí)一個(gè)zmq通信協(xié)議，但是這并不是問題，如果只是想要一個(gè)解決方案并立即投入使用，只需要按照下面的方法編寫，無需關(guān)注zmq的相關(guān)知識(shí)。

基于zmq的分布式日志

實(shí)現(xiàn)思路

通過zmq的多對(duì)一通信，將多個(gè)地方的日志發(fā)送到一個(gè)地方集中處理，從而實(shí)現(xiàn)分布式日志。
這個(gè)方法不僅可以解決python分布日志的問題，還可以很好的兼容其他語言，比如項(xiàng)目中還有C、java，那么可以將它們中的日志也發(fā)送過來，一并處理。2

看到這很多人可能明白了，這個(gè)方法類似官網(wǎng)提供的SocketHandler，但本方法其實(shí)是基于QueueHandler實(shí)現(xiàn)的，有利于發(fā)揮zmq易用性、可插拔、并發(fā)性能好的優(yōu)點(diǎn)。

代碼實(shí)現(xiàn)

首先是集中處理日志的程序，也就是上面所說"多對(duì)一"中的一

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日志等級(jí)啟用，允許對(duì)handler設(shè)置setLevel，F(xiàn)alse則忽視級(jí)別
        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)    # 測(cè)試用
        return logging.makeLogRecord(msg)


def main_logger():
    # 日志集中處理區(qū)，在主程序中調(diào)用一次
    
    # handlers配置區(qū)，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
    
    # 設(shè)置監(jiān)聽的端口，并傳遞handlers
    loggerListener = ZeroMQSocketListener("tcp://127.0.0.1:5555",*(ch,console))
    loggerListener.start()   # 開啟一個(gè)子線程處理記錄器監(jiān)聽
    
# 主進(jìn)程調(diào)用一次，非阻塞
main_logger()

自此，日志集中處理就結(jié)束了，是不是很簡單，而且需要注意，我們這里不需要用到root logger，因?yàn)閆eroMQSocketListener會(huì)自動(dòng)調(diào)用各種handlers將日志內(nèi)容進(jìn)行處理，想當(dāng)于替代了logger的工作，所以也就沒必要聲明一個(gè)logger出來了。

更新：

這里的main_logger()是非阻塞，也就是下面還可以寫其他代碼，但是如果什么代碼都沒有，那么主進(jìn)程就會(huì)直接退出，日志就收不到了。

如果接下來不需要做其他工作，那么請(qǐng)?jiān)趍ain_logger()下方使用while True:time.sleep(0.5) 將主進(jìn)程阻塞。

需要重點(diǎn)關(guān)注通信地址"tcp://127.0.0.1:5555"，因?yàn)槠渌胤降娜罩径紩?huì)發(fā)送到這里來。

接下來是子進(jìn)程中或者是你想記錄日志的任何地方，比如在其他同事的電腦里

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()
        
# 創(chuàng)建遠(yuǎn)端日志
rmtlogger = logging.getLogger('sub_root_name')    ##
rmtlogger.setLevel(logging.INFO)     # 建議設(shè)置一下，有時(shí)候默認(rèn)是WARNING級(jí)別
rmtlogger.propagate=False    # 不允許傳遞，日志傳遞到這里就發(fā)送到主進(jìn)程中

# 配置handler
zmqhandler = ZeroMQSocketHandler()
zmqhandler.setLevel(logging.INFO)
rmtlogger.addHandler(zmqhandler)

# if you have submodule
# import submodule  

# 記錄日志
rmtlogger.info("這是一條遙遠(yuǎn)的日志")

如果是多進(jìn)程環(huán)境下，您大可直接將上面的代碼直接開啟到多個(gè)子進(jìn)程中，并不會(huì)出現(xiàn)網(wǎng)絡(luò)問題。

logger可以通過python日志的name系統(tǒng)進(jìn)行傳遞，也就是說如果子進(jìn)程中還有其他模塊，可以通過日志傳遞系統(tǒng)將其他模塊產(chǎn)生的日志傳遞過來，最后一并發(fā)送給監(jiān)聽器，就像下面：

submodule.py

# subprocess.py的子模塊，如需測(cè)試注意調(diào)用
import logging
subMolduleLogger = logging.getLogger(f'sub_root_name.modulename')

subMolduleLogger.info("這是一條子模塊日志")
# 這部分內(nèi)容需要logging基礎(chǔ)知識(shí)