Idea utama ialah: buka fail, alihkan penunjuk ke hujung fail, kemudian keluarkan data jika ada adalah data, dan tidur untuk tempoh masa jika tiada data.
import time
import sys
from typing import Callable, NoReturn
class Tail(object):
def __init__(
self,
file_name: str,
output: Callable[[str], NoReturn] = sys.stdout.write,
interval: int = 1
):
self.file_name: str = file_name
self.output: Callable[[str], NoReturn] = output
self.interval: int = interval
def __call__(self):
with open(self.file_name) as f:
f.seek(0, 2) # 从文件结尾处开始seek
while True:
line: str = f.readline()
if line:
self.output(line) # 使用print都会每次都打印新的一行
else:
time.sleep(self.interval)
if __name__ == '__main__':
filename: str = sys.argv[0]
Tail(filename)()Selepas itu, buat panggilan berikut:
nama fail python xxx.py
tail -fSecara lalai, 10 baris data terakhir dibaca dahulu, dan kemudian data masa nyata dibaca dari penghujung Untuk fail kecil, anda boleh membaca semua kandungan fail terlebih dahulu dan mengeluarkan 10 baris terakhir, tetapi membaca teks penuh Prestasi mendapatkan semula 10 baris terakhir adalah tidak tinggi, dan syarat sempadan untuk melancarkan 10 baris adalah. juga sangat rumit. Mari kita lihat pelaksanaan membaca teks penuh terlebih dahulu dan kemudian mendapatkan semula 10 baris terakhir:
import time
import sys
from typing import Callable, NoReturn
class Tail(object):
def __init__(
self,
file_name: str,
output: Callable[[str], NoReturn] = sys.stdout.write,
interval: int = 1
):
self.file_name: str = file_name
self.output: Callable[[str], NoReturn] = output
self.interval: int = interval
def __call__(self):
with open(self.file_name) as f:
self.read_last_line(f)
while True:
line: str = f.readline()
if line:
self.output(line) # 使用print都会每次都打印新的一行
else:
time.sleep(self.interval)
def read_last_line(self, f):
last_lines = f.readlines()[-10:]
for line in last_lines:
self.output(line)
if __name__ == '__main__':
filename: str = sys.argv[0]
Tail(filename)() Anda dapat melihat bahawa pelaksanaannya sangat mudah, berbanding dengan versi pertama, hanya ada satu lagi read_last_line的函数 Langkah seterusnya adalah untuk menyelesaikan masalah prestasi Apabila fail sangat besar, logik ini tidak berfungsi, terutamanya beberapa fail log selalunya bersaiz beberapa G. Dalam sistem Linux, tiada antara muka yang boleh menentukan penunjuk untuk melompat ke 10 baris terakhir Anda hanya boleh menggunakan kaedah berikut untuk mensimulasikan output 10 baris terakhir:
Jika panjang aksara lompatan adalah sama dengan 1024 * 10, Kemudian gunakan aksara baris baharu untuk mengira berapa banyak baris panjang aksara telah diambil Jika bilangan baris lebih daripada 10, hanya 10 baris terakhir akan menjadi output. Jika hanya 4 baris telah dibaca, teruskan membaca 6*1024 sehingga 10 baris telah dibaca read_last_line
import time
import sys
from typing import Callable, List, NoReturn
class Tail(object):
def __init__(
self,
file_name: str,
output: Callable[[str], NoReturn] = sys.stdout.write,
interval: int = 1,
len_line: int = 1024
):
self.file_name: str = file_name
self.output: Callable[[str], NoReturn] = output
self.interval: int = interval
self.len_line: int = len_line
def __call__(self, n: int = 10):
with open(self.file_name) as f:
self.read_last_line(f, n)
while True:
line: str = f.readline()
if line:
self.output(line) # 使用print都会每次都打印新的一行
else:
time.sleep(self.interval)
def read_last_line(self, file, n):
read_len: int = self.len_line * n
# 跳转游标到最后
file.seek(0, 2)
# 获取当前结尾的游标位置
now_tell: int = file.tell()
while True:
if read_len > file.tell():
# 如果跳转的字符长度大于原来文件长度,那就把所有文件内容打印出来
file.seek(0) # 由于read方法是按照游标进行打印, 所以要重置游标
last_line_list: List[str] = file.read().split('\n')[-n:]
# 重新获取游标位置
now_tell: int = file.tell()
break
# 跳转到我们预估的字符位置
file.seek(-read_len, 2)
read_str: str = file.read(read_len)
cnt: int = read_str.count('\n')
if cnt >= n:
# 如果获取的行数大于要求的行数,则获取前n行的行数
last_line_list: List[str] = read_str.split('\n')[-n:]
break
else:
# 如果获取的行数小于要求的行数,则预估需要获取的行数,继续获取
if cnt == 0:
line_per: int = read_len
else:
line_per: int = int(read_len / cnt)
read_len = line_per * n
for line in last_line_list:
self.output(line + '\n')
# 重置游标,确保接下来打印的数据不重复
file.seek(now_tell)
if __name__ == '__main__':
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("-f", "--filename")
parser.add_argument("-n", "--num", default=10)
args, unknown = parser.parse_known_args()
if not args.filename:
raise RuntimeError('filename args error')
Tail(args.filename)(int(args.num))3. Edisi ketiga - membaca fail log keluaran dengan elegan
. Anda boleh mendapati bahawa prestasi logik bacaan masa nyata masih sangat lemah. Jika fail dibaca sekali sesaat, prestasi masa nyata akan menjadi terlalu perlahan , pemproses akan menduduki terlalu banyak Situasi prestasi terbaik adalah jika anda boleh mengetahui kemas kini fail dan kemudian mencetak fail, maka prestasi boleh dijamin Mujurlah, dalam Linux read_last_line menyediakan Fungsi ini fail log ialah ia akan dilogrotkan jika log diputarkan, maka kita perlu membuka semula fail dan seterusnya membaca data Dalam kes ini,
import os
import sys
from typing import Callable, List, NoReturn
import pyinotify
multi_event = pyinotify.IN_MODIFY | pyinotify.IN_MOVE_SELF # 监控多个事件
class InotifyEventHandler(pyinotify.ProcessEvent): # 定制化事件处理类，注意继承
"""
执行inotify event的封装
"""
f: 'open()'
filename: str
path: str
wm: 'pyinotify.WatchManager'
output: Callable
def my_init(self, **kargs):
"""pyinotify.ProcessEvent要求不能直接继承__init__, 而是要重写my_init, 我们重写这一段并进行初始化"""
# 获取文件
filename: str = kargs.pop('filename')
if not os.path.exists(filename):
raise RuntimeError('Not Found filename')
if '/' not in filename:
filename = os.getcwd() + '/' + filename
index = filename.rfind('/')
if index == len(filename) - 1 or index == -1:
raise RuntimeError('Not a legal path')
self.f = None
self.filename = filename
self.output: Callable = kargs.pop('output')
self.wm = kargs.pop('wm')
# 只监控路径,这样就能知道文件是否移动
self.path = filename[:index]
self.wm.add_watch(self.path, multi_event)
def read_line(self):
"""统一的输出方法"""
for line in self.f.readlines():
self.output(line)
def process_IN_MODIFY(self, event):
"""必须为process_事件名称，event表示事件对象, 这里表示监控到文件发生变化, 进行文件读取"""
if event.pathname == self.filename:
self.read_line()
def process_IN_MOVE_SELF(self, event):
"""必须为process_事件名称，event表示事件对象, 这里表示监控到文件发生重新打开, 进行文件读取"""
if event.pathname == self.filename:
# 检测到文件被移动重新打开文件
self.f.close()
self.f = open(self.filename)
self.read_line()
def __enter__(self) -> 'InotifyEventHandler':
self.f = open(self.filename)
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.f.close()
class Tail(object):
def __init__(
self,
file_name: str,
output: Callable[[str], NoReturn] = sys.stdout.write,
interval: int = 1,
len_line: int = 1024
):
self.file_name: str = file_name
self.output: Callable[[str], NoReturn] = output
self.interval: int = interval
self.len_line: int = len_line
wm = pyinotify.WatchManager() # 创建WatchManager对象
inotify_event_handler = InotifyEventHandler(
**dict(filename=file_name, wm=wm, output=output)
) # 实例化我们定制化后的事件处理类, 采用**dict传参数
wm.add_watch('/tmp', multi_event) # 添加监控的目录，及事件
self.notifier = pyinotify.Notifier(wm, inotify_event_handler) # 在notifier实例化时传入,notifier会自动执行
self.inotify_event_handle: 'InotifyEventHandler' = inotify_event_handler
def __call__(self, n: int = 10):
"""通过inotify的with管理打开文件"""
with self.inotify_event_handle as i:
# 先读取指定的行数
self.read_last_line(i.f, n)
# 启用inotify的监听
self.notifier.loop()
def read_last_line(self, file, n):
read_len: int = self.len_line * n
# 获取当前结尾的游标位置
file.seek(0, 2)
now_tell: int = file.tell()
while True:
if read_len > file.tell():
# 如果跳转的字符长度大于原来文件长度,那就把所有文件内容打印出来
file.seek(0)
last_line_list: List[str] = file.read().split('\n')[-n:]
# 重新获取游标位置
now_tell: int = file.tell()
break
file.seek(-read_len, 2)
read_str: str = file.read(read_len)
cnt: int = read_str.count('\n')
if cnt >= n:
# 如果获取的行数大于要求的行数,则获取前n行的行数
last_line_list: List[str] = read_str.split('\n')[-n:]
break
else:
# 如果获取的行数小于要求的行数,则预估需要获取的行数,继续获取
if cnt == 0:
line_per: int = read_len
else:
line_per: int = int(read_len / cnt)
read_len = line_per * n
for line in last_line_list:
self.output(line + '\n')
# 重置游标,确保接下来打印的数据不重复
file.seek(now_tell)
if __name__ == '__main__':
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("-f", "--filename")
parser.add_argument("-n", "--num", default=10)
args, unknown = parser.parse_known_args()
if not args.filename:
raise RuntimeError('filename args error')
Tail(args.filename)(int(args.num)) Anda boleh melihat bahawa fail terbuka asal ditukar kepada menggunakan inotify untuk membuka fail (pada masa ini, kaedah my_init akan dipanggil untuk inisialisasi ), selepas dibuka, kami masih menjalankan n baris kod asal, dan kemudian memberikannya kepada inotify untuk dijalankan Sebelum inotify dijalankan, kami memasang kaedah fail pembukaan semula dan kaedah pencetakan fail dalam peristiwa yang sepadan dengan inotifyy, dan kemudian apabila inotify berjalan , kaedah yang sepadan akan dilaksanakan mengikut acara yang sepadan. inotify
Atas ialah kandungan terperinci Bagaimana untuk melengkapkan arahan ekor dengan Python. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!
![[Web front-end] Permulaan pantas Node.js](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
