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

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

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

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

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

PermissionError: [WinError 32] 另一個程序正在使用此文件，進程無法訪問。

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

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

解決辦法

題主在網(wǎng)上搜集了各種解決上面問題的辦法，基本以下面三個方向為主：

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

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

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

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

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

基于zmq的分布式日志

實現(xiàn)思路

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

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

代碼實現(xiàn)

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

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設(shè)置setLevel，F(xiàn)alse則忽視級別
        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():
    # 日志集中處理區(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()   # 開啟一個子線程處理記錄器監(jiān)聽
    
# 主進程調(diào)用一次，非阻塞
main_logger()

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

更新：

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

如果接下來不需要做其他工作，那么請在main_logger()下方使用while True:time.sleep(0.5) 將主進程阻塞。

需要重點關(guān)注通信地址"tcp://127.0.0.1:5555"，因為其他地方的日志都會發(fā)送到這里來。

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

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

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

# if you have submodule
# import submodule  

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

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

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

submodule.py

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

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