京东平台商品详情接口技术解密：高性能架构与实战经验

# 接口层示例代码 from flask import Flask, request, jsonify from flask_restful import Api, Resource from service.product_service import ProductService from utils.decorators import check_auth, validate_params  app = Flask(__name__) api = Api(app)  class ProductDetail(Resource):     """商品详情接口"""          @check_auth     @validate_params(['product_id'])     def get(self):         """获取商品详情"""         product_id = request.args.get('product_id')         user_id = request.args.get('user_id', '')                  # 调用业务逻辑层         product_service = ProductService()         result = product_service.get_product_detail(product_id, user_id)                  return jsonify(result)  api.add_resource(ProductDetail, '/api/v1/product/detail')  if __name__ == '__main__':     app.run(host='0.0.0.0', port=8080)

class Product:     """商品核心信息"""     def __init__(self, product_id, title, brand, category,                   description, images, specs, params):         self.product_id = product_id        # 商品ID         self.title = title                  # 商品标题         self.brand = brand                  # 品牌         self.category = category            # 分类         self.description = description      # 商品描述         self.images = images                # 图片列表         self.specs = specs                  # 规格信息         self.params = params                # 参数信息  class PriceInfo:     """商品价格信息"""     def __init__(self, product_id, sku_id, price, original_price,                   promotion_price, market_price, price_unit):         self.product_id = product_id        # 商品ID         self.sku_id = sku_id                # SKU ID         self.price = price                  # 销售价         self.original_price = original_price# 原价         self.promotion_price = promotion_price# 促销价         self.market_price = market_price    # 市场价         self.price_unit = price_unit        # 价格单位  class StockInfo:     """商品库存信息"""     def __init__(self, product_id, sku_id, stock_num, available_num,                   pre_sale, delivery_info, stock_status):         self.product_id = product_id        # 商品ID         self.sku_id = sku_id                # SKU ID         self.stock_num = stock_num          # 总库存         self.available_num = available_num  # 可用库存         self.pre_sale = pre_sale            # 是否预售         self.delivery_info = delivery_info  # 配送信息         self.stock_status = stock_status    # 库存状态

import redis from cachetools import TTLCache import json  class CacheManager:     """缓存管理器"""     def __init__(self):         # 本地缓存，使用LRU策略，容量1000，过期时间60秒         self.local_cache = TTLCache(maxsize=1000, ttl=60)                  # 分布式缓存         self.redis_client = redis.Redis(             host='redis_host',              port=6379,              db=0,             password='your_password'         )          def get(self, key):         """获取缓存数据"""         # 优先从本地缓存获取         value = self.local_cache.get(key)         if value is not None:             return value                  # 从Redis获取         value = self.redis_client.get(key)         if value:             value = json.loads(value)             # 放入本地缓存             self.local_cache[key] = value             return value                  return None          def set(self, key, value, expire=3600):         """设置缓存数据"""         # 转换为JSON格式存储         json_value = json.dumps(value)                  # 同时设置本地缓存和Redis缓存         self.local_cache[key] = value         self.redis_client.setex(key, expire, json_value)          def delete(self, key):         """删除缓存数据"""         if key in self.local_cache:             del self.local_cache[key]         self.redis_client.delete(key)

import asyncio from concurrent.futures import ThreadPoolExecutor  class AsyncDataLoader:     """异步数据加载器"""     def __init__(self):         self.executor = ThreadPoolExecutor(max_workers=10)          async def load_reviews(self, product_id, page=1, page_size=10):         """异步加载商品评价"""         loop = asyncio.get_running_loop()         return await loop.run_in_executor(             self.executor,              lambda: self._fetch_reviews(product_id, page, page_size)         )          async def load_recommendations(self, product_id, user_id):         """异步加载推荐商品"""         loop = asyncio.get_running_loop()         return await loop.run_in_executor(             self.executor,              lambda: self._fetch_recommendations(product_id, user_id)         )          def _fetch_reviews(self, product_id, page, page_size):         # 调用评价服务API         # 实际代码中会使用requests或其他HTTP客户端         return {             'total': 123,             'items': [                 {'id': 1, 'user': 'user1', 'score': 5, 'content': '非常好的商品'},                 {'id': 2, 'user': 'user2', 'score': 4, 'content': '质量不错'}             ]         }          def _fetch_recommendations(self, product_id, user_id):         # 调用推荐服务API         return [             {'id': 1001, 'title': '推荐商品1', 'price': 99.0},             {'id': 1002, 'title': '推荐商品2', 'price': 199.0}         ]

