Bagaimana untuk menghantar log ke pelayan jauh secara tak segerak dalam Python

StreamHandler dan FileHandler

Mula-mula, mari kita tulis satu set kod ringkas untuk dikeluarkan ke cmd dan fail:

# -*- 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("今天天气不错")

Mula-mula mulakan logger dan sediakannya tahap log ialah DEBUG, kemudian mulakan cmd_handler dan file_handler, dan akhirnya menambahnya pada logger Jalankan skrip, dan

[2020-09-23 10:45:56] [DEBUG] 今天天气不错 akan dicetak dalam cmd dan ditulis untuk nyahpepijat dalam fail .log direktori

Tambah HTTPHandler

Jika anda ingin menghantar log ke pelayan jauh semasa merakam, anda boleh menambah HTTPHandler, yang telah ditakrifkan untuk kami dalam pengelogan perpustakaan standard python. banyak pengendali, sebahagian daripadanya boleh kami gunakan secara langsung Kami menggunakan puting beliung secara tempatan untuk menulis antara muka untuk menerima log dan mencetak semua parameter yang diterima

# 添加一个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'],
&#39;funcName&#39;: [b &#39;<module>&#39;],
&#39;created&#39;: [b &#39;1600831054.8881223&#39;],
&#39;msecs&#39;: [b &#39;888.1223201751709&#39;],
&#39;relativeCreated&#39;: [b &#39;22.99976348876953&#39;],
&#39;thread&#39;: [b &#39;14876&#39;],
&#39;threadName&#39;: [b &#39;MainThread&#39;],
&#39;processName&#39;: [b &#39;MainProcess&#39;],
&#39;process&#39;: [b &#39;8648&#39;],
&#39;message&#39;: [b
&#39;\xe4\xbb\x8a\xe5\xa4\xa9\xe5\xa4\xa9\xe6\xb0\x94\xe4\xb8\x8d\xe9\x94\x99&#39;],
&#39;asctime&#39;: [b &#39;2020-09-23 11:17:34&#39;]
}

Boleh dikatakan terdapat banyak maklumat, tetapi ia bukan apa yang kami mahukan, kami hanya mahukan log yang serupa dengan

[2020-09-23 10:45:56][DEBUG] 今天天气不错 logging.handlers.HTTHandler hanya menghantar semua maklumat log ke pelayan mengatur kandungan, Ia dilakukan oleh pelayan Jadi kita boleh mempunyai dua kaedah Satu ialah menukar kod pelayan dan menyusun semula kandungan log mengikut maklumat log yang diluluskan semasa menghantar Hantar kandungan log yang diformat semula ke pelayan.

Kami menggunakan kaedah kedua kerana kaedah ini lebih fleksibel Pelayan hanya digunakan untuk rakaman, dan pelanggan harus memutuskan kandungan yang hendak dihantar.

Kita perlu mentakrifkan semula kelas Kita boleh merujuk kepada kelas logging.handlers.HTTPHandler dan menulis semula kelas httpHandler

Setiap kelas log perlu menulis semula kaedah emit untuk merekodkan What sebenarnya dilaksanakan apabila pengelogan ialah kaedah 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)

Terdapat baris dalam kod di atas yang mentakrifkan parameter yang akan dihantar, msg = self.format(record) baris ini bahawa ia akan ditetapkan mengikut objek log Kandungan yang sepadan dikembalikan dalam format.

Kemudian hantar kandungan melalui perpustakaan permintaan Tanpa mengira menggunakan kaedah dapatkan atau pos, pelayan boleh menerima log secara normal

{'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']}

Tukar jenis bait dan anda akan mendapatnya. . :

[2020-09-23 11:43:50] [DEBUG] Cuaca baik hari ini

Menghantar log jauh secara tidak segerak

Sekarang kami mempertimbangkan masalah Apabila log dihantar ke pelayan jauh, jika pelayan jauh memprosesnya dengan sangat perlahan, ia akan mengambil masa yang tertentu . 🎜>

Output yang diperolehi Untuk:

[2020-09-23 11:47:33] [DEBUG] Cuaca baik hari ini

[2020-09- 23 11:47:38] [DEBUG] Cuaca cerah dan cerah

Kami perasan bahawa selang masa antara mereka juga adalah 5 saat.

Sekarang datang masalah ia pada asalnya hanya log, tetapi kini ia telah menjadi beban yang menyeret ke bawah keseluruhan skrip, jadi kami perlu mengendalikan penulisan log jauh secara tidak segerak.

1 Gunakan multi-threading

Perkara pertama yang perlu difikirkan ialah menggunakan multi-threading untuk melaksanakan kaedah penghantaran log

async def post(self):
  print(self.getParam('log'))
  await asyncio.sleep(5)
  self.write({"msg": 'ok'})

Ini kaedah adalah mungkin Tujuan utama untuk tidak menyekat dicapai, tetapi setiap kali log dicetak, benang perlu dibuka, yang juga merupakan pembaziran sumber. Kami juga boleh menggunakan kumpulan benang untuk memproses

2 Gunakan kumpulan benang untuk memproses

Terdapat kelas ThreadPoolExecutor dan ProcessPoolExecutor dalam concurrent.futures python, iaitu kumpulan benang dan kumpulan proses semasa pemulaan. Tentukan beberapa utas, dan kemudian biarkan utas ini mengendalikan fungsi yang sepadan, supaya anda tidak perlu membuat utas baharu setiap kali

Penggunaan asas kumpulan utas:

logger.debug("今天天气不错")
logger.debug("是风和日丽的")

Jika Terdapat n utas dalam kumpulan benang Apabila bilangan tugasan yang diserahkan lebih besar daripada n, lebihan tugasan akan diletakkan dalam baris gilir.

Ubah suai fungsi emit di atas sekali lagi

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

Mengapa kita hanya memulakan kumpulan benang dengan hanya satu utas Kerana dengan cara ini, kita boleh memastikan log dalam baris gilir lanjutan akan dihantar terlebih dahulu terdapat berbilang benang dalam kolam, Benang, pesanan tidak semestinya dijamin. 3 Gunakan perpustakaan aiohttp tak segerak untuk menghantar permintaan

Kaedah emit dalam kelas CustomHandler di atas menggunakan requests.post untuk menghantar log permintaan itu sendiri disekat dan dijalankan, itulah sebabnya Kewujudannya membuat skrip tersekat untuk masa yang lama, jadi kami boleh menggantikan perpustakaan permintaan menyekat dengan aiohttp tak segerak untuk melaksanakan kaedah get dan pos, dan menulis semula kaedah emit dalam CustomHandler

exector = ThreadPoolExecutor(max_workers=1) # 初始化一个线程池,只有一个线程
exector.submit(fn, args, kwargs) # 将函数submit到线程池中

Pada masa ini, pelaksanaan kod ranap:

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)

Pelayan tidak menerima permintaan untuk menghantar log.

Alasannya ialah kerana fungsi async dengan session.post digunakan dalam kaedah emit, ia perlu dilaksanakan dalam fungsi yang diubah suai dengan async, jadi fungsi emit diubah suai dan diubah suai dengan async, di mana fungsi emit menjadi fungsi tak segerak , objek coroutine dikembalikan Untuk melaksanakan objek coroutine, anda perlu menggunakan await, tetapi await emit() tidak dipanggil di mana-mana dalam skrip, jadi maklumat ranap sistem menunjukkan bahawa coroutine 'CustomHandler.emit' tidak pernah. ditunggu.

既然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的回调中。

Atas ialah kandungan terperinci Bagaimana untuk menghantar log ke pelayan jauh secara tak segerak dalam Python. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!