首先我們先來寫一套簡單輸出到cmd和檔案中的程式碼：
# -*- coding: utf-8 -*- """ ------------------------------------------------- File Name： loger Description : Author : yangyanxing date： 2020/9/23 ------------------------------------------------- """ import logging import sys import os # 初始化logger logger = logging.getLogger("yyx") logger.setLevel(logging.DEBUG) # 设置日志格式 fmt = logging.Formatter('[%(asctime)s] [%(levelname)s] %(message)s', '%Y-%m-%d %H:%M:%S') # 添加cmd handler cmd_handler = logging.StreamHandler(sys.stdout) cmd_handler.setLevel(logging.DEBUG) cmd_handler.setFormatter(fmt) # 添加文件的handler logpath = 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 , 在python標準庫logging.handler中,已經為我們定義好了很多handler,有些我們可以直接用,本地使用tornado寫一個接收日誌的接口,將接收到的參數全都打印出來
# 添加一个httphandler import 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 ''], '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): ''' 重写emit方法，这里主要是为了把初始化时的baseParam添加进来 :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方式,服務端都可以正常的接收到日誌
{'log': [b'[2020-09-23 11:39:45] [DEBUG] \xe4\xbb\x8a\xe5\xa4\xa9\xe5\xa4\xa9\xe6\xb0\x94\xe4\xb8\x8d\xe9\x94\x99']}
將bytes類型轉一下就得到了:
[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秒。
那麼現在問題來了，原本只是一個記錄日誌，現在卻成了拖累整個腳本的累贅，所以我們需要異步的來 處理遠端寫日誌。
首先想的是應該是用多執行緒來執行傳送日誌方法;
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},
這個方法是可以達到不阻塞主目的,但是每打印一條日誌就需要開啟一個線程,也是挺浪費資源的。我們也可以使用線程池來處理
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)
這裡為什麼要只初始化一個只有一個線程的線程池? 因為這樣的話可以保證先進隊列裡的日誌會先被發送,如果池子中有多個執行緒,則不一定保證順序了。
上面的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 -= 1 def 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)) # 添加一个httphandler http_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怎麼異步發送日誌到遠端伺服器的詳細內容。更多資訊請關注PHP中文網其他相關文章！
