微服务通信模式深度解析

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82

# 同步通信示例 - REST API
import requests
import json
from fastapi import FastAPI

app = FastAPI()

class OrderService:
    def __init__(self):
        self.user_service_url = "http://user-service:8001"
        self.payment_service_url = "http://payment-service:8002"
        self.inventory_service_url = "http://inventory-service:8003"

    def create_order(self, order_data):
        """同步创建订单"""
        # 验证用户
        user_response = requests.get(
            f"{self.user_service_url}/users/{order_data['user_id']}"
        )
        if user_response.status_code != 200:
            raise Exception("Invalid user")

        # 检查库存
        inventory_response = requests.post(
            f"{self.inventory_service_url}/check",
            json={"items": order_data["items"]}
        )
        if inventory_response.status_code != 200:
            raise Exception("Insufficient inventory")

        # 处理支付
        payment_response = requests.post(
            f"{self.payment_service_url}/pay",
            json={
                "amount": order_data["total"],
                "user_id": order_data["user_id"]
            }
        )
        if payment_response.status_code != 200:
            raise Exception("Payment failed")

        return {"order_id": "ORD-123", "status": "created"}

# 异步通信示例 - 消息队列
import asyncio
import aiokafka
import json

class AsyncOrderService:
    def __init__(self):
        self.order_events_topic = "order-events"
        self.producer = None

    async def start(self):
        """初始化生产者"""
        self.producer = aiokafka.AIOKafkaProducer(
            bootstrap_servers='kafka:9092',
            value_serializer=lambda v: json.dumps(v).encode('utf-8')
        )
        await self.producer.start()

    async def create_order_async(self, order_data):
        """异步创建订单"""
        # 创建订单记录
        order = await self.save_order(order_data)

        # 发布事件
        await self.publish_event({
            "event_type": "order_created",
            "order_id": order["id"],
            "user_id": order["user_id"],
            "items": order["items"]
        })

        return order

    async def publish_event(self, event):
        """发布事件"""
        await self.producer.send(
            topic=self.order_events_topic,
            value=event
        )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

# user_service.py - 用户服务
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from typing import List

app = FastAPI(title="User Service")

class User(BaseModel):
    id: int
    name: str
    email: str
    address: str

class UserService:
    def __init__(self):
        self.users_db = {}

    def get_user(self, user_id: int):
        """获取用户"""
        user = self.users_db.get(user_id)
        if not user:
            raise HTTPException(status_code=404, detail="User not found")
        return user

    def create_user(self, user: User):
        """创建用户"""
        if user.id in self.users_db:
            raise HTTPException(status_code=400, detail="User already exists")
        self.users_db[user.id] = user
        return user

user_service = UserService()

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    return user_service.get_user(user_id)

@app.post("/users")
async def create_user(user: User):
    return user_service.create_user(user)

# order_service.py - 订单服务（消费者）
class OrderService:
    def __init__(self):
        self.user_service = UserClient()

class UserClient:
    def __init__(self, base_url="http://user-service:8001"):
        self.base_url = base_url
        self.session = requests.Session()

    def get_user(self, user_id: int):
        """获取用户信息"""
        response = self.session.get(f"{self.base_url}/users/{user_id}")
        if response.status_code == 404:
            return None
        response.raise_for_status()
        return response.json()

    def update_user_credit(self, user_id: int, amount: float):
        """更新用户信用额度"""
        response = self.session.patch(
            f"{self.base_url}/users/{user_id}/credit",
            json={"amount": amount}
        )
        response.raise_for_status()
        return response.json()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

// user.proto - Protocol Buffers定义
syntax = "proto3";

package user;

service UserService {
  rpc GetUser(GetUserRequest) returns (UserResponse);
  rpc CreateUser(CreateUserRequest) returns (UserResponse);
  rpc UpdateUser(UpdateUserRequest) returns (UserResponse);
}

message GetUserRequest {
  int32 user_id = 1;
}

message CreateUserRequest {
  string name = 1;
  string email = 2;
  string address = 3;
}

message UpdateUserRequest {
  int32 user_id = 1;
  optional string name = 2;
  optional string email = 3;
  optional string address = 4;
}

