**：
import random
import datetime
import time
import threading
import re
import sys
from queue import queue
from pathlib import path
from user_agents import parse
# 資料匹配處理
pattern = '''(?p[\d\.])\s-\s-\s\[(?p[^\[\]]+)\]\s"(?p.*)\s(?p.*)\s(?p.*)"\s(?p\d)\s(?p\d+)\s"[^"]+"\s"(?p[^"]+)"'''
regex = re.compile(pattern) # 編譯
ops =
# 進一步處理每條日誌訊息(時間格式處理、訊息返回值、訊息大小)
# ops.get(name, lambda x: x)(data) 1，如果name有返回值就使用返回值，否則就使用lambda x:x函式
defextract
(line:str) -> dict:
matcher = regex.match(line)
if matcher:
return
# 開啟檔案，依次讀取每一行
defopenfile
(files:str):
with open(files) as file:
for line in file:
fields = extract(line)
if fields:
yield fields
else:
continue
# 裝在日誌檔案或目錄
defload
(*args):
for item in args:
paths = path(item)
ifnot paths.exists():
continue
if paths.is_dir():
for file in paths.iterdir():
if file.is_file():
yield
from openfile(str(file)) # file為path所以使用str函式轉換
elif paths.is_file():
yield
from openfile(str(paths)) # file為path所以使用str函式轉換
# 臨時測試生成資料
# def source():
# while true:
# yield
# time.sleep(1)
# 滑動視窗
defwindow
(src, handler, width:int, interval:int):
''' :param src: 資料來源，快取佇列
:param handler: 資料處理函式
:param width: 時間視窗寬度，秒
:param interval: 處理時間間隔，秒
'''start = datetime.datetime.strptime('20170101 000000 +0800', '%y%m%d %h%m%s %z')
current = datetime.datetime.strptime('20170101 010000 +0800', '%y%m%d %h%m%s %z')
buffer =
delta = datetime.timedelta(seconds=width-interval)
while
true:
data = src.get() # 從佇列中拿出一條訊息
if data:
current = data['datetime']
if (current-start).total_seconds() >= interval: # 每隔interval計算buffer中的資料一次
ret = handler(buffer)
print('{}'.format(ret))
start = current
# 清理佇列中過時的訊息
buffer = [x for x in buffer if x['datetime'] > current]
# 消費者：統計平均值
defhandler
(iterable):
return sum(map(lambda x: x['value'], iterable)) / len(iterable)
# 消費者：統計狀態碼佔比
defstatus_handler
(iterable):
status = {}
for item in iterable:
key = item['status']
status[key] = status.get(key, 0) + 1
total = len(iterable)
return
# 消費者：瀏覽器分析函式
defbrowser_handler
(iterable):
browsers = {}
for item in iterable:
useragent = item['useragent']
key = (useragent.family, useragent.browser.version_string)
browsers[key] = browsers.get(key, 0) + 1
return browsers
# 資料分發器
defdispatcher
(src):
handlers = # 建立乙個執行緒池，存放
queues = # 建立乙個一對多的佇列
defreg
(handler, width:int, interval:int):
queuobj = queue()
threads = threading.thread(target=window, args=(queuobj, handler, width, interval))
defrun():
for handler in handlers:
handler.start() # 啟動執行緒處理資料
for item in src:
for queue in queues:
queue.put(item) # 將資料來源獲取的資料put到所有佇列中
return reg, run
if __name__ == '__main__':
# dir_file = sys.argv[1]
# reg, run = dispatcher(load(dir_file))
reg, run = dispatcher(load('test.log'))
reg(status_handler, 10, 5) # 註冊
run() # 執行
日誌分析（一） 日誌分析常用命令
1.檢視檔案的內容 如果乙個日誌檔案比較小，直接使用cat檢視 cat n 顯示行號 2.分頁顯示檔案 3.顯示檔案尾 tail 命令可以看到檔案最後幾行，n引數後面跟數字，表示顯示檔案最後幾行。4.顯示檔案頭 head顯示開頭幾行，n引數指定顯示開頭幾行 5.內容排序 sort對資料進行排序 n ...
統一日誌處理
日誌是幹啥的.不多說.這裡只記錄怎麼配置日誌.logger 日誌記錄器.可以配置不同的日誌級別.不同的級別顯示的日誌資訊不同的.越往後的日誌級別會包含前面所有日誌級別顯示的資訊 off,fatal,error,warn,info,debug,all loggin.level.root warn這是 ...
統一日誌框架
常見的框架有log4j log4j2 logback 如果乙個專案中整合元件有單獨的框架那麼日誌配置就很混亂 log4j log4j2是沒有實現slf4j門面的 logback是實現的 就是我們獲取logger的包 是從slf4j獲取的 將我們自己的日誌框架通過slf4j實現 如果是log4j通過s...