import timefrom typing import List, Dict, Anydef batch_query_sku_detail(sku_id_list: List[str]) -> List[Dict[str, Any]]:
"""批量分片查询京东商品详情"""
if not sku_id_list:
return []
batch_size = 500
result_list = []
partitions = [sku_id_list[i:i + batch_size] for i in range(0, len(sku_id_list), batch_size)]
for idx, part in enumerate(partitions):
resp = jd_open_api_client.batch_get_sku_detail(part, get_biz_field_filter())
if resp.get("success") and resp.get("data_list"):
result_list.extend(resp["data_list"])
# 多批次任务平滑限流，避免瞬时流量冲击
if len(partitions) > 1 and idx < len(partitions) - 1:
time.sleep(0.1)
return result_listdef get_biz_field_filter() -> str: """京东详情接口业务字段白名单""" return "skuId,title,price,stock,status,categoryId,saleNum"
from concurrent.futures import ThreadPoolExecutor# 前台实时查询线程池（核心高优先级）jd_api_real_time_pool = ThreadPoolExecutor( max_workers=32, thread_name_prefix="jd-real-time-pool")# 后台批量同步线程池（低优先级）jd_api_batch_pool = ThreadPoolExecutor( max_workers=16, thread_name_prefix="jd-batch-pool")
from cachetools import TTLCache# 京东SKU详情本地热点缓存# 最大容量10000、写入1小时过期、LRU淘汰策略jd_sku_local_cache = TTLCache(maxsize=10000, ttl=3600)
import jsonfrom typing import Optional, Dict, Any# 假设项目中已封装好对应的客户端、常量和缓存对象# from your_project.config import jd_sku_local_cache, redis_client, REDIS_KEY_JD_SKU_DETAIL# from your_project.utils import jd_open_api_client, get_biz_field_filterNULL_PLACEHOLDER = "NULL"def get_sku_detail_cache(sku_id: str) -> Optional[Dict[str, Any]]:
"""
二级缓存查询模板：本地缓存 -> Redis缓存 -> 远端API
自带空值缓存防穿透
"""
# 1. 优先查询本地缓存，微秒级响应
local_cache = jd_sku_local_cache.get(sku_id)
if local_cache is not None:
return local_cache # 2. 查询分布式Redis缓存
redis_key = f"{REDIS_KEY_JD_SKU_DETAIL}{sku_id}"
redis_value = redis_client.get(redis_key)
if redis_value is not None:
if redis_value == NULL_PLACEHOLDER:
return None
cache_dto = json.loads(redis_value)
jd_sku_local_cache[sku_id] = cache_dto return cache_dto # 3. 缓存未命中，穿透查询京东远端API
api_dto = jd_open_api_client.get_single_sku_detail(sku_id, get_biz_field_filter())
if api_dto is not None:
redis_client.set(redis_key, json.dumps(api_dto), ex=300)
jd_sku_local_cache[sku_id] = api_dto else:
redis_client.set(redis_key, NULL_PLACEHOLDER, ex=30)
return api_dtofrom concurrent.futures import ThreadPoolExecutor, Futurefrom typing import List, Dict, Any# 假设项目中已封装好对应的客户端、工具方法和线程池# from your_project.utils import jd_open_api_client, get_biz_field_filter, jd_api_real_time_pooldef async_batch_query(sku_part: List[str]) -> Future[List[Dict[str, Any]]]:
"""异步批量查询SKU详情"""
def _query():
resp = jd_open_api_client.batch_get_sku_detail(sku_part, get_biz_field_filter())
if resp.get("success"):
return resp.get("data_list", [])
return []
return jd_api_real_time_pool.submit(_query)import timefrom functools import wrapsfrom typing import Callable, Type, Tupledef jd_api_retry( max_attempts: int = 3, wait_duration: float = 1.0, retry_exceptions: Tuple[Type[Exception], ...] = (TimeoutError, IOError), ignore_exceptions: Tuple[Type[Exception], ...] = ()): """ 京东API指数退避重试策略 限流异常不重试，避免流量放大 """ def decorator(func: Callable): @wraps(func) def wrapper(*args, **kwargs): last_exception = None for attempt in range(max_attempts): try: return func(*args, **kwargs) except ignore_exceptions: raise except retry_exceptions as e: last_exception = e if attempt < max_attempts - 1: time.sleep(wait_duration * (2 ** attempt)) continue except Exception as e: raise e raise last_exception return wrapper return decorator
import timefrom enum import Enumfrom threading import Lockfrom typing import Callable, TypeVar, Generic, Optional
T = TypeVar('T')class State(Enum):
CLOSED = "closed"
OPEN = "open"
HALF_OPEN = "half_open"class CircuitBreaker:
def __init__(self, failure_rate_threshold: float = 30.0, sliding_window_size: int = 10,
wait_duration_in_open_state: float = 5.0):
self.failure_rate_threshold = failure_rate_threshold
self.sliding_window_size = sliding_window_size
self.wait_duration_in_open_state = wait_duration_in_open_state
self.state = State.CLOSED
self.failure_count = 0
self.success_count = 0
self.total_count = 0
self.last_failure_time = 0
self.lock = Lock()
def call(self, func: Callable[[], T]) -> T:
with self.lock:
if self.state == State.OPEN:
if time.time() - self.last_failure_time >= self.wait_duration_in_open_state:
self.state = State.HALF_OPEN else:
raise Exception("Circuit breaker is OPEN")
try:
result = func()
if self.state == State.HALF_OPEN:
self._reset()
self._record_success()
return result
except Exception as e:
self._record_failure()
self._check_and_transition_state()
raise e def _record_success(self):
self.success_count += 1
self.total_count += 1
if self.total_count > self.sliding_window_size:
self.total_count = self.sliding_window_size
self.success_count = min(self.success_count, self.sliding_window_size)
def _record_failure(self):
self.failure_count += 1
self.total_count += 1
self.last_failure_time = time.time()
if self.total_count > self.sliding_window_size:
self.total_count = self.sliding_window_size
self.failure_count = min(self.failure_count, self.sliding_window_size)
def _check_and_transition_state(self):
if self.total_count >= self.sliding_window_size:
failure_rate = (self.failure_count / self.total_count) * 100
if failure_rate >= self.failure_rate_threshold:
self.state = State.OPEN def _reset(self):
self.state = State.CLOSED
self.failure_count = 0
self.success_count = 0
self.total_count = 0jd_api_circuit_breaker = CircuitBreaker(failure_rate_threshold=30.0, sliding_window_size=10,
wait_duration_in_open_state=5.0)import requestsfrom requests.adapters import HTTPAdapterfrom urllib3.util.retry import Retry# 京东API专用HTTP连接池# 全局最大200连接、单路由最大50连接，适配平台QPS限制session = requests.Session()adapter = HTTPAdapter(
pool_connections=20, # 连接池数量
pool_maxsize=200, # 最大连接数
max_retries=Retry(total=3, backoff_factor=0.3),)session.mount('http://', adapter)session.mount('https://', adapter)# 设置默认超时jd_http_client = session-- 京东订单表状态+时间联合覆盖索引，适配列表查询场景CREATE INDEX idx_jd_order_status_time ON jd_order_detail(order_status, update_time);-- 京东SKU详情主键查询索引，精准匹配查询CREATE INDEX idx_jd_sku_id_status ON jd_sku_detail(sku_id, status);
来自 “ ITPUB博客 ” ，链接：https://blog.itpub.net/70050389/viewspace-3174146/ ，如需转载，请注明出处，否则将追究法律责任。
请登录后发表评论
登录
全部评论
