Python loggingモジュールの非同期スレッドはログを書いて過程の解析を実現します。
logingモジュールを通じて、logging 2モジュールを書き換えて、独自にスレッドを開いて、書くべきログ情報をキューに入れて、ログ出力が主流プロセス性能に影響しないようにします。環境python 3.8
loging 2.py
ログファイルを生成:
-rw-rw-r--1 zxl zxl 10152463 6月24日:52 log_ロゴマーク57554_202000624 log
ファイルの内容は以下の通りです。
以上が本文の全部です。皆さんの勉強に役に立つように、私たちを応援してください。
loging 2.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", module, level)
#
thread1.start()
# thread1.join()
杭のロゴMain.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日:52 log_ロゴマーク57554_202000624 log
ファイルの内容は以下の通りです。
以上が本文の全部です。皆さんの勉強に役に立つように、私たちを応援してください。