what
0f4a4ded53
Invoke 之前完全依赖调用方传入的 ctx 控制超时，池子被占满或 Python 侧 handler 阻塞时，未设置 deadline 的调用会永久阻塞且不报错。新增 WithDefaultTimeout 选项，仅在 ctx 未设置 deadline 时兜底生效，调用方 显式设置的超时优先级更高。 同时补充 example 中的阻塞/超时演示（demoTimeout、demoBlocking）和 pool_test.go 集成测试，覆盖池占满排队、默认超时、显式 deadline 优先级、 流式输出超时后 channel 静默关闭等场景。
653 lines
24 KiB
Markdown
653 lines
24 KiB
Markdown
# gobridge
|
||
|
||
在 Go 与 Python 之间建立双向通信桥接，将 Go 的 `channel` 与 Python 的 `yield` 原生对接，支持普通调用、单向流、双向流。
|
||
|
||
底层通过 **Unix Domain Socket (UDS)** 通信，Go 侧维护 **Worker 进程池**，Python 侧以多线程方式并发处理请求。
|
||
|
||
## 特性
|
||
|
||
- **零配置序列化**：Go struct/slice ↔ Python dict/list 通过 JSON 自动互转
|
||
- **原生流语义**：Go `chan T` 对应 Python `Iterator[T]`，无需额外 API
|
||
- **进程池**：Go 自动启动并管理多个 Python 子进程，崩溃后自动重启
|
||
- **ctx 取消**：Go `context` 取消时自动中断 Python 计算，无需函数内检查
|
||
- **四种调用模式**：普通、流式输出、流式输入、双向流，同一个 `Invoke` 函数自动推断
|
||
|
||
## 安装
|
||
|
||
```bash
|
||
go get git.fsdpf.net/go/gobridge
|
||
```
|
||
|
||
Python 端直接复制 `python/gobridge/` 目录到项目中，无需安装依赖（仅用标准库）。
|
||
|
||
## 快速开始
|
||
|
||
**Python 端（worker.py）：**
|
||
|
||
```python
|
||
from gobridge import gobridge, run
|
||
from typing import Iterator
|
||
|
||
@expose
|
||
def add(a: int, b: int) -> int:
|
||
return a + b
|
||
|
||
run()
|
||
```
|
||
|
||
**Go 端：**
|
||
|
||
```go
|
||
pool, _ := gobridge.NewPool("worker.py")
|
||
defer pool.Close()
|
||
|
||
ctx := context.Background()
|
||
sum, _ := gobridge.Invoke[int](ctx, pool, "add", 3, 4)
|
||
fmt.Println(sum) // 7
|
||
```
|
||
|
||
## 四种调用模式
|
||
|
||
### 1. 普通调用
|
||
|
||
```go
|
||
// Go
|
||
sum, err := gobridge.Invoke[int](ctx, pool, "add", 3, 4)
|
||
|
||
user, err := gobridge.Invoke[User](ctx, pool, "get_user", 42)
|
||
|
||
result, err := gobridge.Invoke[[]User](ctx, pool, "enrich_users", users)
|
||
```
|
||
|
||
```python
|
||
# Python
|
||
@expose
|
||
def add(a: int, b: int) -> int:
|
||
return a + b
|
||
|
||
@expose
|
||
def get_user(uid: int) -> dict:
|
||
return {"id": uid, "name": f"user_{uid}", "score": uid * 1.5}
|
||
|
||
@expose
|
||
def enrich_users(users: list) -> list:
|
||
for u in users:
|
||
u["level"] = "gold" if u["score"] >= 10 else "silver"
|
||
return users
|
||
```
|
||
|
||
### 2. 流式输出（Python yield → Go channel）
|
||
|
||
返回类型为 `chan T` 时自动进入流式输出模式，Python 函数使用 `yield`，Go 侧通过 `range` 消费。
|
||
|
||
```go
|
||
// Go
|
||
ch, err := gobridge.Invoke[chan int](ctx, pool, "range_gen", 1, 6)
|
||
for v := range ch {
|
||
fmt.Println(v) // 1 2 3 4 5
|
||
}
|
||
|
||
userCh, err := gobridge.Invoke[chan User](ctx, pool, "gen_users", 3)
|
||
for u := range userCh {
|
||
fmt.Println(u)
|
||
}
|
||
```
|
||
|
||
```python
|
||
# Python
|
||
@expose
|
||
def range_gen(start: int, stop: int) -> Iterator[int]:
|
||
for i in range(start, stop):
|
||
yield i
|
||
|
||
@expose
|
||
def gen_users(count: int) -> Iterator[dict]:
|
||
for i in range(1, count + 1):
|
||
yield {"id": i, "name": f"user_{i}", "score": float(i * 3)}
|
||
```
|
||
|
||
### 3. 流式输入（Go channel → Python Iterator）
|
||
|
||
参数中含 `chan T` 且返回非 `chan` 时自动进入流式输入模式。
|
||
|
||
```go
|
||
// Go
|
||
inputCh := make(chan int, 10)
|
||
go func() {
|
||
for i := 1; i <= 5; i++ {
|
||
inputCh <- i
|
||
}
|
||
close(inputCh)
|
||
}()
|
||
total, err := gobridge.Invoke[int](ctx, pool, "sum_stream", inputCh)
|
||
fmt.Println(total) // 15
|
||
```
|
||
|
||
```python
|
||
# Python
|
||
@expose
|
||
def sum_stream(numbers: Iterator[int]) -> int:
|
||
return sum(numbers)
|
||
```
|
||
|
||
### 4. 双向流（Go channel 输入 + Go channel 输出）
|
||
|
||
参数含 `chan T` 且返回类型也为 `chan R` 时自动进入双向流模式。
|
||
|
||
```go
|
||
// Go
|
||
inCh := make(chan User, 5)
|
||
go func() {
|
||
for _, u := range users {
|
||
inCh <- u
|
||
}
|
||
close(inCh)
|
||
}()
|
||
outCh, err := gobridge.Invoke[chan User](ctx, pool, "process_users", inCh)
|
||
for u := range outCh {
|
||
fmt.Println(u)
|
||
}
|
||
```
|
||
|
||
```python
|
||
# Python
|
||
@expose
|
||
def process_users(users: Iterator[dict]) -> Iterator[dict]:
|
||
for u in users:
|
||
yield {"id": u["id"], "name": u["name"].upper(), "score": u["score"] * 2}
|
||
```
|
||
|
||
## ctx 取消
|
||
|
||
Go 的 `context` 取消会自动中断 Python 侧的执行，无需在 Python 函数中做任何检查：
|
||
|
||
```go
|
||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
||
defer cancel()
|
||
|
||
// 超时或手动 cancel() 后，Python 计算立即中断，返回 context.DeadlineExceeded
|
||
result, err := gobridge.Invoke[int](ctx, pool, "slow_compute", 1000000)
|
||
```
|
||
|
||
```python
|
||
@expose
|
||
def slow_compute(n: int) -> int:
|
||
total = 0
|
||
for i in range(n):
|
||
total += i # ctx 取消时此处自动抛出 InterruptedError，无需手动检查
|
||
return total
|
||
```
|
||
|
||
**实现机制：**
|
||
1. Go ctx 取消 → 发送 `cancel` 消息给 Python
|
||
2. Python 单 reader 线程收到 `cancel` → 向执行线程注入 `InterruptedError`（`PyThreadState_SetAsyncExc`）
|
||
3. Python 函数在下一条字节码指令处中断
|
||
4. Go 同时关闭连接，解除阻塞的读写操作
|
||
|
||
> 限制：长时间不释放 GIL 的 C 扩展（如大规模 numpy 矩阵运算）无法被中断，需等其释放 GIL 后才触发。
|
||
|
||
### 默认超时（WithDefaultTimeout）
|
||
|
||
`Invoke` 本身不带任何默认超时——如果传入的 `ctx` 没有 deadline（比如直接用 `context.Background()`），且池子被占满（所有 worker 的连接都在处理别的请求），调用会**永久阻塞**，不会自动放弃。
|
||
|
||
`WithDefaultTimeout` 用于兜底这种情况：仅当调用方传入的 `ctx` **未设置** deadline 时才生效，调用方显式设置的 `context.WithTimeout` 优先级更高，不会被覆盖。
|
||
|
||
> 池子被占满时新调用是**排队阻塞等待**，不是立刻失败或被跳过——`example/main.go` 中的 `demoBlocking` 用 `workers=2, maxConns=2`（总容量 4）故意占满连接池，
|
||
> 再发起一个不设超时的调用，实测会阻塞约 1.8s（等到某个占位任务释放连接）才返回，而不是瞬间失败。
|
||
|
||
```go
|
||
pool, _ := gobridge.NewPool("worker.py",
|
||
gobridge.WithDefaultTimeout(500 * time.Millisecond),
|
||
)
|
||
|
||
// 未设置 deadline，池的默认超时自动生效，500ms 后返回 context.DeadlineExceeded
|
||
_, err := gobridge.Invoke[string](context.Background(), pool, "sleep_seconds", 2.0)
|
||
|
||
// 显式传入的 deadline 优先，不受默认超时影响
|
||
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
|
||
defer cancel()
|
||
result, err := gobridge.Invoke[string](ctx, pool, "sleep_seconds", 1.0) // 正常返回
|
||
```
|
||
|
||
**超时是否会通过 `error` 返回，取决于调用模式：**
|
||
|
||
| 调用模式 | 超时表现 |
|
||
|---|---|
|
||
| 普通调用 / 流式输入 | `Invoke` 返回非 nil 的 `error`（`context.DeadlineExceeded` / `context.Canceled`） |
|
||
| 流式输出 / 双向流 | `Invoke` 建立阶段失败会返回 `error`；**建立成功后**若中途超时，只会静默关闭已返回的 channel，不会有第二次 `error` |
|
||
|
||
流式模式下，`for v := range ch` 结束后无法区分"正常读完"还是"被超时打断"，需要调用方自行检查传入的 `ctx.Err()`：
|
||
|
||
```go
|
||
ch, err := gobridge.Invoke[chan int](ctx, pool, "slow_range_gen", 1, 10, 100)
|
||
if err != nil {
|
||
// 建立阶段失败/超时
|
||
}
|
||
for v := range ch {
|
||
fmt.Println(v)
|
||
}
|
||
if ctx.Err() != nil {
|
||
// channel 是因为超时/取消提前关闭的，不是正常 yield 完
|
||
}
|
||
```
|
||
|
||
完整可运行示例见 [example/main.go](example/main.go) 中的 `demoTimeout`（默认超时 / 显式 deadline 优先级 / 流式超时静默关闭 channel）和 `demoBlocking`（池子占满后阻塞排队）。
|
||
|
||
## Session 亲和路由
|
||
|
||
默认情况下，每次 `Invoke` 通过轮询分配 worker 进程。当多次调用需要共享同一 Python 进程的状态时，可以使用 Session 或 StickyCtx 将调用固定到同一进程。
|
||
|
||
### 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。
|
||
|
||
### StickyCtx
|
||
|
||
`StickyCtx` 将亲和键写入 ctx，相同 key 通过哈希稳定路由到同一 worker，无需持有 session 对象：
|
||
|
||
```go
|
||
// 每次调用前附加 key，相同 key 始终走同一 worker
|
||
ctx = gobridge.StickyCtx(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` | `StickyCtx` |
|
||
|---|---|---|
|
||
| 路由方式 | 创建时轮询确定 worker | 按 key 哈希确定 worker |
|
||
| 适用场景 | 显式会话管理 | 按自然键（用户 ID 等）路由 |
|
||
| 携带方式 | 替换 `pool` 参数 | 写入 `ctx` |
|
||
| 超时支持 | `context.WithTimeout(ctx, d)` 正常使用 | 同左 |
|
||
|
||
## []byte ↔ bytes 支持
|
||
|
||
Go 的 `[]byte` 通过 **base64** 编码在 JSON 帧中传输，框架在 Python 侧自动完成编解码，用户侧完全透明。
|
||
|
||
### Python 侧
|
||
|
||
函数参数注解为 `bytes` 时，框架自动将 Go 传入的 base64 字符串解码为 `bytes`；
|
||
返回值为 `bytes` 时，框架自动将其编码为 base64 字符串再发送给 Go。
|
||
|
||
```python
|
||
from gobridge import expose
|
||
from typing import Iterator
|
||
|
||
@expose
|
||
def bytes_reverse(data: bytes) -> bytes:
|
||
return data[::-1]
|
||
|
||
@expose
|
||
def bytes_concat(a: bytes, b: bytes) -> bytes:
|
||
return a + b
|
||
|
||
# 流式输出 bytes：对应 Go Invoke[chan []byte]
|
||
@expose
|
||
def bytes_chunks(data: bytes, size: int) -> Iterator[bytes]:
|
||
for i in range(0, len(data), size):
|
||
yield data[i:i + size]
|
||
```
|
||
|
||
### Go 侧
|
||
|
||
Go 直接使用 `[]byte`，`encoding/json` 自动处理 base64 编解码：
|
||
|
||
```go
|
||
// 普通 []byte 参数与返回值
|
||
rev, err := gobridge.Invoke[[]byte](ctx, pool, "bytes_reverse", []byte("hello"))
|
||
fmt.Printf("%s\n", rev) // olleh
|
||
|
||
cat, err := gobridge.Invoke[[]byte](ctx, pool, "bytes_concat", []byte("foo"), []byte("bar"))
|
||
fmt.Printf("%s\n", cat) // foobar
|
||
|
||
// 流式输出 []byte
|
||
ch, err := gobridge.Invoke[chan []byte](ctx, pool, "bytes_chunks", []byte("abcdefgh"), 3)
|
||
for chunk := range ch {
|
||
fmt.Printf("%s ", chunk) // abc def gh
|
||
}
|
||
```
|
||
|
||
> **效率说明**：base64 编码约使数据体积增大 33%，并有少量 CPU 开销。
|
||
> 对于小块二进制数据（< 1 MB）的 RPC 调用，这通常可以忽略不计；
|
||
> 若需传输大量原始二进制流，建议改用独立的 socket/文件通道。
|
||
|
||
## 注意事项
|
||
|
||
### 在 handler 中使用 `threading.Thread`
|
||
|
||
handler 函数内部可以启动后台线程，但行为取决于是否等待：
|
||
|
||
```python
|
||
# ✅ fire-and-forget：立即返回，连接立即释放，后台线程独立运行
|
||
@expose
|
||
def do_something():
|
||
threading.Thread(target=long_task, daemon=True).start()
|
||
return "ok"
|
||
|
||
# ⚠️ 等待线程：连接被占用直到线程结束，等同于直接在 handler 里执行
|
||
@expose
|
||
def do_something():
|
||
t = threading.Thread(target=long_task)
|
||
t.start()
|
||
t.join() # 连接在此阻塞
|
||
return "ok"
|
||
```
|
||
|
||
**`call_go()` 只能在 handler 的原始线程中调用。** `call_go()` 依赖线程局部变量 `_local.mux` 获取当前连接，后台线程中该变量不存在，调用会抛出 `RuntimeError`：
|
||
|
||
```python
|
||
@expose
|
||
def do_something():
|
||
def bg():
|
||
call_go("Method") # ❌ RuntimeError: call_go() must be called within a gobridge handler
|
||
threading.Thread(target=bg, daemon=True).start()
|
||
return "ok"
|
||
```
|
||
|
||
如果后台线程的结果需要回调 Go，应在 handler 线程中通过 `queue.Queue` 等待后台线程结果后再调用：
|
||
|
||
```python
|
||
@expose
|
||
def do_something():
|
||
result_q = queue.Queue()
|
||
threading.Thread(target=lambda: result_q.put(compute()), daemon=True).start()
|
||
result = result_q.get() # 等待后台线程
|
||
return call_go[str]("Process", result) # ✅ 在 handler 线程中调用
|
||
```
|
||
|
||
## 进程自动重启
|
||
|
||
Python worker 进程崩溃时自动重启，调用方无感知：
|
||
|
||
```
|
||
Python 进程崩溃
|
||
→ monitor goroutine 检测到退出
|
||
→ 排空失效连接
|
||
→ 指数退避重启（100ms → 200ms → ... → 30s）
|
||
→ 新进程就绪后恢复连接池
|
||
```
|
||
|
||
## 配置
|
||
|
||
`NewPool` 使用函数选项模式，第一个参数为脚本路径：
|
||
|
||
```go
|
||
// 最简调用
|
||
pool, err := gobridge.NewPool("worker.py")
|
||
|
||
// 完整配置
|
||
pool, err := gobridge.NewPool("worker.py",
|
||
gobridge.WithWorkers(4), // Python 进程数量，默认 2
|
||
gobridge.WithMaxConns(8), // 每进程最大连接数，默认 4
|
||
gobridge.WithPythonExe("python3"), // 可执行文件，默认 "python3"
|
||
gobridge.WithWorkDir("/path/to/workdir"), // 工作目录，默认继承当前进程
|
||
gobridge.WithEnv("PYTHONUNBUFFERED=1", "K=V"), // 附加环境变量，与当前进程环境合并
|
||
gobridge.WithSocketDir("/var/run/myapp"), // socket 文件目录，默认 /tmp
|
||
gobridge.WithStdout(os.Stdout), // 子进程 stdout，默认 os.Stdout
|
||
gobridge.WithStderr(os.Stderr), // 子进程 stderr，默认 os.Stderr
|
||
gobridge.WithDefaultTimeout(10*time.Second), // Invoke 默认超时，默认不启用
|
||
)
|
||
|
||
// 静默模式：丢弃子进程输出
|
||
pool, err := gobridge.NewPool("worker.py",
|
||
gobridge.WithStdout(io.Discard),
|
||
gobridge.WithStderr(io.Discard),
|
||
)
|
||
```
|
||
|
||
| Option | 说明 | 默认值 |
|
||
|--------|------|--------|
|
||
| `WithWorkers(n)` | Python 进程数量 | `2` |
|
||
| `WithMaxConns(n)` | 每进程最大连接数 | `4` |
|
||
| `WithPythonExe(exe)` | 可执行文件 | `"python3"` |
|
||
| `WithScriptArgs(args...)` | 脚本路径之后的附加参数 | 无 |
|
||
| `WithWorkDir(dir)` | 子进程工作目录 | 继承当前进程 |
|
||
| `WithEnv(kv...)` | 附加环境变量 `"K=V"` | 无 |
|
||
| `WithSocketDir(dir)` | UDS socket 文件目录 | `"/tmp"` |
|
||
| `WithStdout(w)` | 子进程标准输出 | `os.Stdout` |
|
||
| `WithStderr(w)` | 子进程标准错误 | `os.Stderr` |
|
||
| `WithDefaultTimeout(d)` | `Invoke` 默认超时，仅在传入的 `ctx` 未设置 deadline 时生效 | 不启用 |
|
||
|
||
## 使用 uv 管理 Python 环境
|
||
|
||
推荐使用 [uv](https://github.com/astral-sh/uv) 管理 Python 版本和虚拟环境。
|
||
|
||
**方式一：`uv run`（推荐，无需手动激活环境）**
|
||
|
||
```go
|
||
// 等价于执行：uv run worker.py
|
||
pool, err := gobridge.NewPool("run",
|
||
gobridge.WithPythonExe("uv"),
|
||
gobridge.WithScriptArgs("worker.py"),
|
||
gobridge.WithWorkDir("./worker"), // uv 项目目录（含 pyproject.toml）
|
||
)
|
||
```
|
||
|
||
**方式二：直接使用虚拟环境的 python**
|
||
|
||
```bash
|
||
cd worker && uv sync
|
||
```
|
||
|
||
```go
|
||
venvPython, _ := exec.LookPath("worker/.venv/bin/python")
|
||
pool, err := gobridge.NewPool("worker/worker.py",
|
||
gobridge.WithPythonExe(venvPython),
|
||
)
|
||
```
|
||
|
||
**方式三：shell 脚本封装（适合 CI/部署）**
|
||
|
||
```bash
|
||
#!/bin/sh
|
||
# run_worker.sh
|
||
cd "$(dirname "$0")"
|
||
exec uv run python worker.py
|
||
```
|
||
|
||
```go
|
||
pool, err := gobridge.NewPool("./run_worker.sh",
|
||
gobridge.WithPythonExe("/bin/sh"),
|
||
)
|
||
```
|
||
|
||
**典型项目结构：**
|
||
|
||
```
|
||
myproject/
|
||
├── main.go
|
||
├── go.mod
|
||
└── worker/
|
||
├── pyproject.toml
|
||
├── uv.lock
|
||
├── .venv/
|
||
└── worker.py
|
||
```
|
||
|
||
`pyproject.toml`：
|
||
|
||
```toml
|
||
[project]
|
||
name = "worker"
|
||
version = "0.1.0"
|
||
requires-python = ">=3.11"
|
||
dependencies = []
|
||
|
||
[tool.uv.sources]
|
||
# 从 git 仓库安装（推荐）
|
||
gobridge = { git = "https://git.fsdpf.net/go/gobridge.git", subdirectory = "python" }
|
||
|
||
# 本地开发时改用本地路径
|
||
# gobridge = { path = "../../python", editable = true }
|
||
```
|
||
|
||
## 通信协议
|
||
|
||
### 整体架构
|
||
|
||
```
|
||
Go 进程
|
||
┌─────────────────────────────────────────────────────────┐
|
||
│ Invoke[R](ctx, pool, method, args...) │
|
||
│ │ │
|
||
│ ┌────▼─────────────────────────────────────────────┐ │
|
||
│ │ Pool │ │
|
||
│ │ workers[0] workers[1] ... workers[N-1] │ │
|
||
│ │ （轮询选择） │ │
|
||
│ └────┬─────────────────────────────────────────────┘ │
|
||
│ │ 每 worker 维护 M 个可复用连接 │
|
||
└───────┼─────────────────────────────────────────────────┘
|
||
│ Unix Domain Socket（每 worker 独立 .sock 文件）
|
||
┌───────▼──────────────┐ ┌──────────────────────────┐
|
||
│ Python 进程 0 │ │ Python 进程 1 │
|
||
│ worker.py │ │ worker.py │
|
||
└──────────────────────┘ └──────────────────────────┘
|
||
```
|
||
|
||
### Python Worker 内部结构
|
||
|
||
```
|
||
Python 进程
|
||
┌──────────────────────────────────────────────────────────────┐
|
||
│ run() ── UDS server.accept() 循环 │
|
||
│ │ │
|
||
│ 每个连接 → 独立线程 _handle_conn() │
|
||
│ │
|
||
│ ┌─────────────────────────────────────────────────────┐ │
|
||
│ │ _handle_conn（连接线程） │ │
|
||
│ │ │ │
|
||
│ │ ┌──────────────────────────────────────────────┐ │ │
|
||
│ │ │ _ConnMux（单 reader 线程） │ │ │
|
||
│ │ │ │ │ │
|
||
│ │ │ socket ──► 读消息 │ │ │
|
||
│ │ │ │ │ │ │
|
||
│ │ │ ┌────────┼──────────┐ │ │ │
|
||
│ │ │ ▼ ▼ ▼ │ │ │
|
||
│ │ │ call_q chunk_q cancel │ │ │
|
||
│ │ └─────────┬────────┬──────────┼───────────────┘ │ │
|
||
│ │ │ │ │ │ │
|
||
│ │ ▼ │ ▼ │ │
|
||
│ │ 主循环读取 │ PyThreadState_SetAsyncExc │ │
|
||
│ │ _dispatch() │ → 执行线程抛 InterruptedError│ │
|
||
│ │ │ │ │ │
|
||
│ │ ┌──────▼──────┐ │ │ │
|
||
│ │ │ @expose fn│ │ │ │
|
||
│ │ │ │ │ │ │
|
||
│ │ │ 普通函数 │ │ │ │
|
||
│ │ │ return val ──────────────► result/error │ │
|
||
│ │ │ │ │ │ │
|
||
│ │ │ 生成器函数 │ │ │ │
|
||
│ │ │ yield val ───────────────► chunk × N │ │
|
||
│ │ │ │ │ + end │ │
|
||
│ │ │ 流式输入 │ │ │ │
|
||
│ │ │ Iterator ◄─┘ │ ← chunk_q │ │
|
||
│ │ └─────────────┘ │ │
|
||
│ └─────────────────────────────────────────────────────┘ │
|
||
└──────────────────────────────────────────────────────────────┘
|
||
```
|
||
|
||
**消息帧：** `[4字节大端长度][JSON载荷]`
|
||
|
||
**消息类型：**
|
||
|
||
| type | 方向 | 含义 |
|
||
|----------|--------------|------------------------------|
|
||
| `call` | Go → Python | 调用请求 |
|
||
| `result` | Python → Go | 普通返回值 |
|
||
| `chunk` | 双向 | 流数据块 |
|
||
| `end` | 双向 | 流结束标记 |
|
||
| `error` | 双向 | 错误响应 |
|
||
| `cancel` | Go → Python | 取消请求，触发 InterruptedError |
|
||
|
||
## 项目结构
|
||
|
||
```
|
||
gobridge/
|
||
├── protocol.go # Message 结构与类型常量
|
||
├── framing.go # 帧读写（4字节长度前缀 + JSON）
|
||
├── worker.go # Python 子进程管理 + UDS 连接池 + 自动重启
|
||
├── pool.go # 多进程池（轮询负载均衡）+ Option 函数
|
||
├── client.go # Invoke[R] 泛型函数（四种模式自动推断）
|
||
├── example/
|
||
│ ├── main.go # 完整调用示例
|
||
│ └── worker.py # Python 函数示例
|
||
└── python/
|
||
└── gobridge/
|
||
└── __init__.py # Python 库（expose、run、_ConnMux）
|
||
```
|
||
|
||
## 类型对应关系
|
||
|
||
| Go 类型 | Python 类型 |
|
||
|-----------|---------------|
|
||
| `int` | `int` |
|
||
| `float64` | `float` |
|
||
| `string` | `str` |
|
||
| `bool` | `bool` |
|
||
| `[]byte` | `bytes` |
|
||
| `struct` | `dict` |
|
||
| `[]T` | `list` |
|
||
| `chan T` | `Iterator[T]` |
|
||
|
||
> `[]byte` 经 base64 编码在 JSON 帧中传输，框架自动完成编解码，用户侧透明。`chan T` 中的 `T` 同样支持 `[]byte`，即 `chan []byte` ↔ `Iterator[bytes]`。
|
||
|
||
## 参考
|
||
|
||
本项目的进程池、UDS 通信、帧协议设计参考自 [pyproc](https://github.com/YuminosukeSato/pyproc)，在此基础上增加了 Go channel 与 Python yield 的流式对接及 ctx 取消支持。
|