marco
7126fb3ba3
配合 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) <noreply@anthropic.com> --------- Co-authored-by: pure <pure@192.168.0.104> Reviewed-on: #20
83 lines
3.2 KiB
Python
83 lines
3.2 KiB
Python
"""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))
|