message UserResponse {
  int32 id = 1;
  string name = 2;
  string email = 3;
  string address = 4;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71

# user_service_server.py - gRPC服务端
import grpc
from concurrent import futures
import user_pb2
import user_pb2_grpc

class UserServiceImpl(user_pb2_grpc.UserServiceServicer):
    def __init__(self):
        self.users = {}

    def GetUser(self, request, context):
        user = self.users.get(request.user_id)
        if not user:
            context.set_code(grpc.StatusCode.NOT_FOUND)
            context.set_details("User not found")
            return user_pb2.UserResponse()

        return user_pb2.UserResponse(
            id=user['id'],
            name=user['name'],
            email=user['email'],
            address=user['address']
        )

    def CreateUser(self, request, context):
        user_id = len(self.users) + 1
        user = {
            'id': user_id,
            'name': request.name,
            'email': request.email,
            'address': request.address
        }
        self.users[user_id] = user

        return user_pb2.UserResponse(**user)

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    user_pb2_grpc.add_UserServiceServicer_to_server(
        UserServiceImpl(), server
    )
    server.add_insecure_port('[::]:50051')
    server.start()
    server.wait_for_termination()

# order_service_client.py - gRPC客户端
import grpc
import user_pb2
import user_pb2_grpc

class OrderServiceClient:
    def __init__(self):
        self.channel = grpc.insecure_channel('user-service:50051')
        self.stub = user_pb2_grpc.UserServiceStub(self.channel)

    def get_user(self, user_id):
        """获取用户信息"""
        try:
            response = self.stub.GetUser(
                user_pb2.GetUserRequest(user_id=user_id)
            )
            return {
                'id': response.id,
                'name': response.name,
                'email': response.email,
                'address': response.address
            }
        except grpc.RpcError as e:
            if e.code() == grpc.StatusCode.NOT_FOUND:
                return None
            raise

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64

# event_store.py - 事件存储
from abc import ABC, abstractmethod
from typing import List, Dict, Any
from datetime import datetime
import asyncio
import json

class Event(ABC):
    def __init__(self, aggregate_id: str, event_id: str = None):
        self.aggregate_id = aggregate_id
        self.event_id = event_id or str(uuid.uuid4())
        self.timestamp = datetime.utcnow()

    def to_dict(self) -> Dict[str, Any]:
        return {
            'event_type': self.__class__.__name__,
            'aggregate_id': self.aggregate_id,
            'event_id': self.event_id,
            'timestamp': self.timestamp.isoformat(),
            'data': self.__dict__
        }

class OrderCreated(Event):
    def __init__(self, order_id: str, user_id: str, items: List[Dict]):
        super().__init__(order_id)
        self.order_id = order_id
        self.user_id = user_id
        self.items = items

class OrderPaid(Event):
    def __init__(self, order_id: str, payment_id: str, amount: float):
        super().__init__(order_id)
        self.order_id = order_id
        self.payment_id = payment_id
        self.amount = amount

# event_handler.py - 事件处理器
class EventHandler(ABC):
    @abstractmethod
    async def handle(self, event: Event):
        pass

class OrderEventHandler(EventHandler):
    async def handle(self, event: Event):
        if isinstance(event, OrderCreated):
            await self.handle_order_created(event)
        elif isinstance(event, OrderPaid):
            await self.handle_order_paid(event)

    async def handle_order_created(self, event: OrderCreated):
        """处理订单创建事件"""
        # 发送订单确认邮件
        await self.send_confirmation_email(event)
        # 更新库存
        await self.update_inventory(event)
        # 记录审计日志
        await self.log_audit_event(event)

    async def handle_order_paid(self, event: OrderPaid):
        """处理订单支付事件"""
        # 更新订单状态
        await self.update_order_status(event.order_id, 'paid')
        # 触发发货流程
        await self.initiate_shipping(event)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89

# message_broker.py - 消息代理
import aiokafka
import asyncio
import json
from typing import Dict, List, Callable

class MessageBroker:
    def __init__(self, bootstrap_servers: str):
        self.bootstrap_servers = bootstrap_servers
        self.producer = None
        self.consumers: Dict[str, List[Callable]] = {}

    async def start(self):
        """启动消息代理"""
        self.producer = aiokafka.AIOKafkaProducer(
            bootstrap_servers=self.bootstrap_servers,
            value_serializer=lambda v: json.dumps(v).encode('utf-8')
        )
        await self.producer.start()

    async def stop(self):
        """停止消息代理"""
        if self.producer:
            await self.producer.stop()

    async def publish(self, topic: str, message: Dict):
        """发布消息"""
        await self.producer.send(topic, message)

    def subscribe(self, topic: str, handler: Callable):
        """订阅消息"""
        if topic not in self.consumers:
            self.consumers[topic] = []
        self.consumers[topic].append(handler)

    async def start_consuming(self, topics: List[str]):
        """开始消费消息"""
        consumer = aiokafka.AIOKafkaConsumer(
            *topics,
            bootstrap_servers=self.bootstrap_servers,
            value_deserializer=lambda v: json.loads(v.decode('utf-8'))
        )
        await consumer.start()

        try:
            async for msg in consumer:
                handlers = self.consumers.get(msg.topic, [])
                for handler in handlers:
                    await handler(msg.value)
        finally:
            await consumer.stop()

# 使用示例
class OrderService:
    def __init__(self, broker: MessageBroker):
        self.broker = broker

    async def create_order(self, order_data: Dict):
        """创建订单"""
        # 保存订单到数据库
        order = await self.save_order(order_data)

        # 发布事件
        event = OrderCreated(
            order_id=order['id'],
            user_id=order['user_id'],
            items=order['items']
        )
        await self.broker.publish(
            topic='order-events',
            message=event.to_dict()
        )

        return order

class NotificationService:
    def __init__(self, broker: MessageBroker):
        self.broker = broker
        self.broker.subscribe('order-events', self.handle_order_event)

    async def handle_order_event(self, event_data: Dict):
        """处理订单事件"""
        event_type = event_data['event_type']
        data = event_data['data']

        if event_type == 'OrderCreated':
            await self.send_order_confirmation(data)
        elif event_type == 'OrderPaid':
            await self.send_payment_confirmation(data)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82

# api_gateway.py - API网关
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import Response
import httpx
import asyncio
from typing import Dict

class APIGateway:
    def __init__(self):
        self.app = FastAPI(title="API Gateway")
        self.services = {
            'user': 'http://user-service:8001',
            'order': 'http://order-service:8002',
            'payment': 'http://payment-service:8003',
            'inventory': 'http://inventory-service:8004'
        }
        self.setup_routes()

    def setup_routes(self):
        """设置路由"""
        @self.app.api_route("/{service}/{path:path}", methods=["GET", "POST", "PUT", "DELETE"])
        async def proxy_request(service: str, path: str, request: Request):
            if service not in self.services:
                raise HTTPException(status_code=404, detail="Service not found")

            service_url = self.services[service]
            url = f"{service_url}/{path}"

            # 转发请求
            async with httpx.AsyncClient() as client:
                response = await client.request(
                    method=request.method,
                    url=url,
                    headers=dict(request.headers),
                    content=await request.body(),
                    params=request.query_params
                )

            return Response(
                content=response.content,
                status_code=response.status_code,
                headers=dict(response.headers)
            )

        @self.app.post("/orders")
        async def create_order_aggregated(request: Request):
            """聚合创建订单"""
            order_data = await request.json()

            # 验证用户
            async with httpx.AsyncClient() as client:
                user_response = await client.get(
                    f"{self.services['user']}/users/{order_data['user_id']}"
                )
                if user_response.status_code != 200:
                    raise HTTPException(status_code=400, detail="Invalid user")

            # 检查库存
            async with httpx.AsyncClient() as client:
                inventory_response = await client.post(
                    f"{self.services['inventory']}/check",
                    json={"items": order_data["items"]}
                )
                if inventory_response.status_code != 200:
                    raise HTTPException(status_code=400, detail="Insufficient inventory")

            # 创建订单
            async with httpx.AsyncClient() as client:
                order_response = await client.post(
                    f"{self.services['order']}/orders",
                    json=order_data
                )

            return Response(
                content=order_response.content,
                status_code=order_response.status_code,
                headers=dict(order_response.headers)
            )

# 启动网关
gateway = APIGateway()
app = gateway.app

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62

# auth_middleware.py - 认证中间件
import jwt
import functools
from fastapi import HTTPException, Request, Depends
from fastapi.security import HTTPBearer

security = HTTPBearer()

class AuthMiddleware:
    def __init__(self, secret_key: str):
        self.secret_key = secret_key

    def verify_token(self, token: str) -> Dict:
        """验证JWT令牌"""
        try:
            payload = jwt.decode(token, self.secret_key, algorithms=["HS256"])
            return payload
        except jwt.ExpiredSignatureError:
            raise HTTPException(status_code=401, detail="Token has expired")
        except jwt.InvalidTokenError:
            raise HTTPException(status_code=401, detail="Invalid token")

    def require_auth(self, required_permissions: List[str] = None):
        """权限检查装饰器"""
        def decorator(func):
            @functools.wraps(func)
            async def wrapper(request: Request, *args, **kwargs):
                # 提取令牌
                authorization = request.headers.get("Authorization")
                if not authorization:
                    raise HTTPException(status_code=401, detail="No token provided")

                token = authorization.split(" ")[1]

                # 验证令牌
                payload = self.verify_token(token)

                # 检查权限
                if required_permissions:
                    user_permissions = payload.get("permissions", [])
                    if not all(perm in user_permissions for perm in required_permissions):
                        raise HTTPException(status_code=403, detail="Insufficient permissions")

                # 将用户信息添加到请求上下文
                request.state.user = payload

                return await func(request, *args, **kwargs)

            return wrapper
        return decorator

# 使用示例
gateway = APIGateway()
auth = AuthMiddleware(secret_key="your-secret-key")

@gateway.app.post("/admin/orders")
@auth.require_auth(["admin:orders"])
async def admin_create_order(request: Request):
    """管理员创建订单"""
    user = request.state.user
    # 管理员逻辑
    pass

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

# istio-gateway.yaml - 网关配置
apiVersion: networking.istio.io/v1beta1
kind: Gateway
metadata:
  name: myapp-gateway
spec:
  selector:
    istio: ingressgateway
  servers:
  - port:
      number: 80
      name: http
      protocol: HTTP
    hosts:
    - myapp.example.com
  - port:
      number: 443
      name: https
      protocol: HTTPS
    tls:
      mode: SIMPLE
      credentialName: myapp-tls
    hosts:
    - myapp.example.com

---
apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: myapp-vs
spec:
  hosts:
  - myapp.example.com
  gateways:
  - myapp-gateway
  http:
  - match:
    - uri:
        prefix: "/api/v1/users"
    route:
    - destination:
        host: user-service
        port:
          number: 8000
    timeout: 5s
    retries:
      attempts: 3
      perTryTimeout: 2s
  - match:
    - uri:
        prefix: "/api/v1/orders"
    route:
    - destination:
        host: order-service
        port:
          number: 8000
    fault:
      delay:
        percentage:
          value: 0.1
        fixedDelay: 5s

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

# destination-rule.yaml - 目标规则
apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: user-service-dr
spec:
  host: user-service
  trafficPolicy:
    connectionPool:
      tcp:
        maxConnections: 100
      http:
        http1MaxPendingRequests: 50
        maxRequestsPerConnection: 10
    loadBalancer:
      simple: LEAST_CONN
    outlierDetection:
      consecutiveErrors: 3
      interval: 30s
      baseEjectionTime: 30s
---
apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: order-service-dr
spec:
  host: order-service
  trafficPolicy:
    connectionPool:
      tcp:
        maxConnections: 50
      http:
        http1MaxPendingRequests: 25
    circuitBreaker:
      consecutiveErrors: 5
      interval: 30s
      baseEjectionTime: 30s

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

# connection_pool.py - 连接池配置
import httpx
import asyncio
from typing import Optional

class ServiceClient:
    def __init__(self, base_url: str):
        self.base_url = base_url
        self.client = None

    async def __aenter__(self):
        self.client = httpx.AsyncClient(
            base_url=self.base_url,
            limits=httpx.Limits(
                max_keepalive_connections=20,
                max_connections=100,
                keepalive_expiry=30
            ),
            timeout=httpx.Timeout(5.0, read=30.0)
        )
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.client.aclose()

    async def get(self, path: str, params: Optional[Dict] = None):
        """GET请求"""
        response = await self.client.get(path, params=params)
        response.raise_for_status()
        return response.json()

    async def post(self, path: str, data: Optional[Dict] = None):
        """POST请求"""
        response = await self.client.post(path, json=data)
        response.raise_for_status()
        return response.json()

# 使用示例
class OrderService:
    async def get_user_info(self, user_id: int):
        """获取用户信息"""
        async with ServiceClient("http://user-service:8001") as client:
            return await client.get(f"/users/{user_id}")

    async def create_payment(self, order_data: Dict):
        """创建支付"""
        async with ServiceClient("http://payment-service:8002") as client:
            return await client.post("/payments", order_data)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

# cache_layer.py - 缓存层
import redis
import json
import hashlib
from functools import wraps
from typing import Any, Optional

class CacheLayer:
    def __init__(self, redis_url: str):
        self.redis_client = redis.from_url(redis_url)

    def cache_result(self, ttl: int = 300):
        """缓存装饰器"""
        def decorator(func):
            @wraps(func)
            async def wrapper(*args, **kwargs):
                # 生成缓存键
                cache_key = self._generate_cache_key(
                    func.__name__, args, kwargs
                )

                # 尝试从缓存获取
                cached_result = self.redis_client.get(cache_key)
                if cached_result:
                    return json.loads(cached_result)

                # 执行函数
                result = await func(*args, **kwargs)

                # 存入缓存
                self.redis_client.setex(
                    cache_key,
                    ttl,
                    json.dumps(result, default=str)
                )

                return result
            return wrapper
        return decorator

    def _generate_cache_key(self, func_name: str, args, kwargs) -> str:
        """生成缓存键"""
        key_data = f"{func_name}:{str(args)}:{str(kwargs)}"
        return hashlib.md5(key_data.encode()).hexdigest()

    def invalidate_pattern(self, pattern: str):
        """使匹配模式的所有缓存失效"""
        keys = self.redis_client.keys(pattern)
        if keys:
            self.redis_client.delete(*keys)

# 使用示例
cache = CacheLayer("redis://localhost:6379")

class UserService:
    @cache.cache_result(ttl=600)
    async def get_user(self, user_id: int):
        """获取用户（带缓存）"""
        # 实际的数据库查询
        pass

    @cache.cache_result(ttl=3600)
    async def get_user_preferences(self, user_id: int):
        """获取用户偏好（带缓存）"""
        # 实际的数据库查询
        pass

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71

# tracing.py - 分布式追踪
from opentelemetry import trace
from opentelemetry.exporter.jaeger.thrift import JaegerExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
import functools

class TracingManager:
    def __init__(self, service_name: str):
        self.service_name = service_name
        self.setup_tracing()

    def setup_tracing(self):
        """设置追踪"""
        trace.set_tracer_provider(TracerProvider())
        tracer = trace.get_tracer(__name__)

        # 配置Jaeger导出器
        jaeger_exporter = JaegerExporter(
            agent_host_name="jaeger-agent",
            agent_port=6831,
        )

        span_processor = BatchSpanProcessor(jaeger_exporter)
        trace.get_tracer_provider().add_span_processor(span_processor)

        # 自动instrumentation
        HTTPXClientInstrumentor().instrument()
        FastAPIInstrumentor.instrument()

    def trace_function(self, name: str = None):
        """函数追踪装饰器"""
        def decorator(func):
            @functools.wraps(func)
            async def wrapper(*args, **kwargs):
                tracer = trace.get_tracer(__name__)
                span_name = name or f"{func.__module__}.{func.__name__}"

                with tracer.start_as_current_span(span_name) as span:
                    span.set_attribute("service.name", self.service_name)
                    span.set_attribute("function.name", func.__name__)

                    try:
                        result = await func(*args, **kwargs)
                        span.set_attribute("success", True)
                        return result
                    except Exception as e:
                        span.set_attribute("success", False)
                        span.set_attribute("error.message", str(e))
                        raise

            return wrapper
        return decorator

# 使用示例
tracing = TracingManager("order-service")

class OrderService:
    @tracing.trace_function("create_order")
    async def create_order(self, order_data: Dict):
        """创建订单（带追踪）"""
        # 创建订单逻辑
        pass

    @tracing.trace_function("process_payment")
    async def process_payment(self, order_id: str, amount: float):
        """处理支付（带追踪）"""
        # 支付处理逻辑
        pass