快捷導(dǎo)航

python更加靈活的Logger日志詳解

更新時(shí)間：2021年09月17日 09:36:45 作者：FlyLikeButterfly

這篇文章主要介紹了python Logger日志，用到的4個(gè)類，針對(duì)每個(gè)知識(shí)點(diǎn)給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值,需要的朋友可以參考下

用到的4個(gè)類：

1、Logger：

打印日志用的對(duì)象；

設(shè)置日志等級(jí)，添加移除handler，添加移除filter，設(shè)置下級(jí)Logger，使用各種方法打印日志；

創(chuàng)建方式有兩種，使用logging.getLogger("mylog")和創(chuàng)建實(shí)例logging.Logger("mylog")；

推薦使用getLogger()的方式，不傳name參數(shù)的獲得的Logger為root Logger，傳name參數(shù)的上級(jí)為root Logger；

直接實(shí)例化的Logger沒有上級(jí)，日志等級(jí)為NOTSET，并且用該實(shí)例創(chuàng)建的下級(jí)Logger的上級(jí)直接為root Logger；

使用getLogger(name)傳入相同的名字獲得是同一個(gè)Logger；

Logger擁有上下級(jí)關(guān)系，父子記錄器名稱之間用“.”分割，Logger擁有有效日志等級(jí)概念（getEffectiveLevel()方法返回的是日志等級(jí)整數(shù)值），直接新建實(shí)例創(chuàng)建的Logger日志等級(jí)為NOTSET，默認(rèn)的根記錄器為WARNING等級(jí)，如果NOTSET等級(jí)的Logger為根記錄器則處理所有消息，否則會(huì)委托給父級(jí)記錄器，遍歷父記錄器鏈，直到找到非NOTSET的父記錄器并把該等級(jí)作為自己的有效等級(jí)，如果直到根記錄器也是NOTSET則處理所有消息；

2、Formatter：

日志打印格式；

創(chuàng)建方式為logging.Formatter(fmt=None, datefmt=None, style='%', validate=True)

fmt：格式化日志

datefmt：格式化fmt中的%(asctime)s

style：在3.2版本添加

validate：在3.8版本添加，不正確或不匹配的樣式和fmt將引發(fā)ValueError；

跟上一篇一致python的logging日志模塊

3、Filter：

過(guò)濾日志；

通過(guò)繼承l(wèi)ogging.Filter類，并實(shí)現(xiàn) filter(self, record: LogRecord) -> int 方法，返回0或者False表示不通過(guò)，返回非0或者True表示可以通過(guò)；（from logging import LogRecord）

LogRecord可以操作的屬性大概有這么多：

同一個(gè)Handler可以添加多個(gè)Filter，依次過(guò)濾，當(dāng)前Filter通過(guò)后傳遞給下一個(gè)Filter；

4、Handler：

日志輸出方式；

設(shè)置日志等級(jí)，設(shè)置formatter，添加移除filter；

Logger可以添加多個(gè)handler，將日志按不同格式和過(guò)濾送往不同的地方，同一個(gè)Logger只會(huì)添加同一個(gè)handler一次，多次添加無(wú)影響；

子Logger處理完自己的handler后，會(huì)將日志傳遞給父Logger的handler處理，依次向上傳遞，不要將同一個(gè)handler同時(shí)添加到父子Logger里，否則父子Logger都會(huì)處理會(huì)打印多次相同日志；

可以通過(guò)設(shè)置Logger對(duì)象的propagate屬性為False關(guān)閉傳遞給父Logger的handler；

大概有這么多Handler：

（from logging import handlers）

StreamHandler：

構(gòu)造方法 logging.StreamHandler(stream=None)

將日志發(fā)送到像sys.stdout、sys.stderr或類似文件的對(duì)象中（支持write()和flush()方法的對(duì)象），默認(rèn)使用sys.stderr，3.2版本后還有個(gè)terminator屬性，可以設(shè)置終止符（默認(rèn)“\n”）；

FileHandler：

構(gòu)造方法 logging.FileHandler(filename, mode='a', encoding=None, delay=False)

將日志記錄到文件中，默認(rèn)文件將無(wú)限增長(zhǎng)，如果delay為true，打開文件會(huì)延遲到第一次調(diào)用emit()；

3.6版本開始，pathlib.Path也可以作為filename的值

NullHandler：

不做任何格式化和輸出，提供給庫(kù)開發(fā)人員使用；

WatchedFileHandler：

這是一個(gè)FileHandler，監(jiān)控正在記錄日志的文件，如果文件變動(dòng)了則關(guān)閉文件重新打開；（windows系統(tǒng)不適用）

