背景
在 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
'\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):
'''
重写 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 秒。
那么当初问题来了，本来只是一个记录日志，当初却成了连累整个脚本的累赘，所以咱们须要异步的来 解决近程写日志。
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:\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 常识欢送返回公众号：Python 编程学习圈 ，发送“J” 即可收费获取，每日干货分享