Actor

每个 Actor 使用一个泛型 MPSC mailbox。队列不使用互斥锁或自旋锁。默认取得消费权的发送者在自己的 goroutine 中处理消息，直到 mailbox 清空，库不会创建后台 goroutine；也可以在构造时注入 dispatcher，将整个消费任务交给其他 goroutine 或已有线程池。

默认同步执行（inline）：

var total int
counter := actor.New(func(value int) {
    total += value
}, actor.WithFixedCapacity[int](64), actor.WithBatchSize[int](64))

counter.Send(1) // 空闲时，在当前 goroutine 中处理。

New(handler, opts...)：创建 Actor。处理函数或 option 为空、固定容量或批量大小不为正数、dispatcher 接口或 DispatcherFunc 为空时 panic。
Send(message)：无返回值。发布消息后，取得消费权的发送者提交一次完整消费任务；已有任务排队或执行时仅入队。默认同步执行，返回不保证该消息已经处理，也不支持撤消。
WithFixedCapacity[T](n)：预分配并保留 n 个节点，默认 64。固定区全部消费后循环复用；不足时分配动态节点，因此这不是队列总容量限制。
WithBatchSize[T](n)：默认每消费 64 条消息调用 runtime.Gosched()。长链反转同样定期让出 CPU；单条处理函数不会被这个限制中断。
WithPanicHandler[T](func(message T, value any, stack []byte))：可选 panic 回调。每条处理函数及回调自身的 panic 独立恢复，之后继续处理后续消息；没有回调时也继续处理。
WithDispatcher[T](Dispatcher)：在构造时配置消费任务的执行方式，不提供运行中切换。调度策略不改变批量让出 CPU 的频率。

使用 DispatcherFunc 将消费任务交给其他 goroutine：

processed := make(chan int, 1)
receiver := actor.New(func(value int) {
    processed <- value
}, actor.WithDispatcher[int](actor.DispatcherFunc(func(task func()) {
    go task()
})))

receiver.Send(42)
value := <-processed // 用业务自己的同步方式等待结果。

注入已有线程池，可以通过适配器接入其提交方法：

type taskPool interface {
    Submit(task func())
}

type poolDispatcher struct {
    pool taskPool
}

func (d poolDispatcher) Schedule(task func()) {
    d.pool.Submit(task)
}

func newCounter(pool taskPool) *actor.Actor[int] {
    total := 0
    return actor.New(func(value int) {
        total += value
    }, actor.WithDispatcher[int](poolDispatcher{pool: pool}))
}

Dispatcher 只有 Schedule(task func()) 方法。它可以同步或异步执行，但必须接受任务并最终恰好执行一次，不得拒绝、静默丢弃或 panic；共享实例还必须支持并发调用。上述线程池的 Submit 同样需要满足这个约定。mailbox 在调用 Schedule 前已取得消费权，任务尚未执行时的新消息不会重复提交任务；库不为违反调度契约的实现提供回退。自定义 dispatcher 接口中包含 nil 指针时，调用方负责保证其 Schedule 仍有效。

FIFO 按消息成功发布的原子操作排序，固定区与动态区共享这个顺序。生产者先预留固定位置并不会占据 FIFO 顺序；它在发布前暂停时，其他生产者仍能发布并处理消息。消费者通过原子交换取出一批节点，反转链接后按 FIFO 处理，不复制消息来扩容。

处理函数不会并发或重入。处理函数向自身 Actor 发送的消息只会入队，必须等当前处理结束才能执行；不要同步等待它的执行结果。向使用默认 dispatcher 的其他空闲 Actor 发送消息会同步进入对方的处理函数，因此也应避免循环等待和过深的同步调用链。持续繁忙的 mailbox 可能让同步 Send 一直不返回；Gosched 只让出 CPU，不会切换调度策略，也不会打断单条处理函数。

Send 复制的是 Go 值，不深复制其中的指针、切片或 map；调用方需自行同步对共享数据的访问。固定槽位消费后清零，动态节点清除引用后交由 GC 回收。Actor 的可变状态应只在处理函数中访问，或自行同步。Actor 必须通过 New 创建，开始使用后不可复制。

内部职责分别放在 actor.go（门面与组装）、queue.go（存储与 FIFO）、mailbox.go（消费权、消费循环与调度）、invoker.go（处理函数与 panic 隔离）、dispatcher.go（执行策略契约）和 options.go（构造配置）。不包含生命周期、取消或自动重试。

4.5 KiB Raw Permalink Blame History Unescape Escape

Actor

4.5 KiB

Raw Permalink Blame History