BaseRotatingHandler：

構(gòu)造方法 logging.handlers.BaseRotatingHandler(filename, mode, encoding=None, delay=False)

是RotatingFileHandler和TimedRotatingFileHandler的基類，不需要實(shí)例化該類；

RotatingFileHandler：

構(gòu)造方法 logging.handlers.RotatingFileHandler(filename, mode='a', maxBytes=0, backupCount=0, encoding=None, delay=False)

按照日志文件大小和備份日志文件數(shù)量保存日志到文件；

maxBytes或者backupCount為0則不會(huì)滾動(dòng)日志，當(dāng)日志大小接近maxBytes時(shí)會(huì)用“.1”“.2”“.3”...后綴保存舊文件，并保持舊文件數(shù)量不超過(guò)backupCount，當(dāng)前日志一直是沒有后綴的那個(gè)文件；

TimedRotatingFileHandler：

構(gòu)造方法 logging.handlers.TimedRotatingFileHandler(filename, when='h', interval=1, backupCount=0, encoding=None, delay=False, utc=False, atTime=None)

按照時(shí)間間隔和備份文件數(shù)量保存日志到文件；

when：“S”秒，“M”分鐘，“H”小時(shí)，“D”天，“W0”-“W6”工作日（W0為周一），“midnight”午夜（不指定atTime的時(shí)候午夜12點(diǎn)，否則按照atTime的時(shí)間滾動(dòng)）

interval：時(shí)間間隔；

backupCount：保留文件數(shù)量上限；

utc：為true時(shí)文件名后綴使用UTC時(shí)間否則使用本地時(shí)間；

atTime：3.4添加，datetime.time類型，指定一天中發(fā)生滾動(dòng)日志的時(shí)間，只有when為“midnight”或者“W0”-“W6”時(shí)有效；

生成的備份文件文件名后綴格式為%Y-%m-%d_%H-%M-%S，第一次計(jì)算滾動(dòng)時(shí)間（程序啟動(dòng)后）時(shí)會(huì)使用現(xiàn)有日志的最后修改時(shí)間或者當(dāng)前時(shí)間計(jì)算；

SocketHandler：

構(gòu)造方法 logging.handlers.SocketHandler(host, port)

將日志發(fā)送到網(wǎng)絡(luò)套接字，基類使用的是TCP；

發(fā)送的二進(jìn)制bytes是由編碼的，由“內(nèi)容長(zhǎng)度+內(nèi)容”組成，內(nèi)容是用pickle序列化的一個(gè)dict（LogRecord，可以用logging.makeLogRecord(attrdict)轉(zhuǎn)換），長(zhǎng)度是用struct將序列化后的內(nèi)容打包的一個(gè)大端無(wú)符號(hào)long數(shù)值，具體邏輯在SocketHandler類源碼的makePickle()方法中有體現(xiàn)：

搞了半天才搞好的小demo（之前沒看到編碼方式還以為socket的編碼，服務(wù)端接收數(shù)據(jù)解不出）：

#!/usr/bin/env python3
# coding=utf-8
 
import logging
from logging import handlers
import time
 
log = logging.getLogger("mylog")
log.setLevel(logging.DEBUG)
h = logging.handlers.SocketHandler(host="127.0.0.1", port=8899)
f = logging.Formatter("[%(name)s][%(asctime)s]%(message)s")
h.setFormatter(f)
log.addHandler(h)
for x in range(1, 6):
    log.info("test log %d" % x)
    time.sleep(0.5)

服務(wù)端demo：

#!/usr/bin/env python3
# coding=utf-8
 
import socket
import pickle
import struct
import logging
 
 
def unPickle(bs: bytes):
    data_len_bytes = bs[0:4]
    data_len = struct.unpack(">L", data_len_bytes)[0]
    pickled_data = bs[4: data_len + 4 + 1]
    return pickle.loads(pickled_data)
 
 
socket_server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket_server.bind(("127.0.0.1", 8899))
socket_server.listen(3)
while True:
    try:
        client, client_addr = socket_server.accept()
        while True:
            data = client.recv(1024 * 10)
            if data != b'':
                log_dict = unPickle(data)
                log_record = logging.makeLogRecord(log_dict)
                print("recv log:", log_record)
    except Exception as e:
        print("Exception:", e)

先運(yùn)行socket服務(wù)，再測(cè)日志，因?yàn)檫B接不到服務(wù)日志會(huì)被丟棄，運(yùn)行結(jié)果：

DatagramHandler：

