通過logging模組，重寫一個logging2模組，獨立開啟執行緒，將待寫的日誌資訊非同步放入佇列，做到日誌輸出不影響主流程效能，環境python3.8

logging2.py

import os
import threading
import queue
import time
import datetime
import logging
from logging.handlers import RotatingFileHandler

class logging2(threading.Thread):
  AQueue = queue.Queue(100000)
  nPID = os.getpid()
  Adt = datetime.datetime.now().strftime('%Y%m%d')
  nCount = 1
  
  def __init__(self,threadID,name,module,logLevel):
    threading.Thread.__init__(self)
    self.threadID = threadID
    self.name = name
    self.module = module
    
    print("set loglevel: [%s]" % (logLevel) )
    formatter = logging.Formatter('%(asctime)s|%(name)s|%(process)d|%(levelname)s|%(message)s')
    logfile = "log_" + self.module + "_" + str(logging2.nPID) + "_" + str(logging2.Adt) + ".log"
    self.logger = logging.getLogger(__name__)
    
    self.rHandler = RotatingFileHandler(logfile,maxBytes = 10*1024*1024,backupCount = 10)
    self.rHandler.setFormatter(formatter)
    
    self.console = logging.StreamHandler()  
    self.console.setFormatter(formatter)
    
    if logLevel == 'DEBUG' :
      self.logger.setLevel(level = logging.DEBUG)
      self.rHandler.setLevel(logging.DEBUG)
      self.console.setLevel(logging.DEBUG)
    elif logLevel == 'INFO' :
      self.logger.setLevel(level = logging.INFO)
      self.rHandler.setLevel(logging.INFO)
      self.console.setLevel(logging.INFO)
    elif logLevel == 'WARNING' :
      self.logger.setLevel(level = logging.WARN)
      self.rHandler.setLevel(logging.WARN)
      self.console.setLevel(logging.WARN)
    elif logLevel == 'ERROR' :
      self.logger.setLevel(level = logging.ERROR)
      self.rHandler.setLevel(logging.ERROR)
      self.console.setLevel(logging.ERROR)    

    self.logger.addHandler(self.rHandler)
    self.logger.addHandler(self.console)    

  #如果跨天了，則重新生成新的檔名
  def reSetLog(self):
    AdtTemp = datetime.datetime.now().strftime('%Y%m%d')
    #比較新的時間
    if AdtTemp == logging2.Adt:
      return(True)
      
    logging2.Adt = AdtTemp
    logfile = "log_" + self.module + "_" + str(logging2.nPID) + "_" + str(AdtTemp) + ".log"
    self.rHandler = RotatingFileHandler(logfile,maxBytes = 1*1024,backupCount = 10)
    
    self.logger.addHandler(self.rHandler)
    self.logger.addHandler(self.console)  
    logging2.nCount += 1
    
  def run(self):
    print ("開啟日誌執行緒：" + self.name)
    i = 0
    while True:
      #data = "queue test data"
      #debug(data)
      #print("Queuesize: %s" % (logging2.AQueue.qsize()))
      self.reSetLog()
      if logging2.AQueue.empty() == False:
        #從佇列獲取日誌訊息
        data = logging2.AQueue.get()
        #解析日誌訊息，格式：日誌級別，內容
        level = list(data.keys())[0]
        content = data.get(level)
        #把內容按分隔符|解析成list傳入引數
        lstContent = list(content.split('|'))
        if level == 'DEBUG' :
          self.logger.debug(*lstContent)
        elif level == 'INFO' :
          self.logger.info(*lstContent)
        elif level == 'WARNING' :
          self.logger.warn(*lstContent)
        elif level == 'ERROR' :
          self.logger.error(*lstContent)
      else:
        time.sleep(0.5)

    print ("退出執行緒：" + self.name)  
  
def debug(*content):  
  logMsg = ""
  #傳入多個引數用豎線分隔符分開
  for i in range(len(content)):
    if i == len(content)-1:
      logMsg += content[i]
    else:
      logMsg += content[i]+"|"
  logging2.AQueue.put({'DEBUG':logMsg})
      
def info(*content):
  logMsg = ""
  for i in range(len(content)):
    if i == len(content)-1:
      logMsg += content[i]
    else:
      logMsg += content[i]+"|"
  logging2.AQueue.put({'INFO':logMsg})
        
def warn(*content):
  logMsg = ""
  for i in range(len(content)):
    if i == len(content)-1:
      logMsg += content[i]
    else:
      logMsg += content[i]+"|"
  logging2.AQueue.put({'WARNING':logMsg})
    
def error(*content):
  logMsg = ""
  for i in range(len(content)):
    if i == len(content)-1:
      logMsg += content[i]
    else:
      logMsg += content[i]+"|"
  logging2.AQueue.put({'ERROR':logMsg})
          
def init(module,level):
  # 建立新執行緒
  thread1 = logging2(1,"Thread-log",level)
  # 開啟新執行緒
  thread1.start()
#  thread1.join()

測試樁logMain.py

import sys
import os
import time
import threading

if __name__ == '__main__':
  import logging2
  logging2.init("logMain","DEBUG")
  
  teststr = "22222"
  
  while True:
    logging2.debug('this is a debug log test [%s] ',teststr)
    logging2.info('this is a info log test [%s] [%s]',teststr,teststr)
    logging2.warn('this is a warn log test')
    logging2.error('this is a error log test')
    #time.sleep(0.1)
  
  
  print(threading.enumerate())
  
  print('press ctrl_c to exit')

測試結果

生成日誌檔案：

-rw-rw-r--. 1 zxl zxl 10152463 6月 24 17:52 log_logMain_57554_20200624.log

檔案內容如下：

以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支援我們。