背景
在 Python 中使用日志最常用的方式就是在控制台和文件中输出日志了,logging 模块也很好的提供的相应 的类,使用起来也非常方便,但是有时我们可能会有一些需求,如还需要将日志发送到远端,或者直接写入数 据库,这种需求该如何实现呢?
StreamHandler 和 FileHandler
# -*- 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] 今天天气不错
复制代码
添加 HTTPHandler
# 添加一个httphandlerimport logging.handlershttp_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 类
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)
复制代码
这行代码表示,将会根据日志对象设置的格式返回对应的内容。
{'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] 今天天气不错
复制代码
异步的发送远程日志
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] 是风和日丽的
复制代码
那么现在问题来了，原本只是一个记录日志，现在却成了拖累整个脚本的累赘，所以我们需要异步的来 处理远程写日志。
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到线程池中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 库来发送请求
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 awaitedself.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。
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 '
复制代码
这似乎就没有办法了，想要使用异步库来发送，但是却没有可以调用 await 的地方。
import asyncioasync 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()
复制代码
注意看上面的代码,我们并没有在某处使用 await 来执行协程,而是通过将协程注册到某个事件循环对象上, 然后调用该循环的 run_forever() 函数,从而使该循环上的协程对象得以正常的执行。
test 5test2 3test 4test2 2test 3test2 1test 2test 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 = 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 的回调中。
更多学习资料戳下方！！！
https://qrcode.ceba.ceshiren.com/link?name=article&project_id=qrcode&from=infoQ×tamp=1662366626&author=xueqi
评论