class ProductQueryService:     """商品查询服务"""     def __init__(self):         self.cache_manager = CacheManager()         self.async_loader = AsyncDataLoader()         self.product_repository = ProductRepository()         self.price_service = PriceService()         self.stock_service = StockService()         self.promotion_service = PromotionService()          async def get_product_detail(self, product_id, user_id=None):         """获取商品详情"""         # 优先从缓存获取         cache_key = f'product_detail:{product_id}'         result = self.cache_manager.get(cache_key)         if result:             return result                  # 从数据库获取基础信息         product = self.product_repository.get_by_id(product_id)         if not product:             raise ValueError(f"Product {product_id} not found")                  # 获取价格信息         price_info = self.price_service.get_price(product_id)                  # 获取库存信息         stock_info = self.stock_service.get_stock(product_id)                  # 获取促销信息         promotion_info = self.promotion_service.get_promotions(product_id, user_id)                  # 异步获取非关键信息         reviews_task = self.async_loader.load_reviews(product_id)         recommendations_task = self.async_loader.load_recommendations(product_id, user_id)                  reviews, recommendations = await asyncio.gather(reviews_task, recommendations_task)                  # 组装数据         result = {             'product_info': product.to_dict(),             'price_info': price_info.to_dict(),             'stock_info': stock_info.to_dict(),             'promotion_info': promotion_info,             'reviews': reviews,             'recommendations': recommendations         }                  # 设置缓存，有效期5分钟         self.cache_manager.set(cache_key, result, 300)                  return result

import pika import json  class EventBus:     """事件总线"""     def __init__(self, host, username, password):         credentials = pika.PlainCredentials(username, password)         self.connection = pika.BlockingConnection(             pika.ConnectionParameters(host=host, credentials=credentials)         )         self.channel = self.connection.channel()                  # 声明交换器         self.channel.exchange_declare(             exchange='product_events',              exchange_type='topic'         )          def publish_event(self, routing_key, event_data):         """发布事件"""         self.channel.basic_publish(             exchange='product_events',             routing_key=routing_key,             body=json.dumps(event_data),             properties=pika.BasicProperties(                 delivery_mode=2,  # 持久化消息             )         )          def close(self):         """关闭连接"""         self.connection.close()  # 商品信息变更事件处理示例 def handle_product_updated(ch, method, properties, body):     event_data = json.loads(body)     product_id = event_data.get('product_id')          # 清除相关缓存     cache_manager = CacheManager()     cache_manager.delete(f'product_detail:{product_id}')     cache_manager.delete(f'product_price:{product_id}')          # 记录日志     logging.info(f"Product {product_id} updated, cache cleared")

from flask_jwt_extended import (     JWTManager, jwt_required, create_access_token,     get_jwt_identity )  # 配置JWT app.config['JWT_SECRET_KEY'] = 'your-secret-key' jwt = JWTManager(app)  # 登录接口 @app.route('/api/auth/login', methods=['POST']) def login():     username = request.json.get('username', None)     password = request.json.get('password', None)          # 验证用户     if username != 'admin' or password != 'password':         return jsonify({"msg": "Bad credentials"}), 401          # 创建访问令牌     access_token = create_access_token(identity=username)     return jsonify(access_token=access_token), 200  # 需要鉴权的接口 @app.route('/api/private/product', methods=['GET']) @jwt_required() def get_private_product_info():     # 获取用户身份     current_user = get_jwt_identity()          # 根据用户权限返回不同级别的商品信息     product_id = request.args.get('product_id')     product_service = ProductService()          # 检查用户权限     if current_user == 'admin':         # 返回完整信息         return jsonify(product_service.get_admin_product_detail(product_id))     else:         # 返回普通用户可见信息         return jsonify(product_service.get_product_detail(product_id))

from cryptography.fernet import Fernet  class DataEncryptor:     """数据加密器"""     def __init__(self, encryption_key):         self.cipher_suite = Fernet(encryption_key)          def encrypt(self, data):         """加密数据"""         if isinstance(data, str):             data = data.encode()         return self.cipher_suite.encrypt(data).decode()          def decrypt(self, encrypted_data):         """解密数据"""         if isinstance(encrypted_data, str):             encrypted_data = encrypted_data.encode()         return self.cipher_suite.decrypt(encrypted_data).decode()  # 使用示例 encryptor = DataEncryptor("your-32-character-encryption-key") encrypted = encryptor.encrypt("sensitive data") decrypted = encryptor.decrypt(encrypted)

from prometheus_flask_exporter import PrometheusMetrics  # 初始化监控 metrics = PrometheusMetrics(app)  # 自定义指标 request_duration = metrics.histogram(     'http_request_duration_seconds', 'Request duration',     labels={'endpoint': lambda: request.endpoint} )  # 记录请求处理时间 @app.before_request def start_timer():     g.start_time = time.time()  @app.after_request def stop_timer(response):     if hasattr(g, 'start_time'):         request_time = time.time() - g.start_time         request_duration.labels(request.endpoint).observe(request_time)     return response

from sentinel_python.client import SentinelClient from sentinel_python.core.entry import SphU from sentinel_python.core.slots.resource_wrapper import ResourceTypeConstants  # 初始化Sentinel客户端 sentinel_client = SentinelClient(     app_name="product-service",     sentinel_server="sentinel-server:8719" )  # 定义资源 RESOURCE_KEY = "get_product_detail"  def get_product_detail(product_id, user_id=None):     # 流控检查     with SphU.entry(         resource=RESOURCE_KEY,          resource_type=ResourceTypeConstants.COMMON_API,         entry_type=EntryType.IN     ) as entry:         # 业务逻辑         product_service = ProductService()         return product_service.get_product_detail(product_id, user_id)     except BlockException as e:         # 被限流或熔断时的处理         return {             'code': 429,             'message': 'Too many requests, please try again later'         }