1688 平台商品详情接口技术揭秘：架构演进与实战优化

# 数据访问层示例：分库分表实现 class ShardingDBRouter:     """分库分表路由"""     def __init__(self, shard_count=8):         self.shard_count = shard_count          def get_db_key(self, product_id):         """根据商品ID计算数据库分片"""         hash_value = hash(product_id)         return f"db_{hash_value % self.shard_count}"          def get_table_key(self, product_id):         """根据商品ID计算数据表分片"""         hash_value = hash(product_id)         return f"product_{hash_value % self.shard_count}"          def execute_query(self, product_id, query, params=None):         """执行分片查询"""         db_key = self.get_db_key(product_id)         table_key = self.get_table_key(product_id)                  # 根据db_key获取对应的数据库连接         db_connection = self._get_db_connection(db_key)                  # 替换SQL中的表名         query = query.replace("{table}", table_key)                  # 执行查询         return db_connection.execute(query, params)

class ProductBaseInfo:     """商品基础信息模型"""     def __init__(self,                   product_id: str,                  title: str,                  category_id: int,                  brand_id: str,                  supplier_id: str,                  keywords: list,                  description: str,                  create_time: datetime,                  update_time: datetime):         self.product_id = product_id           # 商品ID         self.title = title                     # 商品标题         self.category_id = category_id         # 类目ID         self.brand_id = brand_id               # 品牌ID         self.supplier_id = supplier_id         # 供应商ID         self.keywords = keywords               # 关键词列表         self.description = description         # 商品描述         self.create_time = create_time         # 创建时间         self.update_time = update_time         # 更新时间

class ProductTradeInfo:     """商品交易信息模型"""     def __init__(self,                  product_id: str,                  price: float,                  original_price: float,                  min_order_quantity: int,                  available_quantity: int,                  packaging: str,                  delivery_time: str,                  payment_terms: list,                  logistics_options: list):         self.product_id = product_id                 # 商品ID         self.price = price                           # 当前价格         self.original_price = original_price         # 原价         self.min_order_quantity = min_order_quantity # 最小起订量         self.available_quantity = available_quantity # 可用库存         self.packaging = packaging                   # 包装规格         self.delivery_time = delivery_time           # 发货时间         self.payment_terms = payment_terms           # 支付方式         self.logistics_options = logistics_options   # 物流选项

class ProductPromotionInfo:     """商品营销信息模型"""     def __init__(self,                  product_id: str,                  promotion_id: str,                  promotion_type: str,                  discount_rate: float,                  start_time: datetime,                  end_time: datetime,                  conditions: dict,                  benefits: dict):         self.product_id = product_id           # 商品ID         self.promotion_id = promotion_id       # 促销活动ID         self.promotion_type = promotion_type   # 促销类型         self.discount_rate = discount_rate     # 折扣率         self.start_time = start_time           # 开始时间         self.end_time = end_time               # 结束时间         self.conditions = conditions           # 参与条件         self.benefits = benefits               # 优惠内容

import redis from cachetools import TTLCache, LRUCache import json from typing import Optional, Dict, Any  class CacheService:     """缓存服务"""     def __init__(self):         # 本地进程缓存，使用LRU策略，容量1000，TTL 60秒         self.local_cache = LRUCache(maxsize=1000)                  # 分布式缓存         self.redis_client = redis.Redis(             host='redis-cluster',             port=6379,             password='your_password',             decode_responses=True         )                  # 缓存键前缀         self.PRODUCT_DETAIL_PREFIX = "product:detail:"         self.PRODUCT_PRICE_PREFIX = "product:price:"         self.PRODUCT_STOCK_PREFIX = "product:stock:"          def get_product_detail(self, product_id: str) -> Optional[Dict[str, Any]]:         """获取商品详情缓存"""         # 1. 检查本地缓存         cache_key = f"{self.PRODUCT_DETAIL_PREFIX}{product_id}"         result = self.local_cache.get(cache_key)         if result:             return result                  # 2. 检查Redis缓存         result = self.redis_client.get(cache_key)         if result:             result = json.loads(result)             # 更新本地缓存             self.local_cache[cache_key] = result             return result                  return None          def set_product_detail(self, product_id: str, data: Dict[str, Any], ttl: int = 300) -> None:         """设置商品详情缓存"""         cache_key = f"{self.PRODUCT_DETAIL_PREFIX}{product_id}"                  # 转换为JSON格式         json_data = json.dumps(data)                  # 同时设置本地缓存和Redis缓存         self.local_cache[cache_key] = data         self.redis_client.setex(cache_key, ttl, json_data)          def delete_product_cache(self, product_id: str) -> None:         """删除商品相关缓存"""         keys = [             f"{self.PRODUCT_DETAIL_PREFIX}{product_id}",             f"{self.PRODUCT_PRICE_PREFIX}{product_id}",             f"{self.PRODUCT_STOCK_PREFIX}{product_id}"         ]                  # 删除本地缓存         for key in keys:             if key in self.local_cache:                 del self.local_cache[key]                  # 删除Redis缓存         self.redis_client.delete(*keys)

