StreamHandler和FileHandler
首先我们先来写一套简单输出到cmd和文件中的代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 | # -*- 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( "今天天气不错" )
|
Nach dem Login kopieren
先初始化一个logger, 并且设置它的日志级别是DEBUG,然后添初始化了 cmd_handler和 file_handler,最后将它们添加到logger中, 运行脚本,会在cmd中打印出
[2020-09-23 10:45:56] [DEBUG] 今天天气不错
且会写入到当前目录下的debug.log文件中
添加HTTPHandler
如果想要在记录时将日志发送到远程服务器上,可以添加一个 HTTPHandler , 在python标准库logging.handler中,已经为我们定义好了很多handler,有些我们可以直接用,本地使用tornado写一个接收 日志的接口,将接收到的参数全都打印出来
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 | # 添加一个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']
}
|
Nach dem Login kopieren
可以说是信息非常之多,但是却并不是我们想要的样子,我们只是想要类似于
[2020-09-23 10:45:56][DEBUG] 今天天气不错
这样的日志
logging.handlers.HTTPHandler 只是简单的将日志所有信息发送给服务端,至于服务端要怎么组织内 容是由服务端来完成. 所以我们可以有两种方法,一种是改服务端代码,根据传过来的日志信息重新组织一 下日志内容, 第二种是我们重新写一个类,让它在发送的时候将重新格式化日志内容发送到服务端。
我们采用第二种方法,因为这种方法比较灵活, 服务端只是用于记录,发送什么内容应该是由客户端来决定。
我们需要重新定义一个类,我们可以参考 logging.handlers.HTTPHandler 这个类,重新写一个httpHandler类
每个日志类都需要重写emit方法,记录日志时真正要执行是也就是这个emit方法:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 | 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)
|
Nach dem Login kopieren
上面代码中有一行定义发送的参数 msg = self.format(record)这行代码表示,将会根据日志对象设置的格式返回对应的内容。
之后再将内容通过requests库进行发送,无论使用get 还是post方式,服务端都可以正常的接收到日志
1 2 | {'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']}
|
Nach dem Login kopieren
将bytes类型转一下就得到了:
[2020-09-23 11:43:50] [DEBUG] 今天天气不错
异步的发送远程日志
现在我们考虑一个问题,当日志发送到远程服务器过程中,如果远程服务器处理的很慢,会耗费一定的时间, 那么这时记录日志就会都变慢修改服务器日志处理类,让其停顿5秒钟,模拟长时间的处理流程
1 2 3 4 | async def post(self):
print (self.getParam('log'))
await asyncio.sleep(5)
self.write({ "msg" : 'ok'})
|
Nach dem Login kopieren
此时我们再打印上面的日志:
1 2 | logger.debug( "今天天气不错" )
logger.debug( "是风和日丽的" )
|
Nach dem Login kopieren
得到的输出为:
[2020-09-23 11:47:33] [DEBUG] 今天天气不错
[2020-09-23 11:47:38] [DEBUG] 是风和日丽的
我们注意到,它们的时间间隔也是5秒。
那么现在问题来了,原本只是一个记录日志,现在却成了拖累整个脚本的累赘,所以我们需要异步的来 处理远程写日志。
1使用多线程处理
首先想的是应该是用多线程来执行发送日志方法;
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 | 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},
|
Nach dem Login kopieren
这种方法是可以达到不阻塞主目的,但是每打印一条日志就需要开启一个线程,也是挺浪费资源的。我们也 可以使用线程池来处理
2使用线程池处理
python 的 concurrent.futures 中有ThreadPoolExecutor, ProcessPoolExecutor类,是线程池和进程池, 就是在初始化的时候先定义几个线程,之后让这些线程来处理相应的函数,这样不用每次都需要新创建线程
线程池的基本使用:
1 2 | exector = ThreadPoolExecutor(max_workers=1) # 初始化一个线程池,只有一个线程
exector.submit(fn, args, kwargs) # 将函数submit到线程池中
|
Nach dem Login kopieren
如果线程池中有n个线程,当提交的task数量大于n时,则多余的task将放到队列中。
再次修改上面的emit函数
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 | 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)
|
Nach dem Login kopieren
这里为什么要只初始化一个只有一个线程的线程池? 因为这样的话可以保证先进队列里的日志会先被发 送,如果池子中有多个线程,则不一定保证顺序了。
3使用异步aiohttp库来发送请求
上面的CustomHandler类中的emit方法使用的是requests.post来发送日志,这个requests本身是阻塞运 行的,也正上由于它的存在,才使得脚本卡了很长时间,所们我们可以将阻塞运行的requests库替换为异步 的aiohttp来执行get和post方法, 重写一个CustomHandler中的emit方法
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 | 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())
|
Nach dem Login kopieren
这时代码执行崩溃了:
1 2 3 4 | 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
|
Nach dem Login kopieren
服务端也没有收到发送日志的请求。
究其原因是由于emit方法中使用 async with session.post 函数,它需要在一个使用async 修饰的函数 里执行,所以修改emit函数,使用async来修饰,这里emit函数变成了异步的函数, 返回的是一个 coroutine 对象,要想执行coroutine对象,需要使用await, 但是脚本里却没有在哪里调用 await emit() ,所以崩溃信息 中显示 coroutine 'CustomHandler.emit' was never awaited。
既然emit方法返回的是一个coroutine对象,那么我们将它放一个loop中执行
1 2 3 4 5 | async def main():
await logger.debug( "今天天气不错" )
await logger.debug( "是风和日丽的" )
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
|
Nach dem Login kopieren
执行依然报错:
raise TypeError('An asyncio.Future, a coroutine or an awaitable is '
意思是需要的是一个coroutine,但是传进来的对象不是。
这似乎就没有办法了,想要使用异步库来发送,但是却没有可以调用await的地方。
解决办法是有的,我们使用 asyncio.get_event_loop() 获取一个事件循环对象, 我们可以在这个对象上注册很多协程对象,这样当执行事件循环的时候,就是去执行注册在该事件循环上的协程,
我们通过一个小例子来看一下:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 | 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()
|
Nach dem Login kopieren
我们使用 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() 函数,则注册在它上面的协程也不会执行
1 2 3 4 5 6 | 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()
|
Nach dem Login kopieren
上面的代码将loop.run_forever() 注释掉,换成time.sleep(5) 停5秒, 这时脚本不会有任何输出,在停了5秒 以后就中止了,
回到之前的日志发送远程服务器的代码,我们可以使用aiohttp封装一个发送数据的函数, 然后在emit中将 这个函数注册到全局的事件循环对象loop中,最后再执行loop.run_forever()
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 | 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()
|
Nach dem Login kopieren
这时脚本就可以正常的异步执行了:
loop.create_task(self.submit(msg)) 也可以使用
asyncio.ensure_future(self.submit(msg), loop=loop) 来代替,目的都是将协程对象注册到事件循环中。
但这种方式有一点要注意,loop.run_forever() 将会一直阻塞,所以需要有个地方调用 loop.stop() 方法. 可以注册到某个task的回调中。
Das obige ist der detaillierte Inhalt vonSo senden Sie Protokolle asynchron in Python an einen Remote-Server. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!