跳转到内容

Command Dispatch

Dispatch 系统将命令从进程路由到 handler。进程使用关联标签 yield 命令,handler 执行异步工作,结果通过事件队列返回。

sequenceDiagram
participant P as Process
participant W as Worker
participant R as Registry
participant H as Handler
P->>W: yield(command, tag)
W->>R: getHandler(cmdID)
R-->>W: handler
W->>H: Handle(cmd, tag, receiver)
H-->>H: 异步工作
H->>W: CompleteYield(tag, result)
W->>P: 入队事件,唤醒
P->>P: 使用结果恢复

Registry 使用混合结构存储 handler:

type Registry struct {
handlers [256]Handler // 系统命令:O(1) 索引
extended map[CommandID]Handler // 扩展命令:map 查找
frozen atomic.Bool // 启动后无锁
}

系统命令(0-255)使用数组索引。扩展命令使用 map 查找。Freeze() 后所有查找都是无锁的。

范围模块示例
1-9processSend, Spawn, Terminate, Cancel, Monitor, Unmonitor, Link, Unlink, Exec
10-29clockSleep, Ticker, Timer
30-39socketDial, Listen, Accept, Close
50-59streamRead, Write, Close, Seek
60-69httpRequest, RequestBatch
70-79tty终端 I/O
80-89websocketConnect, Send, Receive
90-99eventSubscribe, Send
100-119sqlQuery, Execute, Prepare, Stmt, Tx ops
120-129storeGet, Set, Delete, Has
130-139securityValidateToken, CreateToken
140-149functionCall, AsyncStart, AsyncCancel
150-159execProcessWait
160-169cloudstorageUpload, Download, List, Presigned URLs
170-179evalCompile, Run
180-189workflowSideEffect, Call, Version, UpsertAttrs
190-199contractOpen, Call, AsyncCall, AsyncCancel
256+custom用户定义的服务

注册在启动期间通过 MustRegisterCommands() 完成。冲突会在启动时 panic。

Command 是具有唯一 CommandID 的数据结构:

const MyCommand dispatcher.CommandID = 200
type MyCmd struct {
Input string
Option int
}
var myCmdPool = sync.Pool{New: func() any { return &MyCmd{} }}
func (c *MyCmd) CmdID() dispatcher.CommandID { return MyCommand }
func (c *MyCmd) Release() {
c.Input = ""
c.Option = 0
myCmdPool.Put(c)
}

池复用消除了热路径中的分配。在包初始化时注册:

func init() {
dispatcher.MustRegisterCommands("myservice", MyCommand)
}

Dispatcher 将相关的 handler 分组。它实现 RegisterAll 来注册 handler,以及用于设置/清理的生命周期方法:

type Handler interface {
Handle(ctx context.Context, cmd Command, tag uint64, receiver ResultReceiver) error
}
type ResultReceiver interface {
CompleteYield(tag uint64, data any, err error)
}
type Dispatcher struct {
// 服务状态
}
func (d *Dispatcher) RegisterAll(register func(id dispatcher.CommandID, h dispatcher.Handler)) {
register(myapi.MyCommand, dispatcher.HandlerFunc(d.handleMyCommand))
}
func (d *Dispatcher) handleMyCommand(ctx context.Context, cmd Command, tag uint64, receiver ResultReceiver) error {
c := cmd.(*myapi.MyCmd)
go func() {
result := doWork(c)
if ctx.Err() == nil {
receiver.CompleteYield(tag, result, nil)
}
}()
return nil
}

作为启动组件注册:

func MyDispatcher() boot.Component {
return boot.New(boot.P{
Name: "dispatcher.myservice",
DependsOn: []boot.Name{DispatcherName},
Load: func(ctx context.Context) (context.Context, error) {
reg := dispatcher.GetRegistrar(ctx)
svc := myservice.NewDispatcher()
svc.RegisterAll(reg.Register)
return ctx, nil
},
})
}

当进程需要异步工作时,它使用关联标签 yield 一个命令:

type Yield struct {
Cmd Command
Tag uint64 // 用于关联的进程本地计数器
}

Worker 在每一步之后从 StepOutput 提取 yield 并分发给 handler。每个标签唯一标识请求,以便结果可以匹配回来。