Python整個(gè)底層的邏輯還是比較的復(fù)雜的，內(nèi)容也是比較的多，那么，Python如何進(jìn)行異步處理操作呢？通俗來(lái)說(shuō)，就是怎么同步發(fā)送日志到遠(yuǎn)程服務(wù)器？下面就一步一步的給大家解答，請(qǐng)仔細(xì)閱讀哦。

　　背景

　　在Python中使用日志最常用的方式就是在控制臺(tái)和文件中輸出日志了,logging模塊也很好的提供的相應(yīng)的類,使用起來(lái)也非常方便,但是有時(shí)我們可能會(huì)有一些需求,如還需要將日志發(fā)送到遠(yuǎn)端,或者直接寫(xiě)入數(shù)據(jù)庫(kù),這種需求該如何實(shí)現(xiàn)呢?

　　StreamHandler和FileHandler

　　首先我們先來(lái)寫(xiě)一套簡(jiǎn)單輸出到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)
　　#設(shè)置日志格式
　　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("今天天氣不錯(cuò)")

　　先初始化一個(gè)logger,并且設(shè)置它的日志級(jí)別是DEBUG,然后添初始化了cmd_handler和file_handler,最后將它們添加到logger中,運(yùn)行腳本,會(huì)在cmd中打印出

　　[2020-09-23 10:45:56][DEBUG]今天天氣不錯(cuò)且會(huì)寫(xiě)入到當(dāng)前目錄下的debug.log文件中

　　添加HTTPHandler

　　如果想要在記錄時(shí)將日志發(fā)送到遠(yuǎn)程服務(wù)器上,可以添加一個(gè)HTTPHandler,在python標(biāo)準(zhǔn)庫(kù)logging.handler中,已經(jīng)為我們定義好了很多handler,有些我們可以直接用,本地使用tornado寫(xiě)一個(gè)接收日志的接口,將接收到的參數(shù)全都打印出來(lái)

　　#添加一個(gè)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("今天天氣不錯(cuò)")
　　結(jié)果在服務(wù)端我們收到了很多信息
　　{
　　'name':[b'yyx'],
　　'msg':[b
　　'xe4xbbx8axe5xa4xa9xe5xa4xa9xe6xb0x94xe4xb8x8dxe9x94x99'],
　　'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
　　'xe4xbbx8axe5xa4xa9xe5xa4xa9xe6xb0x94xe4xb8x8dxe9x94x99'],
　　'asctime':[b'2020-09-23 11:17:34']
　　}

　　可以說(shuō)是信息非常之多,但是卻并不是我們想要的樣子,我們只是想要類似于

　　[2020-09-23 10:45:56][DEBUG]今天天氣不錯(cuò)這樣的日志

　　logging.handlers.HTTPHandler只是簡(jiǎn)單的將日志所有信息發(fā)送給服務(wù)端,至于服務(wù)端要怎么組織內(nèi)容是由服務(wù)端來(lái)完成.所以我們可以有兩種方法,一種是改服務(wù)端代碼,根據(jù)傳過(guò)來(lái)的日志信息重新組織一下日志內(nèi)容,第二種是我們重新寫(xiě)一個(gè)類,讓它在發(fā)送的時(shí)候?qū)⒅匦赂袷交罩緝?nèi)容發(fā)送到服務(wù)端。

　　我們采用第二種方法,因?yàn)檫@種方法比較靈活，服務(wù)端只是用于記錄，發(fā)送什么內(nèi)容應(yīng)該是由客戶端來(lái)決定。

　　我們需要重新定義一個(gè)類,我們可以參考logging.handlers.HTTPHandler這個(gè)類,重新寫(xiě)一個(gè)httpHandler類

