责任链模式是一种设计模式,它允许将请求沿着一系列处理者传递,直到有一个处理者处理它为止。这种模式在Python中非常实用,因为它可以简化代码结构,提高模块化,并增加代码的灵活性。下面,我将通过5个实用案例来教你如何使用Python实现责任链模式。
案例一:电子邮件处理
概述
在这个案例中,我们将使用责任链模式来处理不同类型的电子邮件。每个处理者将负责处理特定类型的电子邮件,如果它不能处理,则将请求传递给链中的下一个处理者。
代码示例
class EmailProcessor:
def __init__(self):
self.handlers = []
def add_handler(self, handler):
self.handlers.append(handler)
def process_email(self, email):
for handler in self.handlers:
handler.handle(email)
class SpamFilter:
def handle(self, email):
if "spam" in email.body:
print("Spam detected, discarded.")
else:
return email
class HighPriorityFilter:
def handle(self, email):
if email.priority == "high":
print("High priority email received.")
return email
else:
return email
# 使用示例
processor = EmailProcessor()
processor.add_handler(SpamFilter())
processor.add_handler(HighPriorityFilter())
email = Email("user@example.com", "test@example.com", "This is a test email.", "normal")
processor.process_email(email)
案例二:错误处理
概述
在这个案例中,我们将使用责任链模式来处理不同类型的错误。每个错误处理者将尝试处理特定的错误类型,如果它不能处理,则将错误传递给链中的下一个处理者。
代码示例
class ErrorHandler:
def __init__(self):
self.handlers = []
def add_handler(self, handler):
self.handlers.append(handler)
def handle_error(self, error):
for handler in self.handlers:
handler.process(error)
class GeneralErrorHandler:
def process(self, error):
print(f"General error: {error}")
class SpecificErrorHandler:
def process(self, error):
if "division by zero" in str(error):
print("Division by zero error caught!")
# 使用示例
error_handler = ErrorHandler()
error_handler.add_handler(GeneralErrorHandler())
error_handler.add_handler(SpecificErrorHandler())
try:
1 / 0
except Exception as e:
error_handler.handle_error(e)
案例三:用户认证
概述
在这个案例中,我们将使用责任链模式来处理用户认证过程。每个认证处理者将尝试验证用户的不同认证信息,如果它不能验证,则将请求传递给链中的下一个处理者。
代码示例
class AuthenticationProcessor:
def __init__(self):
self.handlers = []
def add_handler(self, handler):
self.handlers.append(handler)
def authenticate(self, username, password):
for handler in self.handlers:
if handler.authenticate(username, password):
return True
return False
class UsernameHandler:
def authenticate(self, username, password):
if username == "admin":
return True
return False
class PasswordHandler:
def authenticate(self, username, password):
if password == "admin":
return True
return False
# 使用示例
auth_processor = AuthenticationProcessor()
auth_processor.add_handler(UsernameHandler())
auth_processor.add_handler(PasswordHandler())
if auth_processor.authenticate("admin", "admin"):
print("Authentication successful!")
else:
print("Authentication failed!")
案例四:支付处理
概述
在这个案例中,我们将使用责任链模式来处理支付过程。每个支付处理者将尝试处理不同的支付方式,如果它不能处理,则将请求传递给链中的下一个处理者。
代码示例
class PaymentProcessor:
def __init__(self):
self.handlers = []
def add_handler(self, handler):
self.handlers.append(handler)
def process_payment(self, payment):
for handler in self.handlers:
if handler.process(payment):
return True
return False
class CreditCardHandler:
def process(self, payment):
if payment.method == "credit_card":
print("Processing credit card payment...")
return True
return False
class PayPalHandler:
def process(self, payment):
if payment.method == "paypal":
print("Processing PayPal payment...")
return True
return False
# 使用示例
payment_processor = PaymentProcessor()
payment_processor.add_handler(CreditCardHandler())
payment_processor.add_handler(PayPalHandler())
payment = Payment("user@example.com", "credit_card", 100)
if payment_processor.process_payment(payment):
print("Payment processed successfully!")
else:
print("Payment failed!")
案例五:日志记录
概述
在这个案例中,我们将使用责任链模式来处理日志记录过程。每个日志处理者将尝试记录不同级别的日志,如果它不能记录,则将请求传递给链中的下一个处理者。
代码示例
class LoggerProcessor:
def __init__(self):
self.handlers = []
def add_handler(self, handler):
self.handlers.append(handler)
def log(self, level, message):
for handler in self.handlers:
handler.log(level, message)
class ConsoleLogger:
def log(self, level, message):
print(f"{level}: {message}")
class FileLogger:
def log(self, level, message):
with open("logs.txt", "a") as file:
file.write(f"{level}: {message}\n")
# 使用示例
logger_processor = LoggerProcessor()
logger_processor.add_handler(ConsoleLogger())
logger_processor.add_handler(FileLogger())
logger_processor.log("INFO", "This is an informational message.")
logger_processor.log("ERROR", "This is an error message.")
通过以上5个实用案例,你可以了解到如何在Python中实现责任链模式。这种模式在处理复杂流程时非常有用,可以帮助你构建更加高效和灵活的代码。
