pin_drop当前位置:知识文库 ❯ 图文
Python策略模式完全指南 - 消除条件分支的最佳实践
一、什么是策略模式
策略模式（Strategy Pattern）是行为型设计模式中的一种。它定义了一系列算法，将每个算法封装成独立的策略类，使它们可以相互替换，而客户端代码无需修改。
策略模式的核心价值在于：用组合替代继承，用多态替代条件分支，让代码更符合开闭原则和单一职责原则。
策略模式的核心要素
-
Strategy（策略接口）：定义所有支持的算法的公共接口
-
ConcreteStrategy（具体策略）：实现具体的算法
-
Context（上下文）：持有一个策略对象的引用，负责调用策略
-
动态切换：运行时可以替换不同的策略
二、策略模式的基础实现
1. 经典实现 - 排序算法
代码示例
from abc import ABC, abstractmethod
# 策略接口
class SortStrategy(ABC):
@abstractmethod
def sort(self, data):
pass
# 具体策略
class QuickSort(SortStrategy):
def sort(self, data):
print("使用快速排序")
return sorted(data)
class MergeSort(SortStrategy):
def sort(self, data):
print("使用归并排序")
return sorted(data)
class BubbleSort(SortStrategy):
def sort(self, data):
print("使用冒泡排序")
n = len(data)
for i in range(n):
for j in range(0, n-i-1):
if data[j] > data[j+1]:
data[j], data[j+1] = data[j+1], data[j]
return data
# 上下文
class Sorter:
def __init__(self, strategy: SortStrategy = None):
self._strategy = strategy
@property
def strategy(self):
return self._strategy
@strategy.setter
def strategy(self, strategy: SortStrategy):
self._strategy = strategy
def sort(self, data):
if self._strategy is None:
raise ValueError("未设置排序策略")
return self._strategy.sort(data)
# 使用
sorter = Sorter(QuickSort())
result = sorter.sort([3, 1, 4, 1, 5, 9, 2, 6])
print(f"排序结果: {result}")
# 运行时切换策略
sorter.strategy = MergeSort()
result = sorter.sort([3, 1, 4, 1, 5, 9, 2, 6])
print(f"排序结果: {result}")
2. 消除条件分支 - 折扣计算
代码示例
# 传统方式：使用大量的if-elif
def calculate_discount_old(customer_type, amount):
if customer_type == "normal":
return amount
elif customer_type == "vip":
return amount * 0.9
elif customer_type == "svip":
return amount * 0.8
elif customer_type == "internal":
return amount * 0.5
else:
return amount
# 策略模式方式
from abc import ABC, abstractmethod
class DiscountStrategy(ABC):
@abstractmethod
def calculate(self, amount):
pass
class NormalDiscount(DiscountStrategy):
def calculate(self, amount):
return amount
class VIPDiscount(DiscountStrategy):
def calculate(self, amount):
return amount * 0.9
class SVIPDiscount(DiscountStrategy):
def calculate(self, amount):
return amount * 0.8
class InternalDiscount(DiscountStrategy):
def calculate(self, amount):
return amount * 0.5
class DiscountCalculator:
"""折扣计算器 - 上下文"""
_strategies = {
"normal": NormalDiscount(),
"vip": VIPDiscount(),
"svip": SVIPDiscount(),
"internal": InternalDiscount(),
}
@classmethod
def calculate(cls, customer_type, amount):
strategy = cls._strategies.get(customer_type)
if not strategy:
raise ValueError(f"未知的客户类型: {customer_type}")
return strategy.calculate(amount)
# 使用
print(f"普通客户: {DiscountCalculator.calculate('normal', 1000)}")
print(f"VIP客户: {DiscountCalculator.calculate('vip', 1000)}")
print(f"SVIP客户: {DiscountCalculator.calculate('svip', 1000)}")
print(f"内部员工: {DiscountCalculator.calculate('internal', 1000)}")
三、Python中的优雅实现
Python的动态特性让策略模式的实现更加简洁，我们可以利用函数、lambda和字典来实现策略模式。
1. 使用函数作为策略
代码示例
# 策略函数
def quick_sort(data):
"""快速排序策略"""
if len(data) <= 1:
return data
pivot = data[len(data) // 2]
left = [x for x in data if x < pivot]
middle = [x for x in data if x == pivot]
right = [x for x in data if x > pivot]
return quick_sort(left) + middle + quick_sort(right)
def selection_sort(data):
"""选择排序策略"""
arr = data.copy()
for i in range(len(arr)):
min_idx = i
for j in range(i+1, len(arr)):
if arr[j] < arr[min_idx]:
min_idx = j
arr[i], arr[min_idx] = arr[min_idx], arr[i]
return arr
def builtin_sort(data):
"""内置排序策略"""
return sorted(data)
# 策略上下文
class SortContext:
def __init__(self):
self._strategies = {}
def register(self, name, strategy_func):
"""注册策略"""
self._strategies[name] = strategy_func
def sort(self, strategy_name, data):
"""执行排序"""
strategy = self._strategies.get(strategy_name)
if not strategy:
raise ValueError(f"未注册的排序策略: {strategy_name}")
return strategy(data)
# 使用
ctx = SortContext()
ctx.register("quick", quick_sort)
ctx.register("selection", selection_sort)
ctx.register("builtin", builtin_sort)
data = [64, 34, 25, 12, 22, 11, 90]
print(f"快速排序: {ctx.sort('quick', data)}")
print(f"选择排序: {ctx.sort('selection', data)}")
print(f"内置排序: {ctx.sort('builtin', data)}")
2. 使用lambda表达式
代码示例
# 使用lambda实现简单的策略
strategies = {
"add": lambda x, y: x + y,
"subtract": lambda x, y: x - y,
"multiply": lambda x, y: x * y,
"divide": lambda x, y: x / y if y != 0 else float('inf'),
"power": lambda x, y: x ** y,
}
def calculate(operation, x, y):
"""执行计算"""
strategy = strategies.get(operation)
if not strategy:
raise ValueError(f"不支持的操作: {operation}")
return strategy(x, y)
print(f"10 + 5 = {calculate('add', 10, 5)}")
print(f"10 - 5 = {calculate('subtract', 10, 5)}")
print(f"10 × 5 = {calculate('multiply', 10, 5)}")
print(f"10 ÷ 5 = {calculate('divide', 10, 5)}")
print(f"10 ^ 5 = {calculate('power', 10, 5)}")
四、策略模式的进阶用法
1. 使用装饰器注册策略
代码示例
class StrategyRegistry:
"""策略注册表"""
def __init__(self):
self._strategies = {}
def register(self, name):
"""策略注册装饰器"""
def decorator(strategy_cls):
self._strategies[name] = strategy_cls
return strategy_cls
return decorator
def get(self, name):
"""获取策略实例"""
strategy_cls = self._strategies.get(name)
if not strategy_cls:
raise ValueError(f"未注册的策略: {name}")
return strategy_cls()
# 创建全局注册表
compressor_registry = StrategyRegistry()
# 注册压缩策略
@compressor_registry.register("zip")
class ZipCompressor:
def compress(self, data):
import zipfile
import io
buffer = io.BytesIO()
with zipfile.ZipFile(buffer, 'w', zipfile.ZIP_DEFLATED) as zf:
zf.writestr("data.txt", data)
return buffer.getvalue()
def name(self):
return "ZIP压缩"
@compressor_registry.register("gzip")
class GzipCompressor:
def compress(self, data):
import gzip
return gzip.compress(data.encode())
def name(self):
return "GZIP压缩"
@compressor_registry.register("bz2")
class Bz2Compressor:
def compress(self, data):
import bz2
return bz2.compress(data.encode())
def name(self):
return "BZ2压缩"
# 使用
def compress_data(algorithm, data):
compressor = compressor_registry.get(algorithm)
result = compressor.compress(data)
print(f"使用{compressor.name()}压缩，原始大小: {len(data)}字节，压缩后: {len(result)}字节")
return result
compress_data("gzip", "Hello, World! " * 100)
compress_data("bz2", "Hello, World! " * 100)
2. 策略组合 - 链式策略
代码示例
class TextFilter:
"""文本过滤器 - 策略基类"""
def apply(self, text):
raise NotImplementedError
class LowercaseFilter(TextFilter):
def apply(self, text):
return text.lower()
class TrimFilter(TextFilter):
def apply(self, text):
return text.strip()
class RemovePunctuationFilter(TextFilter):
def apply(self, text):
import re
return re.sub(r'[^\w\s]', '', text)
class RemoveStopwordsFilter(TextFilter):
stopwords = {"the", "a", "an", "is", "are", "was", "were", "in", "on", "at"}
def apply(self, text):
words = text.split()
return " ".join(w for w in words if w.lower() not in self.stopwords)
class TextPipeline:
"""文本处理管道 - 组合多个策略"""
def __init__(self):
self._filters = []
def add_filter(self, text_filter):
self._filters.append(text_filter)
return self
def process(self, text):
result = text
for f in self._filters:
result = f.apply(result)
return result
# 使用
pipeline = TextPipeline()
pipeline.add_filter(TrimFilter())
pipeline.add_filter(LowercaseFilter())
pipeline.add_filter(RemovePunctuationFilter())
pipeline.add_filter(RemoveStopwordsFilter())
text = " The Quick Brown Fox, jumps over the lazy dog! "
result = pipeline.process(text)
print(f"原始: '{text}'")
print(f"处理: '{result}'")
五、与条件分支的对比
六、实际应用场景
1. 路由策略 - Web框架
代码示例
class RouteStrategy:
"""路由策略基类"""
def match(self, path):
raise NotImplementedError
def handle(self, request):
raise NotImplementedError
class StaticRoute(RouteStrategy):
"""静态路由"""
def __init__(self, path, handler):
self.path = path
self.handler = handler
def match(self, path):
return path == self.path
def handle(self, request):
return self.handler(request)
class DynamicRoute(RouteStrategy):
"""动态路由（支持参数）"""
def __init__(self, pattern, handler):
import re
self.pattern = pattern
self.regex = re.compile(pattern)
self.handler = handler
def match(self, path):
return self.regex.match(path)
def handle(self, request):
match = self.regex.match(request["path"])
request["params"] = match.groupdict() if match.groupdict() else match.groups()
return self.handler(request)
class Router:
"""路由器 - 策略上下文"""
def __init__(self):
self._routes = []
def add_route(self, route):
self._routes.append(route)
def route(self, path):
"""装饰器注册路由"""
def decorator(handler):
if "{" in path:
import re
pattern = re.sub(r'\{(\w+)\}', r'(?P<\1>[^/]+)', path)
self.add_route(DynamicRoute(f"^{pattern}$", handler))
else:
self.add_route(StaticRoute(path, handler))
return handler
return decorator
def handle_request(self, request):
for route in self._routes:
if route.match(request["path"]):
return route.handle(request)
return {"status": 404, "body": "Not Found"}
# 使用
router = Router()
@router.route("/home")
def home_handler(request):
return {"status": 200, "body": "Welcome Home"}
@router.route("/users/{user_id}")
def user_handler(request):
return {"status": 200, "body": f"User {request['params']['user_id']}"}
@router.route("/posts/{post_id}/comments/{comment_id}")
def comment_handler(request):
params = request['params']
return {"status": 200, "body": f"Post {params[0]}, Comment {params[1]}"}
# 测试请求
print(router.handle_request({"path": "/home"}))
print(router.handle_request({"path": "/users/123"}))
print(router.handle_request({"path": "/posts/456/comments/789"}))
2. 支付策略
代码示例
from abc import ABC, abstractmethod
class PaymentStrategy(ABC):
@abstractmethod
def pay(self, amount):
pass
class CreditCardPayment(PaymentStrategy):
def __init__(self, card_number, cvv):
self.card_number = card_number
self.cvv = cvv
def pay(self, amount):
# 模拟信用卡支付
masked = f"****{self.card_number[-4:]}"
print(f"信用卡支付 {amount}元，卡号: {masked}")
return {"status": "success", "method": "credit_card", "amount": amount}
class PayPalPayment(PaymentStrategy):
def __init__(self, email):
self.email = email
def pay(self, amount):
print(f"PayPal支付 {amount}元，账户: {self.email}")
return {"status": "success", "method": "paypal", "amount": amount}
class CryptoPayment(PaymentStrategy):
def __init__(self, wallet_address):
self.wallet_address = wallet_address
def pay(self, amount):
print(f"加密货币支付 {amount}元，钱包: {self.wallet_address[:10]}...")
return {"status": "success", "method": "crypto", "amount": amount}
class ShoppingCart:
"""购物车 - 策略上下文"""
def __init__(self):
self._items = []
self._payment_strategy = None
def add_item(self, name, price):
self._items.append({"name": name, "price": price})
@property
def total(self):
return sum(item["price"] for item in self._items)
def set_payment(self, strategy: PaymentStrategy):
self._payment_strategy = strategy
def checkout(self):
if not self._payment_strategy:
raise ValueError("请选择支付方式")
total = self.total
print(f"订单总额: {total}元")
result = self._payment_strategy.pay(total)
self._items.clear()
return result
# 使用
cart = ShoppingCart()
cart.add_item("Python教程", 99.0)
cart.add_item("Django实战", 129.0)
# 选择不同支付方式
cart.set_payment(CreditCardPayment("1234567890123456", "123"))
cart.checkout()
cart.set_payment(PayPalPayment("user@example.com"))
cart.checkout()
小贴士
Python中很多内置函数天然支持策略模式。例如sorted()函数的key参数就是策略模式的体现——你可以传入任意排序策略函数。类似地，map()、filter()等函数也体现了策略模式的思想。
七、小结与练习题
核心要点总结
-
核心思想：将算法封装为可互换的策略，运行时动态选择
-
消除条件分支：用策略字典/注册表替代冗长的if-elif链
-
Python优势：函数作为一等公民，让策略模式实现更加简洁
-
策略组合：可以将多个策略组合成管道，实现复杂的数据处理流程
练习题
练习1
实现一个文件格式转换器，支持多种输出格式（PDF、HTML、Markdown、TXT），使用策略模式让新格式可以轻松扩展，并支持转换策略的组合使用。
练习2
设计一个物流费用计算器，支持不同快递公司（顺丰、圆通、中通、EMS）和不同配送方式（标准、加急、次日达）的计费策略，在实际场景中应用策略模式消除条件分支。
常见问题
什么时候应该使用策略模式而不是if-elif？
当你的条件分支超过3-4个，或者需要频繁添加新分支时，就应该考虑策略模式。另外，如果每个分支的逻辑比较复杂（超过10行代码），将其提取为独立策略类会让代码更清晰易维护。
策略模式会增加很多类文件吗？
在Python中不一定需要为每个策略创建独立类。可以使用函数、lambda、字典映射来简化实现。对于简单策略，一行lambda就足够了；对于复杂策略，使用类更利于封装和测试。
策略模式和状态模式有什么区别？
两者结构相似，但意图不同：策略模式关注的是算法的可替换性，策略之间是平等的、独立的选择；状态模式关注的是对象状态变化带来的行为变化，状态之间有明确的转换规则，且状态会影响对象的多个行为。
策略之间可以共享数据吗？
可以。上下文对象可以作为参数传递给策略方法，让不同策略访问共享数据。另一种方式是在策略构造函数中注入共享依赖。但要注意保持策略的独立性，避免策略之间产生隐式耦合。
标签：
Python
策略模式
设计模式
动态切换
消除条件分支
开闭原则
本文涉及AI创作
内容由AI创作，请仔细甄别