import asyncio import aiohttp from typing import Dict, Any, List  class AsyncDataFetcher:     """异步数据获取器"""     async def fetch_product_data(self, product_id: str) -> Dict[str, Any]:         """并发获取商品多维度数据"""         async with aiohttp.ClientSession() as session:             # 创建并发任务             tasks = [                 self._fetch_base_info(session, product_id),                 self._fetch_trade_info(session, product_id),                 self._fetch_promotion_info(session, product_id),                 self._fetch_review_summary(session, product_id),                 self._fetch_supplier_info(session, product_id)             ]                          # 并发执行任务             base_info, trade_info, promotion_info, review_summary, supplier_info = await asyncio.gather(*tasks)                          # 组装数据             return {                 'base_info': base_info,                 'trade_info': trade_info,                 'promotion_info': promotion_info,                 'review_summary': review_summary,                 'supplier_info': supplier_info             }          async def _fetch_base_info(self, session: aiohttp.ClientSession, product_id: str) -> Dict[str, Any]:         """获取商品基础信息"""         async with session.get(f'http://product-service/api/products/{product_id}/base_info') as response:             return await response.json()          async def _fetch_trade_info(self, session: aiohttp.ClientSession, product_id: str) -> Dict[str, Any]:         """获取商品交易信息"""         async with session.get(f'http://trade-service/api/products/{product_id}/trade_info') as response:             return await response.json()          async def _fetch_promotion_info(self, session: aiohttp.ClientSession, product_id: str) -> Dict[str, Any]:         """获取商品促销信息"""         async with session.get(f'http://promotion-service/api/products/{product_id}/promotions') as response:             return await response.json()          async def _fetch_review_summary(self, session: aiohttp.ClientSession, product_id: str) -> Dict[str, Any]:         """获取商品评价摘要"""         async with session.get(f'http://review-service/api/products/{product_id}/review_summary') as response:             return await response.json()          async def _fetch_supplier_info(self, session: aiohttp.ClientSession, product_id: str) -> Dict[str, Any]:         """获取供应商信息"""         async with session.get(f'http://supplier-service/api/products/{product_id}/supplier') as response:             return await response.json()

from flask import Flask, request, jsonify from flask_restful import Api, Resource from service.product_service import ProductService from utils.cache import CacheService from utils.decorators import rate_limit, validate_params  app = Flask(__name__) api = Api(app) cache_service = CacheService()  class ProductDetailAPI(Resource):     """商品详情API"""          @rate_limit(limit=100, period=60)  # 限流：每分钟100次请求     @validate_params(['product_id'])     def get(self):         """获取商品详情"""         product_id = request.args.get('product_id')         user_id = request.args.get('user_id', '')                  # 优先从缓存获取         cache_data = cache_service.get_product_detail(product_id)         if cache_data:             return jsonify(cache_data)                  # 缓存未命中，从各服务获取数据         product_service = ProductService()         result = product_service.get_product_detail(product_id, user_id)                  # 设置缓存         cache_service.set_product_detail(product_id, result)                  return jsonify(result)  api.add_resource(ProductDetailAPI, '/api/v1/products/<product_id>/detail')  if __name__ == '__main__':     app.run(host='0.0.0.0', port=8080)