構(gòu)造方法 logging.handlers.DatagramHandler(host, port)

繼承了SocketHandler，使用UDP發(fā)送，類似SocketHandler的使用；

SysLogHandler：

構(gòu)造方法 logging.handlers.SysLogHandler(address=('localhost', SYSLOG_UDP_PORT), facility=LOG_USER, socktype=socket.SOCK_DGRAM)

將日志發(fā)送到遠(yuǎn)程或者本地的Unix系統(tǒng)日志，未指定address則使用('localhost', 514)地址；

NTEventLogHandler：

構(gòu)造方法 logging.handlers.NTEventLogHandler(appname, dllname=None, logtype='Application')

將日志發(fā)送到本地的Windows NT、Windows 2000、Windows XP系統(tǒng)日志，使用時(shí)需要Mark Hammond's Win32的python擴(kuò)展；

SMTPHandler：

構(gòu)造方法 logging.handlers.SMTPHandler(mailhost, fromaddr, toaddrs, subject, credentials=None, secure=None, timeout=1.0)

將日志發(fā)送到電子郵件；

mailhost使用(host, port)元組，toaddrs是一個(gè)字符串列表，credentials可以用(username, password)元組指定用戶名密碼

MemoryHandler：

構(gòu)造方法 logging.handlers.MemoryHandler(capacity, flushLevel=ERROR, target=None, flushOnClose=True)

支持在內(nèi)存中緩沖日志，當(dāng)緩存將滿或者出現(xiàn)某種嚴(yán)重情況時(shí)把日志發(fā)送到目標(biāo)handler；

是BufferingHandler的子類；

每當(dāng)向緩沖區(qū)添加日志的時(shí)候都會(huì)調(diào)用shouldFlush()判斷是否需要刷新，如果需要?jiǎng)t會(huì)調(diào)用flush()刷新；

HTTPHandler：

構(gòu)造方法 logging.handlers.HTTPHandler(host, url, method='GET', secure=False, credentials=None, context=None)

通過(guò)GET或者POST向一個(gè)web服務(wù)器發(fā)送日志；

如果要指定端口，host可以使用host:port值；

secure為true，則使用HTTPS；

對(duì)HTTPHandler使用setFormatter()是無(wú)效的，HTTPHandler沒有調(diào)用format()格式化，而是調(diào)用了mapLogRecord()方法然后使用urllib.parse.urlencode()編碼的；

mapLogRecord()函數(shù)很簡(jiǎn)單（有需要可以重寫）：

而LogRecord()里是沒有asctime字段的，所以log.asctime是錯(cuò)誤的，但是logRecord里有created和msecs字段表時(shí)間：

小小的demo：

#!/usr/bin/env python3
# coding=utf-8
 
import logging
from logging import handlers
import time
 
log = logging.getLogger("mylog")
log.setLevel(logging.DEBUG)
h_get = handlers.HTTPHandler(host="127.0.0.1:8080", url="test_http_log", method="GET",
                             secure=False, credentials=None, context=None)
log.addHandler(h_get)
h_post = handlers.HTTPHandler(host="127.0.0.1:8080", url="test_http_log", method="POST",
                              secure=False, credentials=None, context=None)
log.addHandler(h_post)
 
h = logging.StreamHandler()
f = logging.Formatter("[%(levelname)s][%(asctime)s]%(message)s")
h.setFormatter(f)
log.addHandler(h)
for x in range(1, 3):
    log.info("test HTTP log %d" % x)
    time.sleep(0.5)

http服務(wù)端：

#!/usr/bin/env python3
# coding=utf-8
 
