以下是实现高性能日志系统的完整代码，包含异步写入、磁盘缓存和日志轮转功能：
import logging
import logging.handlers
import queue
import threading
import time
from collections import deque
class AsyncLogHandler(logging.Handler):
def __init__(self, base_handler, max_queue_size=10000,
cache_size=1000, flush_interval=5):
super().__init__()
self.base_handler = base_handler
self.log_queue = queue.Queue(max_queue_size)
self.cache = deque(maxlen=cache_size)
self.flush_interval = flush_interval
self._running = True
self._worker = threading.Thread(target=self._process_logs)
self._worker.daemon = True
self._worker.start()
def _process_logs(self):
while self._running or not self.log_queue.empty():
try:
record = self.log_queue.get(timeout=self.flush_interval)
self.cache.append(record)
self.base_handler.emit(record)
except queue.Empty:
if self.cache:
self._flush_cache()
except Exception as e:
print(f"Logging error: {e}")
def _flush_cache(self):
for record in self.cache:
self.base_handler.emit(record)
self.cache.clear()
def emit(self, record):
try:
self.log_queue.put_nowait(record)
except queue.Full:
self._flush_cache()
self.log_queue.put_nowait(record)
def close(self):
self._running = False
self._worker.join()
self._flush_cache()
self.base_handler.close()
def create_async_logger(name, filename, level=logging.INFO,
max_bytes=10*1024*1024, backup_count=5):
logger = logging.getLogger(name)
logger.setLevel(level)
# 创建RotatingFileHandler
file_handler = logging.handlers.RotatingFileHandler(
filename, maxBytes=max_bytes, backupCount=backup_count
)
file_handler.setFormatter(logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
))
# 包装为异步处理器
async_handler = AsyncLogHandler(file_handler)
logger.addHandler(async_handler)
return logger
import os
import tempfile
from async_logger import create_async_logger
class LogManager:
def __init__(self, log_dir=None, cache_size=1000):
self.log_dir = log_dir or tempfile.gettempdir()
self.cache_size = cache_size
self.loggers = {}
os.makedirs(self.log_dir, exist_ok=True)
def get_logger(self, name, level=logging.INFO):
if name not in self.loggers:
log_file = os.path.join(self.log_dir, f"{name}.log")
self.loggers[name] = create_async_logger(
name, log_file, level=level
)
return self.loggers[name]
def flush_all(self):
for logger in self.loggers.values():
for handler in logger.handlers:
if isinstance(handler, AsyncLogHandler):
handler._flush_cache()
def close_all(self):
self.flush_all()
for logger in self.loggers.values():
for handler in logger.handlers:
handler.close()
log_manager import LogManager
import time
import random
def main():
log_manager = LogManager(log_dir="./logs")
logger = log_manager.get_logger("app")
# 模拟高并发日志写入
for i in range(1000):
logger.info(f"Processing task {i}")
if random.random() < 0.1:
logger.error(f"Error occurred in task {i}")
time.sleep(0.001) # 模拟处理时间
log_manager.close_all()
if __name__ == "__main__":
main()
这个日志系统实现了异步写入、内存缓存和自动日志轮转功能，适合高并发场景。核心组件包括异步处理器、日志管理器和示例用法。
http://www.thedesignrepublic.com/
http://www.yxhgq.com/
http://www.jianhuotech.com/
http://www.bjyozd.com
http://www.sunny-way.cn/
http://www.cnsenkai.com/
http://www.zluren.com.cn/
http://www.cddtk119.com/
http://www.boao168.com.cn/
http://www.czs365.net/
http://www.ianvan.com/
http://www.fangzhoushidai.com
http://www.qiaosen-kj.com
http://gzhyxjj.cn
http://xatlzg.com
http://www.doushiwang.cn
http://www.sxyuanzheng.com
http://www.jwdyd.com
http://www.xaminglang.com
http://www.lboneti.cn
http://www.ahlongda.com
http://www.gdlanye.com
http://www.ktlcutter.com
http://www.hbzfjn.com
http://www.shbaimule.com
http://www.clgzkj.com
http://www.fjjhwy.cn
http://www.goodluck-lift.com
http://www.storlead.shop
http://magicians.com.cn
http://mall.yeepayer.com
http://tzkqzj.com
http://www.jinfanweilai.com
https://www.durkflex.cn/
http://www.szfmhj168.com.cn/
http://hfnz.ahaiba.com/
http://sjmzjypx010.ahaiba.com/
http://nyh.ahaiba.com/