diff --git a/README.md b/README.md index 8403588..b32bb04 100644 --- a/README.md +++ b/README.md @@ -186,6 +186,89 @@ def slow_compute(n: int) -> int: > 限制:长时间不释放 GIL 的 C 扩展(如大规模 numpy 矩阵运算)无法被中断,需等其释放 GIL 后才触发。 +## Session 亲和路由 + +默认情况下,每次 `Invoke` 通过轮询分配 worker 进程。当多次调用需要共享同一 Python 进程的状态时,可以使用 Session 或 WithAffinity 将调用固定到同一进程。 + +### NewSession + +`NewSession` 返回一个固定到某个 worker 的 `Pool` 视图,通过它发起的所有 `Invoke` 始终路由到同一 Python 进程: + +```go +pool, _ := gobridge.NewPool("worker.py", gobridge.WithWorkers(2)) + +sessA := gobridge.NewSession(pool) // 固定到 worker 1 +sessB := gobridge.NewSession(pool) // 固定到 worker 0(轮询下一个) +sessC := gobridge.NewSession(pool) // 固定到 worker 1(与 sessA 同进程) + +gobridge.Invoke(ctx, sessA, "init", "A", 100) +gobridge.Invoke(ctx, sessA, "step", "A", 10) // 始终走 worker 1 +gobridge.Invoke(ctx, sessC, "step", "C", 99) // 也走 worker 1,但 session_id 不同 +``` + +**Python 侧**用模块级 dict 以 `session_id` 为 key 隔离状态: + +```python +_sessions = {} + +@expose +def init(session_id: str, value: int): + _sessions[session_id] = {"value": value} + +@expose +def step(session_id: str, delta: int) -> int: + _sessions[session_id]["value"] += delta + return _sessions[session_id]["value"] +``` + +> `NewSession` 不拥有底层 pool 的生命周期,调用 `session.Close()` 是空操作,只需关闭原始 pool。 + +### WithAffinity + +`WithAffinity` 将亲和键写入 ctx,相同 key 通过哈希稳定路由到同一 worker,无需持有 session 对象: + +```go +// 每次调用前附加 key,相同 key 始终走同一 worker +ctx = gobridge.WithAffinity(ctx, "user-42") +gobridge.Invoke(ctx, pool, "method_a", ...) +gobridge.Invoke(ctx, pool, "method_b", ...) +``` + +适合按用户 ID、租户 ID 等自然键做路由,不需要显式创建 session 对象。 + +### 进程级全局变量 + +同一 worker 进程内的所有 session 共享进程级变量(如模块级 `dict`、计数器等),不同 worker 进程之间完全隔离: + +```python +_counter = 0 # 进程级全局变量 + +@expose +def increment(delta: int) -> int: + global _counter + _counter += delta + return _counter +``` + +```go +sessA := gobridge.NewSession(pool) // worker 1 +sessC := gobridge.NewSession(pool) // worker 1(与 A 同进程) +sessB := gobridge.NewSession(pool) // worker 0(独立进程) + +gobridge.Invoke(ctx, sessA, "increment", 10) // worker 1: counter = 10 +gobridge.Invoke(ctx, sessC, "increment", 5) // worker 1: counter = 15(共享) +gobridge.Invoke(ctx, sessB, "increment", 99) // worker 0: counter = 99(独立) +``` + +### 两种方式对比 + +| | `NewSession` | `WithAffinity` | +|---|---|---| +| 路由方式 | 创建时轮询确定 worker | 按 key 哈希确定 worker | +| 适用场景 | 显式会话管理 | 按自然键(用户 ID 等)路由 | +| 携带方式 | 替换 `pool` 参数 | 写入 `ctx` | +| 超时支持 | `context.WithTimeout(ctx, d)` 正常使用 | 同左 | + ## 注意事项 ### 在 handler 中使用 `threading.Thread`