1
# 违反 SRP：类承担多个职责
2
class UserManager:
3
    def create_user(self, name, email):
4
        # 用户创建
5
        pass
6

7
    def send_email(self, user, message):
8
        # 发送邮件
9
        pass
10

11
    def generate_report(self):
12
        # 生成报告
13
        pass
14

15
# 符合 SRP：职责分离
16
class UserService:
17
    def create_user(self, name, email):
18
        pass
19

20
class EmailService:
21
    def send_email(self, user, message):
22
        pass
23

24
class ReportService:
25
    def generate_report(self):
26
        pass

1
# 违反 OCP：新增类型需要修改现有代码
2
class DiscountCalculator:
3
    def calculate(self, customer_type, amount):
4
        if customer_type == "regular":
5
            return amount * 0.9
6
        elif customer_type == "vip":
7
            return amount * 0.8
8
        # 新增类型需要修改这里
9

10
# 符合 OCP：通过扩展而非修改
11
from abc import ABC, abstractmethod
12

13
class DiscountStrategy(ABC):
14
    @abstractmethod
15
    def calculate(self, amount: float) -> float:
16
        pass
17

18
class RegularDiscount(DiscountStrategy):
19
    def calculate(self, amount: float) -> float:
20
        return amount * 0.9
21

22
class VIPDiscount(DiscountStrategy):
23
    def calculate(self, amount: float) -> float:
24
        return amount * 0.8
25

26
# 新增类型只需添加新类，无需修改现有代码
27
class SVIPDiscount(DiscountStrategy):
28
    def calculate(self, amount: float) -> float:
29
        return amount * 0.7

1
# 同步通信：REST/gRPC
2
class UserService:
3
    def __init__(self, http_client):
4
        self.http = http_client
5

6
    def get_order(self, order_id):
7
        response = self.http.get(f"/orders/{order_id}")
8
        return response.json()
9

10
# 异步通信：消息队列
11
class OrderService:
12
    def __init__(self, message_queue):
13
        self.queue = message_queue
14

15
    def create_order(self, order):
16
        self.queue.publish("order.created", order)

1
# 客户端发现
2
apiVersion: v1
3
kind: Service
4
metadata:
5
  name: user-service
6
spec:
7
  selector:
8
    app: user
9
  ports:
10
    - port: 80
11
      targetPort: 8080
12

13
---
14

15
apiVersion: v1
16
kind: Service
17
metadata:
18
  name: api-gateway
19
spec:
20
  type: NodePort
21
  ports:
22
    - port: 80
23
      targetPort: 8080

1
# 响应式系统特性
2
class ReactiveSystem:
3
    def __init__(self):
4
        self.principles = {
5
            "responsive": "及时响应用户",
6
            "resilient": "故障时仍可响应",
7
            "elastic": "负载变化时弹性伸缩",
8
            "message_driven": "异步通信松耦合"
9
        }

1
# CQRS 示例
2
class CommandHandler:
3
    def handle(self, command):
4
        """处理写操作"""
5
        if isinstance(command, CreateOrderCommand):
6
            return self.create_order(command)
7
        elif isinstance(command, UpdateOrderCommand):
8
            return self.update_order(command)
9

10
class QueryHandler:
11
    def handle(self, query):
12
        """处理读操作"""
13
        if isinstance(query, GetOrderQuery):
14
            return self.get_order(query.order_id)
15
        elif isinstance(query, ListOrdersQuery):
16
            return self.list_orders(query.filters)