docs: 添加 Session 亲和路由章节(NewSession / WithAffinity)
This commit is contained in:
83
README.md
83
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`
|
||||
|
||||
Reference in New Issue
Block a user