from http.server import ThreadingHTTPServer, BaseHTTPRequestHandler
from urllib import parse
import logging
import time
 
 
class MyHttpServer(BaseHTTPRequestHandler):
    def do_GET(self):
        url = parse.urlparse(self.path)
        q = parse.parse_qs(url.query)
        log = logging.makeLogRecord(q)
        log_created = time.localtime(float(log.created[0]))
        format_time = time.strftime("%Y/%m/%d %H:%M:%S", log_created)
        print("-GET:", log.name, log.levelname, log.msg, format_time+","+str(int(float(log.msecs[0])//1)))
        self.send_response(200)
        self.end_headers()
        self.wfile.flush()
 
    def do_POST(self):
        length = int(self.headers["content-length"])
        data = self.rfile.read(length)
        data = data.decode(encoding="utf-8")
        data_dict = parse.parse_qs(data)
        data = logging.makeLogRecord(data_dict)
        print("=POST:", data)
        self.send_response(200)
        self.end_headers()
        self.wfile.flush()
 
 
httpserver = ThreadingHTTPServer(("127.0.0.1", 8080), MyHttpServer)
httpserver.serve_forever()

先運(yùn)行服務(wù)端再跑日志demo，運(yùn)行結(jié)果：

QueueHandler/QueueListener：

構(gòu)造方法 logging.handlers.QueueHandler(queue)

logging.handlers.QueueListener(queue, *handlers, respect_handler_level=False)

將日志發(fā)送到隊(duì)列，用于處理隊(duì)列或者多線程模塊情況；

queue可以是類隊(duì)列的任何對(duì)象（原樣傳給dequeue()函數(shù)），也可以用queue.SimpleQueue代替queue；

QueueHandler的小demo：

#!/usr/bin/env python3
# coding=utf-8
 
import logging
from logging import handlers
from logging import LogRecord
import queue
 
 
class MyLogHandler(logging.Handler):
    def handle(self, record: LogRecord) -> None:
        print(record)
 
 
log = logging.getLogger("mylog")
log.setLevel(logging.DEBUG)
q = queue.SimpleQueue()
h = handlers.QueueHandler(q)
f = logging.Formatter("[%(levelname)s]%(message)s")
h.setFormatter(f)
log.addHandler(h)
 
listener = handlers.QueueListener(q, MyLogHandler(), respect_handler_level=False)
 
listener.start()
print("listener started")
log.info("test queue log1 info")
log.error("test queue log2 error")
 
listener.stop()
print("listener stopted")
log.info("test queue log3 info")
log.error("test queue log4 error")
print("test end")

執(zhí)行結(jié)果：

常用Demo：

#!/usr/bin/env python3
# coding=utf-8
 
import logging
from logging import LogRecord
from logging import handlers
import sys
 
# log1 = logging.Logger("a", logging.DEBUG)
log1 = logging.getLogger("a")
_log1 = logging.getLogger("a")
print(log1 is _log1, id(log1), id(_log1))
log1.setLevel(logging.DEBUG)
log2 = log1.getChild("b")
log3 = log1.getChild("c")
print(log1)
print(log2)
print(log3)
 
h1 = logging.StreamHandler()
format1 = logging.Formatter("[H1] [%(levelname)-8s] %(message)s")
h1.setFormatter(format1)
log1.addHandler(h1)
 
h2 = logging.StreamHandler()
format2 = logging.Formatter(fmt="[H2][%(levelname)7s][%(asctime)s]%(message)s", datefmt="%c")
h2.setFormatter(format2)
 
 
class myFilter(logging.Filter):
    def filter(self, record: LogRecord) -> int:
        print("record:" + repr(record))
        if "HELLO" in record.msg:
            return True
        else:
            return False
 
 
h2.addFilter(myFilter())
log2.addHandler(h2)
# log2.addFilter(myFilter())
 
h3 = logging.StreamHandler()
format3 = logging.Formatter("[H3]{%(levelname)s}{%(name)s}%(message)s")
h3.setFormatter(format3)
h3.setStream(sys.stdout)
h3.setLevel(logging.INFO)
log3.addHandler(h3)
#
h3_file = logging.FileHandler(filename="test_Logging2.log", mode="w")
h3_file.setFormatter(logging.Formatter("[H3File][%(levelname)8s][%(asctime)s]%(message)s"))
log3.addHandler(h3_file)
#
h3_rotatingfile = handlers.RotatingFileHandler(filename="test_Logging2_rotating.log", mode="a",
                                               maxBytes=256, backupCount=3,
                                               encoding="utf-8", delay=False)
h3_rotatingfile.setFormatter(logging.Formatter("[h3_rotating][%(levelname)s] %(message)s"))
log3.addHandler(h3_rotatingfile)
#
h3_timedrotationgfile = handlers.TimedRotatingFileHandler(filename="test_Logging2_timedrotating.log",
                                                          when="S", interval=5,
                                                          backupCount=5, encoding="utf-8",
                                                          delay=False, utc=False,
                                                          # atTime=
                                                          )
h3_timedrotationgfile.setFormatter(logging.Formatter("[H3timed][%(levelname)s]%(message)s"))
log3.addHandler(h3_timedrotationgfile)
log3.propagate = False
 
log1.debug("test log1")
log1.warning("warn log1")
log2.debug("test log2 HELLO")
log2.info("info log2")
log2.warning("warn log2 HELLO")
log2.error("error log2")
log3.debug("test log3")
log3.warning("warn log3")

多次執(zhí)行結(jié)果中的一次：