瞭解python 中紀錄檔非同步傳送到遠端伺服器
欄目瞭解python中紀錄檔非同步傳送到遠端伺服器的方法。
在python中使用紀錄檔最常用的方式就是在控制檯和檔案中輸出紀錄檔了,logging模組也很好的提供的相應的類,使用起來也非常方便,但是有時我們可能會有一些需求,如還需要將紀錄檔傳送到遠端,或者直接寫入資料庫,這種需求該如何實現呢?
一、StreamHandler和FileHandler
首先我們先來寫一套簡單輸出到cmd和檔案中的程式碼
# -*- coding: utf-8 -*-"""
-------------------------------------------------
File Name： loger
Description :
Author : yangyanxing
date： 2020/9/23
-------------------------------------------------
"""import loggingimport sysimport os# 初始化loggerlogger = logging.getLogger("yyx")
logger.setLevel(logging.DEBUG)# 設定紀錄檔格式fmt = logging.Formatter('[%(asctime)s] [%(levelname)s] %(message)s', '%Y-%m-%d %H:%M:%S')# 新增cmd handlercmd_handler = logging.StreamHandler(sys.stdout)
cmd_handler.setLevel(logging.DEBUG)
cmd_handler.setFormatter(fmt)# 新增檔案的handlerlogpath = os.path.join(os.getcwd(), 'debug.log')
file_handler = logging.FileHandler(logpath)
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(fmt)# 將cmd和file handler新增到logger中logger.addHandler(cmd_handler)
logger.addHandler(file_handler)
logger.debug("今天天氣不錯")複製程式碼首先初始化一個logger, 並且設定它的紀錄檔級別是DEBUG,然後添初始化了 cmd_handler和 file_handler, 最後將它們新增到logger中, 執行指令碼,會在cmd中列印出 [2020-09-23 10:45:56] [DEBUG] 今天天氣不錯 且會寫入到當前目錄下的debug.log檔案中.
二、新增HTTPHandler
如果想要在記錄時將紀錄檔傳送到遠端伺服器上,可以新增一個 HTTPHandler , 在python標準庫logging.handler中,已經為我們定義好了很多handler,有些我們可以直接用,本地使用tornado寫一個接收紀錄檔的介面,將接收到的引數全都列印出來
# 新增一個httphandlerimport logging.handlers
http_handler = logging.handlers.HTTPHandler(r"127.0.0.1:1987", '/api/log/get')
http_handler.setLevel(logging.DEBUG)
http_handler.setFormatter(fmt)
logger.addHandler(http_handler)
logger.debug("今天天氣不錯")複製程式碼結果在伺服器端我們收到了很多資訊
{
'name': [b 'yyx'],
'msg': [b '\xe4\xbb\x8a\xe5\xa4\xa9\xe5\xa4\xa9\xe6\xb0\x94\xe4\xb8\x8d\xe9\x94\x99'],
'args': [b '()'],
'levelname': [b 'DEBUG'],
'levelno': [b '10'],
'pathname': [b 'I:/workplace/yangyanxing/test/loger.py'],
'filename': [b 'loger.py'],
'module': [b 'loger'],
'exc_info': [b 'None'],
'exc_text': [b 'None'],
'stack_info': [b 'None'],
'lineno': [b '41'],
'funcName': [b '<module>'],
'created': [b '1600831054.8881223'],
'msecs': [b '888.1223201751709'],
'relativeCreated': [b '22.99976348876953'],
'thread': [b '14876'],
'threadName': [b 'MainThread'],
'processName': [b 'MainProcess'],
'process': [b '8648'],
'message': [b '\xe4\xbb\x8a\xe5\xa4\xa9\xe5\xa4\xa9\xe6\xb0\x94\xe4\xb8\x8d\xe9\x94\x99'],
'asctime': [b '2020-09-23 11:17:34']
}複製程式碼可以說是資訊非常之多,但是卻並不是我們想要的樣子,我們只是想要類似於 [2020-09-23 10:45:56] [DEBUG] 今天天氣不錯 這樣的紀錄檔.
logging.handlers.HTTPHandler 只是簡單的將紀錄檔所有資訊傳送給伺服器端,至於伺服器端要怎麼組織內容是由伺服器端來完成. 所以我們可以有兩種方法,一種是改伺服器端程式碼,根據傳過來的紀錄檔資訊重新組織一下紀錄檔內容, 第二種是我們重新寫一個類,讓它在傳送的時候將重新格式化紀錄檔內容傳送到伺服器端.
我們採用第二種方法,因為這種方法比較靈活， 伺服器端只是用於記錄，傳送什麼內容應該是由使用者端來決定。
我們需要重新定義一個類,我們可以參考logging.handlers.HTTPHandler 這個類,重新寫一個httpHandler類
每個紀錄檔類都需要重寫emit方法,記錄紀錄檔時真正要執行是也就是這個emit方法
class CustomHandler(logging.Handler):
def __init__(self, host, uri, method="POST"):
logging.Handler.__init__(self)
self.url = "%s/%s" % (host, uri)
method = method.upper() if method not in ["GET", "POST"]: raise ValueError("method must be GET or POST")
self.method = method def emit(self, record):
'''
:param record:
:return:
'''
msg = self.format(record) if self.method == "GET": if (self.url.find("?") >= 0):
sep = '&'
else:
sep = '?'
url = self.url + "%c%s" % (sep, urllib.parse.urlencode({"log": msg}))
requests.get(url, timeout=1) else:
headers = { "Content-type": "application/x-www-form-urlencoded", "Content-length": str(len(msg))
}
requests.post(self.url, data={'log': msg}, headers=headers, timeout=1)複製程式碼上面程式碼中有一行定義傳送的引數 msg = self.format(record)
這行程式碼錶示,將會根據紀錄檔物件設定的格式返回對應的內容. 之後再將內容通過requests庫進行傳送,無論使用get 還是post方式,伺服器端都可以正常的接收到紀錄檔
[2020-09-23 11:43:50] [DEBUG] 今天天氣不錯
三、非同步的傳送遠端紀錄檔
現在我們考慮一個問題,當紀錄檔傳送到遠端伺服器過程中,如果遠端伺服器處理的很慢,會耗費一定的時間,那麼這時記錄紀錄檔就會都變慢
修改伺服器紀錄檔處理類,讓其停頓5秒鐘,模擬長時間的處理流程
async def post(self):
print(self.getParam('log')) await asyncio.sleep(5)
self.write({"msg": 'ok'})複製程式碼此時我們再列印上面的紀錄檔
logger.debug("今天天氣不錯")
logger.debug("是風和日麗的")複製程式碼得到的輸出為
[2020-09-23 11:47:33] [DEBUG] 今天天氣不錯 [2020-09-23 11:47:38] [DEBUG] 是風和日麗的複製程式碼
我們注意到,它們的時間間隔也是5秒。
那麼現在問題來了，原本只是一個記錄紀錄檔，現在卻成了拖累整個指令碼的累贅，所以我們需要非同步的來處理遠端寫紀錄檔。
3.1 使用多執行緒處理
首先想的是應該是用多執行緒來執行傳送紀錄檔方法
def emit(self, record):
msg = self.format(record) if self.method == "GET": if (self.url.find("?") >= 0):
sep = '&'
else:
sep = '?'
url = self.url + "%c%s" % (sep, urllib.parse.urlencode({"log": msg}))
t = threading.Thread(target=requests.get, args=(url,))
t.start() else:
headers = { "Content-type": "application/x-www-form-urlencoded", "Content-length": str(len(msg))
}
t = threading.Thread(target=requests.post, args=(self.url,), kwargs={"data":{'log': msg}, "headers":headers})
t.start()複製程式碼這種方法是可以達到不阻塞主目的,但是每列印一條紀錄檔就需要開啟一個執行緒,也是挺浪費資源的。我們也可以使用執行緒池來處理
3.2 使用執行緒池處理
python 的 concurrent.futures 中有ThreadPoolExecutor, ProcessPoolExecutor類,是執行緒池和程序池,就是在初始化的時候先定義幾個執行緒,之後讓這些執行緒來處理相應的函數,這樣不用每次都需要新建立執行緒
執行緒池的基本使用
exector = ThreadPoolExecutor(max_workers=1) # 初始化一個執行緒池,只有一個執行緒exector.submit(fn, args, kwargs) # 將函數submit到執行緒池中複製程式碼
如果執行緒池中有n個執行緒,當提交的task數量大於n時,則多餘的task將放到佇列中.
再次修改上面的emit函數
exector = ThreadPoolExecutor(max_workers=1)def emit(self, record):
msg = self.format(record)
timeout = aiohttp.ClientTimeout(total=6) if self.method == "GET": if (self.url.find("?") >= 0):
sep = '&'
else:
sep = '?'
url = self.url + "%c%s" % (sep, urllib.parse.urlencode({"log": msg}))
exector.submit(requests.get, url, timeout=6) else:
headers = { "Content-type": "application/x-www-form-urlencoded", "Content-length": str(len(msg))
}
exector.submit(requests.post, self.url, data={'log': msg}, headers=headers, timeout=6)複製程式碼這裡為什麼要只初始化一個只有一個執行緒的執行緒池? 因為這樣的話可以保證先進佇列裡的紀錄檔會先被傳送,如果池子中有多個執行緒,則不一定保證順序了。
3.3 使用非同步aiohttp庫來傳送請求
上面的CustomHandler類中的emit方法使用的是requests.post來傳送紀錄檔,這個requests本身是阻塞執行的,也正上由於它的存在,才使得指令碼卡了很長時間,所們我們可以將阻塞執行的requests庫替換為非同步的aiohttp來執行get和post方法, 重寫一個CustomHandler中的emit方法
class CustomHandler(logging.Handler):
def __init__(self, host, uri, method="POST"):
logging.Handler.__init__(self)
self.url = "%s/%s" % (host, uri)
method = method.upper() if method not in ["GET", "POST"]: raise ValueError("method must be GET or POST")
self.method = method async def emit(self, record):
msg = self.format(record)
timeout = aiohttp.ClientTimeout(total=6) if self.method == "GET": if (self.url.find("?") >= 0):
sep = '&'
else:
sep = '?'
url = self.url + "%c%s" % (sep, urllib.parse.urlencode({"log": msg})) async with aiohttp.ClientSession(timeout=timeout) as session: async with session.get(self.url) as resp:
print(await resp.text()) else:
headers = { "Content-type": "application/x-www-form-urlencoded", "Content-length": str(len(msg))
} async with aiohttp.ClientSession(timeout=timeout, headers=headers) as session: async with session.post(self.url, data={'log': msg}) as resp:
print(await resp.text())複製程式碼這時程式碼執行崩潰了
C:\Python37\lib\logging\__init__.py:894: RuntimeWarning: coroutine 'CustomHandler.emit' was never awaited self.emit(record) RuntimeWarning: Enable tracemalloc to get the object allocation traceback複製程式碼
伺服器端也沒有收到傳送紀錄檔的請求。
究其原因是由於emit方法中使用async with session.post 函數,它需要在一個使用async 修飾的函數裡執行,所以修改emit函數,使用async來修飾,這裡emit函數變成了非同步的函數, 返回的是一個coroutine 物件,要想執行coroutine物件,需要使用await, 但是指令碼裡卻沒有在哪裡呼叫 await emit() ,所以崩潰資訊中顯示coroutine 'CustomHandler.emit' was never awaited.
既然emit方法返回的是一個coroutine物件,那麼我們將它放一個loop中執行
async def main():
await logger.debug("今天天氣不錯") await logger.debug("是風和日麗的")
loop = asyncio.get_event_loop()
loop.run_until_complete(main())複製程式碼執行依然報錯
raise TypeError('An asyncio.Future, a coroutine or an awaitable is '複製程式碼意思是需要的是一個coroutine,但是傳進來的物件不是。
這似乎就沒有辦法了，想要使用非同步庫來傳送，但是卻沒有可以呼叫await的地方.
解決辦法是有的,我們使用 asyncio.get_event_loop() 獲取一個事件迴圈物件, 我們可以在這個物件上註冊很多協程物件,這樣當執行事件迴圈的時候,就是去執行註冊在該事件迴圈上的協程, 我們通過一個小例子來看一下
import asyncio
async def test(n):
while n > 0: await asyncio.sleep(1)
print("test {}".format(n))
n -= 1
return n
async def test2(n):
while n >0: await asyncio.sleep(1)
print("test2 {}".format(n))
n -= 1def stoploop(task):
print("執行結束, task n is {}".format(task.result()))
loop.stop()
loop = asyncio.get_event_loop()
task = loop.create_task(test(5))
task2 = loop.create_task(test2(3))
task.add_done_callback(stoploop)
task2 = loop.create_task(test2(3))
loop.run_forever()複製程式碼我們使用loop = asyncio.get_event_loop() 建立了一個事件迴圈物件loop, 並且在loop上建立了兩個task, 並且給task1新增了一個回撥函數,在task1它執行結束以後,將loop停掉.
注意看上面的程式碼,我們並沒有在某處使用await來執行協程,而是通過將協程註冊到某個事件迴圈物件上,然後呼叫該回圈的run_forever() 函數,從而使該回圈上的協程物件得以正常的執行.
上面得到的輸出為
test 5 test2 3 test 4 test2 2 test 3 test2 1 test 2 test 1 執行結束, task n is 0複製程式碼
可以看到,使用事件迴圈物件建立的task,在該回圈執行run_forever() 以後就可以執行了.
如果不執行loop.run_forever() 函數,則註冊在它上面的協程也不會執行
loop = asyncio.get_event_loop() task = loop.create_task(test(5)) task.add_done_callback(stoploop) task2 = loop.create_task(test2(3)) time.sleep(5)# loop.run_forever()複製程式碼
上面的程式碼將loop.run_forever() 註釋掉,換成time.sleep(5) 停5秒, 這時指令碼不會有任何輸出,在停了5秒以後就中止了.
回到之前的紀錄檔傳送遠端伺服器的程式碼,我們可以使用aiohttp封裝一個傳送資料的函數, 然後在emit中將這個函數註冊到全域性的事件迴圈物件loop中,最後再執行loop.run_forever() .
loop = asyncio.get_event_loop()class CustomHandler(logging.Handler):
def __init__(self, host, uri, method="POST"):
logging.Handler.__init__(self)
self.url = "%s/%s" % (host, uri)
method = method.upper() if method not in ["GET", "POST"]: raise ValueError("method must be GET or POST")
self.method = method # 使用aiohttp封裝傳送資料函數
async def submit(self, data):
timeout = aiohttp.ClientTimeout(total=6) if self.method == "GET": if self.url.find("?") >= 0:
sep = '&'
else:
sep = '?'
url = self.url + "%c%s" % (sep, urllib.parse.urlencode({"log": data})) async with aiohttp.ClientSession(timeout=timeout) as session: async with session.get(url) as resp:
print(await resp.text()) else:
headers = { "Content-type": "application/x-www-form-urlencoded",
} async with aiohttp.ClientSession(timeout=timeout, headers=headers) as session: async with session.post(self.url, data={'log': data}) as resp:
print(await resp.text()) return True
def emit(self, record):
msg = self.format(record)
loop.create_task(self.submit(msg))# 新增一個httphandlerhttp_handler = CustomHandler(r"http://127.0.0.1:1987", 'api/log/get')
http_handler.setLevel(logging.DEBUG)
http_handler.setFormatter(fmt)
logger.addHandler(http_handler)
logger.debug("今天天氣不錯")
logger.debug("是風和日麗的")
loop.run_forever()複製程式碼這時指令碼就可以正常的非同步執行了.
loop.create_task(self.submit(msg)) 也可以使用
asyncio.ensure_future(self.submit(msg), loop=loop)
來代替,目的都是將協程物件註冊到事件迴圈中.
但這種方式有一點要注意,loop.run_forever() 將會一直阻塞,所以需要有個地方呼叫loop.stop()方法. 可以註冊到某個task的回撥中.
相關免費學習推薦：
以上就是了解python 中紀錄檔非同步傳送到遠端伺服器的詳細內容，更多請關注TW511.COM其它相關文章！