　　每個(gè)日志類都需要重寫(xiě)emit方法,記錄日志時(shí)真正要執(zhí)行是也就是這個(gè)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):
　　'''
　　重寫(xiě)emit方法，這里主要是為了把初始化時(shí)的baseParam添加進(jìn)來(lái)
　　: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)

　　上面代碼中有一行定義發(fā)送的參數(shù)msg=self.format(record)這行代碼表示,將會(huì)根據(jù)日志對(duì)象設(shè)置的格式返回對(duì)應(yīng)的內(nèi)容。

　　之后再將內(nèi)容通過(guò)requests庫(kù)進(jìn)行發(fā)送,無(wú)論使用get還是post方式,服務(wù)端都可以正常的接收到日志

　　{'log':[b'[2020-09-23 11:39:45][DEBUG]
　　xe4xbbx8axe5xa4xa9xe5xa4xa9xe6xb0x94xe4xb8x8dxe9x94x99']}

　　將bytes類型轉(zhuǎn)一下就得到了:

　　[2020-09-23 11:43:50][DEBUG]今天天氣不錯(cuò)

　　異步的發(fā)送遠(yuǎn)程日志

　　現(xiàn)在我們考慮一個(gè)問(wèn)題,當(dāng)日志發(fā)送到遠(yuǎn)程服務(wù)器過(guò)程中,如果遠(yuǎn)程服務(wù)器處理的很慢,會(huì)耗費(fèi)一定的時(shí)間,那么這時(shí)記錄日志就會(huì)都變慢修改服務(wù)器日志處理類,讓其停頓5秒鐘,模擬長(zhǎng)時(shí)間的處理流程

　　async def post(self):
　　print(self.getParam('log'))
　　await asyncio.sleep(5)
　　self.write({"msg":'ok'})

　　此時(shí)我們?cè)俅蛴∩厦娴娜罩荆?/p>

　　logger.debug("今天天氣不錯(cuò)")
　　logger.debug("是風(fēng)和日麗的")

　　得到的輸出為：

　　[2020-09-23 11:47:33][DEBUG]今天天氣不錯(cuò)

　　[2020-09-23 11:47:38][DEBUG]是風(fēng)和日麗的

　　我們注意到,它們的時(shí)間間隔也是5秒。

　　那么現(xiàn)在問(wèn)題來(lái)了，原本只是一個(gè)記錄日志，現(xiàn)在卻成了拖累整個(gè)腳本的累贅，所以我們需要異步的來(lái)處理遠(yuǎn)程寫(xiě)日志。

　　1使用多線程處理

　　首先想的是應(yīng)該是用多線程來(lái)執(zhí)行發(fā)送日志方法;

　　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},

　　這種方法是可以達(dá)到不阻塞主目的,但是每打印一條日志就需要開(kāi)啟一個(gè)線程,也是挺浪費(fèi)資源的。我們也可以使用線程池來(lái)處理

　　2使用線程池處理

　　python的concurrent.futures中有ThreadPoolExecutor,ProcessPoolExecutor類,是線程池和進(jìn)程池,就是在初始化的時(shí)候先定義幾個(gè)線程,之后讓這些線程來(lái)處理相應(yīng)的函數(shù),這樣不用每次都需要新創(chuàng)建線程

　　線程池的基本使用:

　　exector=ThreadPoolExecutor(max_workers=1)#初始化一個(gè)線程池,只有一個(gè)線程
　　exector.submit(fn,args,kwargs)#將函數(shù)submit到線程池中

　　如果線程池中有n個(gè)線程,當(dāng)提交的task數(shù)量大于n時(shí),則多余的task將放到隊(duì)列中。

　　再次修改上面的emit函數(shù)

　　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)

　　這里為什么要只初始化一個(gè)只有一個(gè)線程的線程池?因?yàn)檫@樣的話可以保證先進(jìn)隊(duì)列里的日志會(huì)先被發(fā)送,如果池子中有多個(gè)線程,則不一定保證順序了。

　　3使用異步aiohttp庫(kù)來(lái)發(fā)送請(qǐng)求

　　上面的CustomHandler類中的emit方法使用的是requests.post來(lái)發(fā)送日志,這個(gè)requests本身是阻塞運(yùn)行的,也正上由于它的存在,才使得腳本卡了很長(zhǎng)時(shí)間,所們我們可以將阻塞運(yùn)行的requests庫(kù)替換為異步的aiohttp來(lái)執(zhí)行g(shù)et和post方法,重寫(xiě)一個(gè)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())

　　這時(shí)代碼執(zhí)行崩潰了:

　　C:Python37liblogging__init__.py:894:RuntimeWarning:coroutine
　　'CustomHandler.emit'was never awaited
　　self.emit(record)
　　RuntimeWarning:Enable tracemalloc to get the object allocation traceback

　　服務(wù)端也沒(méi)有收到發(fā)送日志的請(qǐng)求。

　　究其原因是由于emit方法中使用async with session.post函數(shù),它需要在一個(gè)使用async修飾的函數(shù)里執(zhí)行,所以修改emit函數(shù),使用async來(lái)修飾,這里emit函數(shù)變成了異步的函數(shù),返回的是一個(gè)coroutine對(duì)象,要想執(zhí)行coroutine對(duì)象,需要使用await,但是腳本里卻沒(méi)有在哪里調(diào)用await emit(),所以崩潰信息中顯示coroutine'CustomHandler.emit'was never awaited。

　　既然emit方法返回的是一個(gè)coroutine對(duì)象,那么我們將它放一個(gè)loop中執(zhí)行

　　async def main():
　　await logger.debug("今天天氣不錯(cuò)")
　　await logger.debug("是風(fēng)和日麗的")
　　loop=asyncio.get_event_loop()
　　loop.run_until_complete(main())

　　執(zhí)行依然報(bào)錯(cuò):

　　raise TypeError('An asyncio.Future,a coroutine or an awaitable is'

　　意思是需要的是一個(gè)coroutine,但是傳進(jìn)來(lái)的對(duì)象不是。

　　這似乎就沒(méi)有辦法了，想要使用異步庫(kù)來(lái)發(fā)送，但是卻沒(méi)有可以調(diào)用await的地方。

　　解決辦法是有的,我們使用asyncio.get_event_loop()獲取一個(gè)事件循環(huán)對(duì)象,我們可以在這個(gè)對(duì)象上注冊(cè)很多協(xié)程對(duì)象,這樣當(dāng)執(zhí)行事件循環(huán)的時(shí)候,就是去執(zhí)行注冊(cè)在該事件循環(huán)上的協(xié)程,

　　我們通過(guò)一個(gè)小例子來(lái)看一下:

　　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("執(zhí)行結(jié)束,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()創(chuàng)建了一個(gè)事件循環(huán)對(duì)象loop,并且在loop上創(chuàng)建了兩個(gè)task,并且給task1添加了一個(gè)回調(diào)函數(shù),在task1它執(zhí)行結(jié)束以后,將loop停掉。

　　注意看上面的代碼,我們并沒(méi)有在某處使用await來(lái)執(zhí)行協(xié)程,而是通過(guò)將協(xié)程注冊(cè)到某個(gè)事件循環(huán)對(duì)象上,然后調(diào)用該循環(huán)的run_forever()函數(shù),從而使該循環(huán)上的協(xié)程對(duì)象得以正常的執(zhí)行。

　　上面得到的輸出為:

　　test 5

　　test2 3

　　test 4

　　test2 2

　　test 3

　　test2 1

　　test 2

　　test 1

　　執(zhí)行結(jié)束,task n is 0

　　可以看到,使用事件循環(huán)對(duì)象創(chuàng)建的task,在該循環(huán)執(zhí)行run_forever()以后就可以執(zhí)行了如果不執(zhí)行l(wèi)oop.run_forever()函數(shù),則注冊(cè)在它上面的協(xié)程也不會(huì)執(zhí)行

　　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秒,這時(shí)腳本不會(huì)有任何輸出,在停了5秒以后就中止了，

　　回到之前的日志發(fā)送遠(yuǎn)程服務(wù)器的代碼,我們可以使用aiohttp封裝一個(gè)發(fā)送數(shù)據(jù)的函數(shù),然后在emit中將這個(gè)函數(shù)注冊(cè)到全局的事件循環(huán)對(duì)象loop中,最后再執(zhí)行l(wèi)oop.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封裝發(fā)送數(shù)據(jù)函數(shù)
　　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))
　　#添加一個(gè)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("今天天氣不錯(cuò)")
　　logger.debug("是風(fēng)和日麗的")
　　loop.run_forever()

　　這時(shí)腳本就可以正常的異步執(zhí)行了:

　　loop.create_task(self.submit(msg))也可以使用

　　asyncio.ensure_future(self.submit(msg),loop=loop)來(lái)代替,目的都是將協(xié)程對(duì)象注冊(cè)到事件循環(huán)中。

　　但這種方式有一點(diǎn)要注意,loop.run_forever()將會(huì)一直阻塞,所以需要有個(gè)地方調(diào)用loop.stop()方法.可以注冊(cè)到某個(gè)task的回調(diào)中。

　　綜上所述，文章就給大家介紹到這里了，希望可以給帶來(lái)更多幫助。