Python でログをリモートサーバーに非同期で送信する方法を理解する
- coldplay.xixi転載
- 2020-10-22 18:51:552469ブラウズ
Python ビデオ チュートリアル 列では、Python でリモート サーバーにログを非同期的に送信する方法を説明します。
Python でログを使用する最も一般的な方法は、コンソールとファイルにログを出力することです。ログ モジュールも対応するクラスを提供します。これも非常に便利です。使用する必要がありますが、リモート エンドにログを送信したり、データベースに直接書き込む必要があるなどのニーズがある場合があります。このニーズを実現するにはどうすればよいですか?
1. StreamHandler と FileHandler
まず、単純に cmd とファイルに出力する一連のコードを作成します。
# -*- 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("今天天气不错")复制代码
最初にロガーを初期化し、そのログ レベルを DEBUG に設定します。次に、 cmd_handler と file_handler を初期化し、最後にそれらをロガーに追加します。スクリプトを実行すると、 が cmd に出力されます [2020-09-23 10:45:56] [デバッグ] 今日は天気が良いです と、現在のディレクトリの debug.log ファイルに書き込まれます。
2. HTTPHandler の追加
記録時にログをリモート サーバーに送信したい場合は、Python の標準ライブラリlogging.handlerにHTTPHandlerを追加できます。 , 多くのハンドラーが定義されており、そのうちのいくつかは直接使用できます。トルネードをローカルで使用して、ログを受信するためのインターフェイスを作成し、受信したパラメーターをすべて出力します。
# 添加一个httphandlerimport 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] [デバッグ] 今日は天気が良いです このような log.
logging.handlers.HTTPHandler は、単純にログ内のすべての情報をサーバーに送信します。サーバー側で整理する 内容はサーバー側で完結するので、渡されたログ情報に合わせてサーバーのコードを変更してログ内容を再構成する方法と、クラスを書き換えてそのままにする方法の2通りが考えられます。送信されました。再フォーマットされたログの内容がサーバーに送信されます。
この方法のほうが柔軟性が高いため、2 番目の方法を使用します。サーバーは記録のみに使用されます。どのような内容を送信するかはクライアントが決定する必要があります。 。
クラスを再定義する必要があります。logging.handlers.HTTPHandler を参照して、httpHandler クラスを書き換えることができます。
各ログ クラスは、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):
'''
: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) # このコード行は、ログ オブジェクトによって設定された形式に従って、対応するコンテンツが返されることを示します。
その後、コンテンツはリクエスト ライブラリを通じて送信され、get メソッドまたは post メソッドを使用した場合でも、サーバーは正常にログを受信できます
3. リモート ログの非同期送信
ここで問題を考えます。ログがリモート サーバーに送信されるとき、リモート サーバーがそれを適切に処理するかどうかです。一定の時間が経過すると、ログ記録の速度が低下します。
サーバー ログ処理クラスを変更し、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 秒であることがわかります。
ここで問題が発生します。元々は単なるログでしたが、今ではスクリプト全体を引きずる重荷になっているため、リモート ログの書き込みを非同期で処理する必要があります。
3.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}, "headers":headers})
t.start()复制代码
このメソッドは、主な目的を達成できます。ブロッキングではありませんが、ログが出力されるたびにスレッドを開く必要があり、これもリソースの無駄です。スレッド プールを使用して処理することもできます
3.2 スレッド プールを使用して処理する
Python の concurrent.futures には、スレッド プールとプロセス プールである ThreadPoolExecutor クラスと ProcessPoolExecutor クラスがあります。複数のスレッドを定義し、それらのスレッドが対応する関数を処理できるようにすることで、毎回新しいスレッドを作成する必要がなくなります
スレッド プールの基本的な使用法
exector = ThreadPoolExecutor(max_workers=1) # 初始化一个线程池,只有一个线程exector.submit(fn, args, kwargs) # 将函数submit到线程池中复制代码
スレッド プール内の n 個のスレッド。送信されたタスクの数が n より大きい場合、超過したタスクはキューに配置されます。
上記の発行関数を再度変更します
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)复制代码
なぜ初期化だけを行うのかここに 1 つのスレッドしかないスレッド? プール? この方法では、アドバンスト キュー内のログが最初に送信されることが保証できるため、プール内に複数のスレッドがある場合、順序は必ずしも保証されません。
3.3 非同期 aiohttp ライブラリを使用してリクエストを送信する
上記の CustomHandler クラスの Emit メソッドは、requests.post を使用してログを送信します。リクエスト自体はブロックされ実行されているため、その存在はスクリプトが長時間スタックするため、ブロッキング リクエスト ライブラリを非同期 aiohttp に置き換えて get メソッドと post メソッドを実行し、CustomHandler
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())复制代码
で Emit メソッドを書き換えることができます。この時点で、コードの実行はクラッシュしました
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 -= 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()复制代码
我们使用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))# 添加一个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的回调中.
相关免费学习推荐：python视频教程
以上がPython でログをリモートサーバーに非同期で送信する方法を理解するの詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。