from typing import Dict, Any from utils.async_fetcher import AsyncDataFetcher from repository.product_repo import ProductRepository from repository.sku_repo import SkuRepository  class ProductService:     """商品服务"""     def __init__(self):         self.async_fetcher = AsyncDataFetcher()         self.product_repo = ProductRepository()         self.sku_repo = SkuRepository()          async def get_product_detail(self, product_id: str, user_id: str = '') -> Dict[str, Any]:         """获取商品详情"""         # 1. 并发获取基础数据         product_data = await self.async_fetcher.fetch_product_data(product_id)                  # 2. 获取SKU信息         sku_list = self.sku_repo.get_skus_by_product_id(product_id)                  # 3. 获取用户个性化信息（如历史浏览、收藏等）         user_info = self._get_user_personalization(user_id, product_id)                  # 4. 整合数据         result = {             'product_id': product_id,             'base_info': product_data['base_info'],             'trade_info': product_data['trade_info'],             'promotion_info': product_data['promotion_info'],             'review_summary': product_data['review_summary'],             'supplier_info': product_data['supplier_info'],             'sku_list': sku_list,             'user_personalization': user_info,             'timestamp': int(time.time())         }                  return result          def _get_user_personalization(self, user_id: str, product_id: str) -> Dict[str, Any]:         """获取用户个性化信息"""         if not user_id:             return {}                  # 实际项目中会调用用户服务获取个性化信息         # 这里简化处理，返回示例数据         return {             'is_favorite': False,             'browsed_count': 3,             'last_browsed_time': '2025-07-25 14:30:22',             'recommended_price': 99.9         }

from apscheduler.schedulers.background import BackgroundScheduler  class CacheWarmUpService:     """缓存预热服务"""     def __init__(self, cache_service, product_service):         self.cache_service = cache_service         self.product_service = product_service         self.scheduler = BackgroundScheduler()                  # 注册定时任务         self.scheduler.add_job(             self.warm_up_top_products,              'interval',              hours=1,              id='cache_warm_up'         )          def start(self):         """启动缓存预热服务"""         self.scheduler.start()          def warm_up_top_products(self):         """预热热门商品缓存"""         try:             # 获取热门商品ID列表             top_product_ids = self._get_top_product_ids()                          # 并发预热缓存             loop = asyncio.get_event_loop()             tasks = [self._warm_up_product(product_id) for product_id in top_product_ids]             loop.run_until_complete(asyncio.gather(*tasks))                          logging.info(f"Successfully warmed up {len(top_product_ids)} product caches")         except Exception as e:             logging.error(f"Cache warm up failed: {str(e)}")          async def _warm_up_product(self, product_id: str):         """预热单个商品缓存"""         product_data = await self.product_service.get_product_detail(product_id)         self.cache_service.set_product_detail(product_id, product_data, ttl=1800)          def _get_top_product_ids(self, limit: int = 100) -> List[str]:         """获取热门商品ID列表"""         # 实际项目中会从热门商品排行榜获取         # 这里简化处理，返回示例数据         return [f"product_{i}" for i in range(1, limit+1)]

from sentinel_python.client import SentinelClient from sentinel_python.core.entry import SphU from sentinel_python.core.slots.block import BlockException  # 初始化Sentinel客户端 sentinel_client = SentinelClient(     app_name="product-service",     sentinel_server="sentinel-server:8719" )  # 定义资源 PRODUCT_DETAIL_RESOURCE = "product_detail"  class ProductService:     # ... 其他代码 ...          async def get_product_detail(self, product_id: str, user_id: str = ''):         try:             # 资源保护             with SphU.entry(resource=PRODUCT_DETAIL_RESOURCE):                 return await self._get_product_detail_internal(product_id, user_id)         except BlockException as e:             # 触发熔断或限流时的降级处理             logging.warning(f"Request blocked: {e}")             return self._get_product_detail_fallback(product_id)          def _get_product_detail_fallback(self, product_id: str):         """熔断降级处理"""         # 从本地缓存或只读副本获取基础信息         cache_data = self.cache_service.get_product_detail(product_id)         if cache_data:             # 移除可能过时的数据             cache_data.pop('trade_info', None)             cache_data.pop('promotion_info', None)             return cache_data                  # 返回默认数据         return {             'product_id': product_id,             'base_info': {'name': '商品信息加载中', 'description': '商品信息暂时不可用'},             'error': 'Service temporarily unavailable, please try again later',             'timestamp': int(time.time())         }

