Python如何异步发送日志到远程服务器？下文给大家解答

　　Python整个底层的逻辑还是比较的复杂的，内容也是比较的多，那么，Python如何进行异步处理操作呢？通俗来说，就是怎么同步发送日志到远程服务器？下面就一步一步的给大家解答，请仔细阅读哦。

　　背景

　　在Python中使用日志最常用的方式就是在控制台和文件中输出日志了,logging模块也很好的提供的相应的类,使用起来也非常方便,但是有时我们可能会有一些需求,如还需要将日志发送到远端,或者直接写入数据库,这种需求该如何实现呢?

　　StreamHandler和FileHandler

　　首先我们先来写一套简单输出到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

　　如果想要在记录时将日志发送到远程服务器上,可以添加一个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
　　'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']
　　}

　　可以说是信息非常之多,但是却并不是我们想要的样子,我们只是想要类似于

　　[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]
　　xe4xbbx8axe5xa4xa9xe5xa4xa9xe6xb0x94xe4xb8x8dxe9x94x99']}

　　将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秒。

　　那么现在问题来了，原本只是一个记录日志，现在却成了拖累整个脚本的累赘，所以我们需要异步的来处理远程写日志。

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

　　这种方法是可以达到不阻塞主目的,但是每打印一条日志就需要开启一个线程,也是挺浪费资源的。我们也可以使用线程池来处理

　　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使用异步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:Python37liblogging__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的回调中。

　　综上所述，文章就给大家介绍到这里了，希望可以给带来更多帮助。