本文目录导读：
我来为你介绍几种Python记录进程运行异常日志的方法：
使用logging模块记录异常
import logging
import traceback
import sys
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('app.log', encoding='utf-8'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
def risky_function():
try:
# 可能出错的代码
result = 1 / 0
except Exception as e:
# 记录完整的异常信息
logger.error(f"发生异常: {str(e)}")
logger.error(f"异常类型: {type(e).__name__}")
logger.error(f"详细追踪:\n{traceback.format_exc()}")
raise
# 使用示例
if __name__ == "__main__":
try:
risky_function()
except Exception as e:
logger.critical(f"进程终止: {str(e)}")
sys.exit(1)
使用装饰器自动记录异常
import logging
import functools
import traceback
logging.basicConfig(level=logging.INFO, filename='app.log')
logger = logging.getLogger(__name__)
def log_exceptions(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception as e:
logger.error(f"函数 {func.__name__} 异常:")
logger.error(f"参数: args={args}, kwargs={kwargs}")
logger.error(f"错误信息: {str(e)}")
logger.error(f"追踪信息:\n{traceback.format_exc()}")
raise
return wrapper
@log_exceptions
def process_data(data):
return data / 0 # 这里会触发异常
# 使用
try:
process_data(10)
except:
print("异常已记录到日志")
全局异常捕获
import logging
import sys
import traceback
logging.basicConfig(level=logging.INFO, filename='app.log')
logger = logging.getLogger(__name__)
def global_exception_handler(exc_type, exc_value, exc_traceback):
"""全局异常处理器"""
if issubclass(exc_type, KeyboardInterrupt):
# 不记录键盘中断
sys.__excepthook__(exc_type, exc_value, exc_traceback)
return
logger.critical(f"未捕获的异常:")
logger.critical(f"类型: {exc_type.__name__}")
logger.critical(f"值: {exc_value}")
logger.critical(f"追踪信息:\n{''.join(traceback.format_tb(exc_traceback))}")
# 设置全局异常处理器
sys.excepthook = global_exception_handler
# 测试
def cause_error():
raise ValueError("这是一个测试异常")
cause_error()
完整的多进程异常日志示例
import logging
import multiprocessing
import traceback
import time
import random
class ProcessLogger:
def __init__(self, log_file='process.log'):
self.log_file = log_file
self.setup_logger()
def setup_logger(self):
self.logger = logging.getLogger('ProcessLogger')
self.logger.setLevel(logging.DEBUG)
# 文件处理器
fh = logging.FileHandler(self.log_file, encoding='utf-8')
fh.setLevel(logging.DEBUG)
# 控制台处理器
ch = logging.StreamHandler()
ch.setLevel(logging.INFO)
# 格式器
formatter = logging.Formatter(
'%(asctime)s - %(processName)s - %(levelname)s - %(message)s'
)
fh.setFormatter(formatter)
ch.setFormatter(formatter)
self.logger.addHandler(fh)
self.logger.addHandler(ch)
def worker_function(self, worker_id):
"""工作进程函数"""
process_name = multiprocessing.current_process().name
try:
self.logger.info(f"进程 {process_name} (ID: {worker_id}) 开始工作")
# 模拟随机异常
if random.random() < 0.3:
raise ValueError(f"进程 {worker_id} 遇到随机错误")
# 模拟正常工作
time.sleep(random.uniform(0.5, 2))
result = random.randint(1, 100)
self.logger.info(f"进程 {process_name} 完成工作，结果: {result}")
return result
except Exception as e:
self.logger.error(f"进程 {process_name} 异常:")
self.logger.error(f"错误信息: {str(e)}")
self.logger.error(f"追踪信息:\n{traceback.format_exc()}")
return None
def run_workers(self, num_workers=5):
"""运行多个工作进程"""
with multiprocessing.Pool(processes=num_workers) as pool:
results = pool.map(self.worker_function, range(num_workers))
# 统计结果
successful = [r for r in results if r is not None]
failed = [r for r in results if r is None]
self.logger.info(f"执行完成: 成功 {len(successful)}, 失败 {len(failed)}")
return results
# 使用示例
if __name__ == "__main__":
process_logger = ProcessLogger()
results = process_logger.run_workers(5)
带邮件通知的异常日志
import logging
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import traceback
class EmailLogHandler(logging.Handler):
def __init__(self, smtp_server, port, from_addr, to_addrs, password):
super().__init__()
self.smtp_server = smtp_server
self.port = port
self.from_addr = from_addr
self.to_addrs = to_addrs
self.password = password
def emit(self, record):
if record.levelno >= logging.ERROR: # 只发送ERROR及以上级别
try:
msg = MIMEMultipart()
msg['From'] = self.from_addr
msg['To'] = ', '.join(self.to_addrs)
msg['Subject'] = f'异常告警: {record.name}'
body = self.format(record)
msg.attach(MIMEText(body, 'plain'))
with smtplib.SMTP(self.smtp_server, self.port) as server:
server.starttls()
server.login(self.from_addr, self.password)
server.send_message(msg)
except:
pass
# 配置日志
logger = logging.getLogger('AppLogger')
logger.setLevel(logging.DEBUG)
# 文件处理器
file_handler = logging.FileHandler('app.log', encoding='utf-8')
file_handler.setLevel(logging.DEBUG)
# 邮件处理器 (需要配置实际的邮件服务器信息)
email_handler = EmailLogHandler(
smtp_server='smtp.gmail.com',
port=587,
from_addr='your_email@gmail.com',
to_addrs=['admin@example.com'],
password='your_password'
)
email_handler.setLevel(logging.ERROR)
# 格式器
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s\n%(exc_info)s'
)
file_handler.setFormatter(formatter)
email_handler.setFormatter(formatter)
logger.addHandler(file_handler)
logger.addHandler(email_handler)
# 使用
try:
# 你的代码
risky_operation()
except Exception as e:
logger.error(f"进程运行异常: {str(e)}", exc_info=True)
关键建议
-
选择合适的日志级别：
- DEBUG: 详细调试信息
- INFO: 一般运行信息
- WARNING: 警告信息
- ERROR: 错误信息
- CRITICAL: 严重错误
-
包含关键信息：
- 时间戳
- 进程ID/名称
- 错误类型
- 完整追踪信息
- 相关上下文数据
-
日志轮转：
from logging.handlers import RotatingFileHandler
handler = RotatingFileHandler('app.log', maxBytes=1024*1024, backupCount=5)
选择最适合你项目需求的方案即可。
