From 7126fb3ba3aa9222a4dd770601480dff830d3a71 Mon Sep 17 00:00:00 2001 From: marco Date: Sun, 7 Jun 2026 02:33:50 +0800 Subject: [PATCH] =?UTF-8?q?feat(concurrency):=20trace=20=E4=B8=80=E8=87=B4?= =?UTF-8?q?=E6=80=A7=20hash=20=E4=BA=B2=E5=92=8C=20+=20=E9=80=8F=E4=BC=A0?= =?UTF-8?q?=20stream=20=E4=BC=98=E5=8C=96=20(#20)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 配合 pricebot 单机多进程:透传层按 trace_id 把同一次比价/领券的所有帧 亲和到同一 pricebot 实例,使进程内内存维护 state、无需 Redis。本仓改动: - 一致性 hash: 新增 app/core/pricebot_router.py(ketama 环,1000 虚拟节点) + pick_pricebot(trace_id),md5 纯函数、跨 worker 一致;实测负载偏差 4%、 6→7 扩容只重映射 ~13%(取模会 ~85%)。 - 实例列表: app/core/config.py 加 PRICEBOT_INSTANCES(逗号分隔)+ pricebot_instances property;留空 fallback 单实例,向后兼容。 - 透传亲和 + stream: api/v1/compare.py + coupon.py 改为读原始字节、只 loads 一次拿 trace_id 选实例、转发原始 bytes 不重序列化(省 ~半透传 CPU, 单 worker 即可扛高并发,避开多 worker 破坏短信内存表的坑);覆盖 intent/recognize、intent/step、price/step、coupon/step 四端点。 - .env.example 加 PRICEBOT_INSTANCES + PRICEBOT_COMPARE_TIMEOUT_SEC。 Co-Authored-By: Claude Opus 4.8 (1M context) --------- Co-authored-by: pure Reviewed-on: https://gitea.shaguabijia.com/WonderableAI/shaguabijia-app-server/pulls/20 --- .env.example | 10 ++++- app/api/v1/compare.py | 24 ++++++++--- app/api/v1/coupon.py | 23 ++++++++--- app/core/config.py | 15 ++++++- app/core/pricebot_router.py | 82 +++++++++++++++++++++++++++++++++++++ 5 files changed, 139 insertions(+), 15 deletions(-) create mode 100644 app/core/pricebot_router.py diff --git a/.env.example b/.env.example index 82c5bd1..cf775d5 100644 --- a/.env.example +++ b/.env.example @@ -41,12 +41,18 @@ MT_CPS_APP_SECRET= # 默认渠道追踪标识(sid),用于区分不同 app 的 CPS 数据 MT_CPS_DEFAULT_SID=sgbjia -# ===== Pricebot 上游 (领券业务透传目标) ===== -# 客户端调本服务的 /api/v1/coupon/step,我们透传到 pricebot-backend 的 /api/coupon/step。 +# ===== Pricebot 上游 (领券/比价业务透传目标) ===== +# 客户端调本服务的 /api/v1/coupon/step 等,我们透传到 pricebot-backend。 # 本地开发用 localhost:8000。生产部署改成内网地址(如 http://pricebot.internal:8000)。 PRICEBOT_BASE_URL=http://localhost:8000 +# 【多实例】单机多进程部署时,填逗号分隔的实例列表(端口与 pricebot 集群对齐),透传层按 +# trace_id 一致性 hash 选实例 → 同一比价所有帧落同一进程(进程内维护 state,无需 Redis)。 +# 留空 = 单实例(用上面的 PRICEBOT_BASE_URL)。详见 pricebot-backend/docs/并发部署设计.md +# PRICEBOT_INSTANCES=http://127.0.0.1:8001,http://127.0.0.1:8002,http://127.0.0.1:8003,http://127.0.0.1:8004,http://127.0.0.1:8005,http://127.0.0.1:8006 # 领券单帧最多 wait 6s,加网络往返,30s 兜底 PRICEBOT_REQUEST_TIMEOUT_SEC=30 +# 比价(intent/recognize + price/step)透传超时:大上下文 LLM + 逐帧 LLM,给 60s +PRICEBOT_COMPARE_TIMEOUT_SEC=60 # ===== CORS ===== # 逗号分隔,生产留空(只让 app 调,不开放 web)。本地开发可加 http://localhost:5173 之类 diff --git a/app/api/v1/compare.py b/app/api/v1/compare.py index 2a3b5f3..42bcddd 100644 --- a/app/api/v1/compare.py +++ b/app/api/v1/compare.py @@ -18,6 +18,7 @@ pricebot 协议文档: """ from __future__ import annotations +import json import logging from typing import Any @@ -25,6 +26,7 @@ import httpx from fastapi import APIRouter, HTTPException, Request, status from app.core.config import settings +from app.core.pricebot_router import pick_pricebot logger = logging.getLogger("shagua.compare") @@ -38,25 +40,35 @@ async def _passthrough(request: Request, upstream_path: str) -> dict[str, Any]: 打日志。比价单帧是大上下文 / 逐帧 LLM,超时用 PRICEBOT_COMPARE_TIMEOUT_SEC(60s, 比领券的 30s 长)。 """ + # 读原始字节,避免"反序列化→再序列化"的双重 JSON(省 ~一半透传 CPU,让 app-server + # 单 worker 也扛得住高并发)。只 json.loads 一次拿 trace_id 做亲和 + 打日志,转发时 + # 直接发原始 bytes(content=raw),不重新 dumps。 + raw = await request.body() try: - body = await request.json() + meta = json.loads(raw) except Exception as e: raise HTTPException(status_code=400, detail=f"invalid json body: {e}") from e + if not isinstance(meta, dict): + meta = {} - url = f"{settings.PRICEBOT_BASE_URL.rstrip('/')}{upstream_path}" + # 按 trace_id 一致性 hash 选 pricebot 实例(同一比价的所有帧落同一进程,内存维护 state) + base = pick_pricebot(meta.get("trace_id")) + url = f"{base.rstrip('/')}{upstream_path}" timeout = settings.PRICEBOT_COMPARE_TIMEOUT_SEC logger.info( "compare %s device_id=%s trace_id=%s step=%s", upstream_path, - body.get("device_id"), - body.get("trace_id"), - body.get("step"), + meta.get("device_id"), + meta.get("trace_id"), + meta.get("step"), ) try: async with httpx.AsyncClient(timeout=timeout) as client: - resp = await client.post(url, json=body) + resp = await client.post( + url, content=raw, headers={"Content-Type": "application/json"} + ) except httpx.RequestError as e: logger.error("[pricebot] request failed: %s", e) raise HTTPException( diff --git a/app/api/v1/coupon.py b/app/api/v1/coupon.py index 29ba79c..0ed84ef 100644 --- a/app/api/v1/coupon.py +++ b/app/api/v1/coupon.py @@ -10,6 +10,7 @@ pricebot 协议文档: """ from __future__ import annotations +import json import logging from typing import Any @@ -17,6 +18,7 @@ import httpx from fastapi import APIRouter, HTTPException, Request, status from app.core.config import settings +from app.core.pricebot_router import pick_pricebot logger = logging.getLogger("shagua.coupon") @@ -33,24 +35,33 @@ async def coupon_step( - 透传: 不做 schema 校验,pricebot 自己校验 - 失败: 网络不可达 / pricebot 5xx → 502 + 友好 message """ + # 读原始字节,避免"反序列化→再序列化"的双重 JSON(省透传 CPU)。只 loads 一次拿 + # trace_id 做亲和 + 打日志,转发时直接发原始 bytes,不重新 dumps。 + raw = await request.body() try: - body = await request.json() + meta = json.loads(raw) except Exception as e: raise HTTPException(status_code=400, detail=f"invalid json body: {e}") from e + if not isinstance(meta, dict): + meta = {} - url = f"{settings.PRICEBOT_BASE_URL.rstrip('/')}/api/coupon/step" + # 按 trace_id 一致性 hash 选 pricebot 实例(同一领券任务的所有帧落同一进程) + base = pick_pricebot(meta.get("trace_id")) + url = f"{base.rstrip('/')}/api/coupon/step" timeout = settings.PRICEBOT_REQUEST_TIMEOUT_SEC logger.info( "coupon_step device_id=%s trace_id=%s step=%s", - body.get("device_id"), - body.get("trace_id"), - body.get("step"), + meta.get("device_id"), + meta.get("trace_id"), + meta.get("step"), ) try: async with httpx.AsyncClient(timeout=timeout) as client: - resp = await client.post(url, json=body) + resp = await client.post( + url, content=raw, headers={"Content-Type": "application/json"} + ) except httpx.RequestError as e: logger.error("[pricebot] request failed: %s", e) raise HTTPException( diff --git a/app/core/config.py b/app/core/config.py index 83d0ba6..bbcbc31 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -135,15 +135,28 @@ class Settings(BaseSettings): """回调开关打开且验签密钥已配,才接受发奖回调。""" return bool(self.PANGLE_CALLBACK_ENABLED and self.PANGLE_REWARD_SECRET) - # ===== Pricebot 上游 (领券业务透传目标) ===== + # ===== Pricebot 上游 (领券/比价业务透传目标) ===== # pricebot-backend 默认跑在 8000。/api/v1/coupon/step 会透传到这里的 /api/coupon/step PRICEBOT_BASE_URL: str = "http://localhost:8000" + # 多实例(单机多进程)时填逗号分隔的实例列表,例如: + # PRICEBOT_INSTANCES=http://127.0.0.1:8001,http://127.0.0.1:8002,http://127.0.0.1:8003 + # 透传层按 trace_id 一致性 hash 选实例(见 app/core/pricebot_router.py),保证同一次 + # 比价/领券的所有帧落同一 pricebot 进程,从而用进程内内存维护 session/coordinator, + # 无需 Redis。留空 → 退回单实例 [PRICEBOT_BASE_URL],零改动兼容。 + PRICEBOT_INSTANCES: str = "" # 领券一帧最多 wait 6s,加网络往返,timeout 给 30s 比较稳 PRICEBOT_REQUEST_TIMEOUT_SEC: int = 30 # 比价(intent/recognize + price/step)透传超时:意图识别是大上下文 LLM、 # price/step 每帧也是 LLM,可能 >30s;对齐客户端 agent ApiClient 的 60s 读超时。 PRICEBOT_COMPARE_TIMEOUT_SEC: int = 60 + @property + def pricebot_instances(self) -> list[str]: + """pricebot 上游实例列表。空 → 单实例兜底 [PRICEBOT_BASE_URL]。""" + if not self.PRICEBOT_INSTANCES.strip(): + return [self.PRICEBOT_BASE_URL] + return [u.strip() for u in self.PRICEBOT_INSTANCES.split(",") if u.strip()] + # ===== 媒体文件(用户头像上传)===== # 落盘根目录(data/ 已 gitignore,上传不进库);对外经 StaticFiles 挂在 MEDIA_URL_PREFIX。 # 生产可改由 nginx 直接 serve MEDIA_ROOT,绕过应用进程。 diff --git a/app/core/pricebot_router.py b/app/core/pricebot_router.py new file mode 100644 index 0000000..1dc8f52 --- /dev/null +++ b/app/core/pricebot_router.py @@ -0,0 +1,82 @@ +"""pricebot 上游实例选择：按 trace_id 一致性 hash(ketama 风格)选实例。 + +单机多进程下,保证同一 trace_id(一次比价/领券的所有帧)落同一 pricebot 进程, +从而用进程内内存维护 session/coordinator,不需要 Redis。 + +为什么用一致性 hash 而非简单取模(crc32 % N): + 加/减实例时,取模会让几乎所有 trace 重新映射(扩缩容全量中断进行中的比价); + 一致性 hash(虚拟节点)只重映射约 1/N 的 trace,扩缩容对存量冲击最小。 + ⚠️ 但内存态下,被重映射的那 1/N trace 仍会丢失 session(状态没外置), + 所以扩缩容仍建议挑低峰。 + +hash 用 md5(纯函数),app-server 多 worker / 多实例算出的环一致,亲和不会因 +app-server 自身水平扩展而被破坏。 +""" +from __future__ import annotations + +import bisect +import hashlib +from typing import Dict, List, Optional + +from app.core.config import settings + +# 每个真实实例在 hash 环上的虚拟节点数。越多分布越均匀。实例数少时(6~10)需要更多 +# 虚拟节点才均匀:实测 150→负载偏差~16%、1000→~6%。环只在启动/实例变更时构建一次, +# 1000×N 个点的构建与 bisect 查找开销都可忽略。 +_VNODES_PER_NODE = 1000 + + +def _hash(s: str) -> int: + """取 md5 前 8 hex(32-bit)做环坐标。确定性、跨进程一致。""" + return int(hashlib.md5(s.encode("utf-8")).hexdigest()[:8], 16) + + +class _HashRing: + """ketama 风格一致性 hash 环。不可变,实例列表变化时整体重建(见 _get_ring)。""" + + def __init__(self, nodes: List[str]): + self._keys: List[int] = [] # 升序的环坐标 + self._key_to_node: Dict[int, str] = {} + for node in nodes: + for v in range(_VNODES_PER_NODE): + h = _hash(f"{node}#{v}") + # 极小概率撞坐标,撞了跳过该虚拟节点(不影响正确性,仅少一个 vnode) + if h not in self._key_to_node: + self._key_to_node[h] = node + self._keys.append(h) + self._keys.sort() + + def pick(self, key: str) -> str: + h = _hash(key) + idx = bisect.bisect(self._keys, h) + if idx == len(self._keys): + idx = 0 # 环回绕 + return self._key_to_node[self._keys[idx]] + + +# 按实例列表缓存环,列表变了(改 PRICEBOT_INSTANCES + 重启)才重建。 +_ring: Optional[_HashRing] = None +_ring_nodes: tuple = () + + +def _get_ring(nodes: List[str]) -> _HashRing: + global _ring, _ring_nodes + key = tuple(nodes) + if _ring is None or key != _ring_nodes: + _ring = _HashRing(nodes) + _ring_nodes = key + return _ring + + +def pick_pricebot(trace_id: Optional[str]) -> str: + """按 trace_id 选 pricebot 实例 base url。 + + - 实例列表来自 settings.pricebot_instances(空则单实例兜底 [PRICEBOT_BASE_URL]) + - trace_id 缺失/空 或 单实例 → 直接返回第一个,不进环 + """ + nodes = settings.pricebot_instances + if len(nodes) == 1 or not trace_id: + return nodes[0] + # str() 防御:协议保证 trace_id 是 str,但万一传入非 str(如 int)也不至于在 + # _hash 的 .encode() 处炸,确定性地选到实例。 + return _get_ring(nodes).pick(str(trace_id))