import hmac import hashlib import time  class ApiSignature:     """API签名验证"""     def __init__(self, secret_key):         self.secret_key = secret_key          def generate_signature(self, params: dict, timestamp: int) -> str:         """生成签名"""         # 1. 排序参数         sorted_params = sorted(params.items(), key=lambda x: x[0])                  # 2. 拼接参数         param_str = '&'.join([f"{k}={v}" for k, v in sorted_params])                  # 3. 添加时间戳         string_to_sign = f"{param_str}&timestamp={timestamp}"                  # 4. HMAC-SHA256加密         signature = hmac.new(             self.secret_key.encode(),             string_to_sign.encode(),             hashlib.sha256         ).hexdigest()                  return signature          def verify_signature(self, params: dict, signature: str, timestamp: int) -> bool:         """验证签名"""         # 检查时间戳有效性（防止重放攻击）         current_time = int(time.time())         if abs(current_time - timestamp) > 300:  # 超过5分钟             return False                  # 生成预期签名         expected_signature = self.generate_signature(params, timestamp)                  # 比较签名         return hmac.compare_digest(expected_signature, signature)

class DataMasking:     """数据脱敏处理"""     @staticmethod     def mask_phone(phone: str) -> str:         """脱敏手机号"""         if not phone or len(phone) < 11:             return phone         return f"{phone[:3]}****{phone[-4:]}"          @staticmethod     def mask_email(email: str) -> str:         """脱敏邮箱"""         if not email or '@' not in email:             return email         username, domain = email.split('@')         if len(username) <= 2:             masked_username = username[0] + '*' * (len(username) - 1)         else:             masked_username = username[0] + '*' * (len(username) - 2) + username[-1]         return f"{masked_username}@{domain}"          @staticmethod     def mask_id_card(id_card: str) -> str:         """脱敏身份证号"""         if not id_card or len(id_card) < 15:             return id_card         return f"{id_card[:6]}********{id_card[-4:]}"

from opentelemetry import trace from opentelemetry.exporter.jaeger.thrift import JaegerExporter from opentelemetry.sdk.resources import SERVICE_NAME, Resource from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor  # 配置Jaeger追踪 resource = Resource(attributes={SERVICE_NAME: "product-service"}) jaeger_exporter = JaegerExporter(     agent_host_name="jaeger-agent",     agent_port=6831, )  provider = TracerProvider(resource=resource) processor = BatchSpanProcessor(jaeger_exporter) provider.add_span_processor(processor) trace.set_tracer_provider(provider)  tracer = trace.get_tracer(__name__)  class ProductService:     # ... 其他代码 ...          async def get_product_detail(self, product_id: str, user_id: str = ''):         with tracer.start_as_current_span("get_product_detail") as span:             span.set_attribute("product_id", product_id)             span.set_attribute("user_id", user_id)                          try:                 # 1. 获取基础信息                 with tracer.start_as_current_span("fetch_base_info"):                     base_info = await self._fetch_base_info(product_id)                                  # 2. 获取交易信息                 with tracer.start_as_current_span("fetch_trade_info"):                     trade_info = await self._fetch_trade_info(product_id)                                  # 3. 获取促销信息                 with tracer.start_as_current_span("fetch_promotion_info"):                     promotion_info = await self._fetch_promotion_info(product_id)                                  # 4. 整合数据                 with tracer.start_as_current_span("assemble_data"):                     result = self._assemble_data(                         base_info, trade_info, promotion_info, product_id, user_id                     )                                  return result             except Exception as e:                 span.set_status(trace.Status(trace.StatusCode.ERROR, str(e)))                 raise

import sentry_sdk from sentry_sdk.integrations.flask import FlaskIntegration  # 初始化Sentry sentry_sdk.init(     dsn="https://your-sentry-dsn@o123456.ingest.sentry.io/678901",     integrations=[FlaskIntegration()],     traces_sample_rate=1.0, )  # 在关键业务逻辑中添加错误捕获 class ProductService:     # ... 其他代码 ...          async def get_product_detail(self, product_id: str, user_id: str = ''):         try:             return await self._get_product_detail_internal(product_id, user_id)         except Exception as e:             # 记录错误到Sentry             sentry_sdk.capture_exception(e)                          # 记录本地日志             logging.error(f"Failed to get product detail: {str(e)}", exc_info=True)                          # 返回友好错误信息             return {                 'error': 'Failed to fetch product information',                 'error_code': 'PRODUCT_FETCH_ERROR',                 'timestamp': int(time.time())             }