写一个所谓的通用状态机
技术背景
当状态转移关系比较繁杂的时候,通常需要引入状态机,让业务层不用每次都去维护繁杂的状态,工程里最常见的写法是:
强类型状态和事件 + 显式转换表 + Guard 校验 + 内存状态变更 + Application Service 负责事务和持久化。
permission := repo.GetForUpdate(...)
result, err := machine.Fire(permission, input)
if err != nil {
return err
}
repo.Save(permission)
outbox.Append(result.Events...)
最近哥们儿在做即时通讯的权限部分,需要实现类似抖音、小红书的陌生人消息功能,很显然需要引入一手状态机,其实 Outbox 部分就已经需要状态机了,不过也是瞎搞的,招笑。
这次 AI 哥 A 明白了之后,把之前的 💩 修改一下。
基本结构
对于发信权限表,状态其实并不复杂,包含几种关系:没有打过招呼的透明关系、好友关系、打过招呼的陌生人关系、互相打过招呼的开放关系以及经典的红色感叹号❗️拉黑关系!
要写一个状态机,首先要理清楚状态有哪些,事件有哪些,状态的流转链路是怎样的。
我们可以简化假设存在以下关系:
状态:
Pending:已经发送招呼,等待对方回复
Open:会话开放,双方可以正常聊天
Banned:会话被禁止
事件:
Reply:对方第一次回复
Ban:拉黑
Unban:解除拉黑
定义状态
由于好友关系是通过外部微服务 RPC 获取的,所以我没有打算将好友态写入状态机,而是通过含淘汰算法的本地缓存来维护好友关系:
- Eviction: SampledLFU - on par with exact LRU and better performance on Search and Database traces.
- Admission: TinyLFU - extra performance with little memory overhead (12 bits per counter).
关系变更状态由好友服务实时下发,保证即时通讯服务不会出现异常的发信权限。
(为确保关系状态的实时性,目前必须这样设计,并没有想到更好的办法!)
package permission
type Status int8
const (
StatusPending Status = iota + 1
StatusOpen
StatusBanned
)
func (s Status) String() string {
switch s {
case StatusPending:
return "pending"
case StatusOpen:
return "open"
case StatusBanned:
return "banned"
default:
return "unknown"
}
}
定义事件
type Event int8
const (
EventReply Event = iota + 1
EventBan
EventUnban
)
func (e Event) String() string {
switch e {
case EventReply:
return "reply"
case EventBan:
return "ban"
case EventUnban:
return "unban"
default:
return "unknown"
}
}
事件决定了状态的流转方向,做过工作流引擎的应该很熟悉这个逻辑。
比如在这个场景中,我们有这个流转关系:StatusPending + EventReply = StatusOpen
用文字描述就是,某一个用户和对方打了招呼的状态 + 对方回复事件 = 开放聊天状态。
即:状态 + 事件 = 新的状态。
定义状态机输入
上面只是简单描述了有什么状态,有什么事件,但是不管是状态还是事件,其实都是和实体要产生联系的,所以我们在使用状态机时,需要告诉状态机我们操作的是哪个实体的状态,例如:
- 谁回复;
- 谁拉黑;
- 什么时候。
type Input struct {
Event Event
OperatorID string
At time.Time
}
定义转换规则
如果没有状态机约束,那业务层可以随便流转状态,比如我突发性的从 Open 状态流转到 Pending 状态,这显然是不符合预期的异常流转,导致流转链路混乱,同时难以维护。
一个流转规则通常包含:
- 当前状态
- 事件
- 目标状态
- Guard 条件
- Action 附加操作
type transitionKey struct {
From Status
Event Event
}
type Transition struct {
To Status
// Guard 只负责校验,不修改状态。
Guard func(
permission *ConversationPermission,
input Input,
) error
// Action 修改状态之外的领域字段,并返回领域事件。
Action func(
permission *ConversationPermission,
input Input,
) []DomainEvent
}
定义领域事件
可选,状态流转后,我们可能需要触发一些 Outbox 事件,在此处定义:
type DomainEvent interface {
EventName() string
}
type ConversationOpened struct {
ConversationID string
OpenedAt time.Time
}
func (ConversationOpened) EventName() string {
return "conversation.permission.opened"
}
type ConversationBanned struct {
ConversationID string
OperatorID string
BannedAt time.Time
}
func (ConversationBanned) EventName() string {
return "conversation.permission.banned"
}
实现状态机
一个相对完整且通用的状态机一般由 注册 Transition + Fire 组成。
var ErrInvalidTransition = errors.New("invalid state transition")
type StateMachine struct {
transitions map[transitionKey]Transition
}
func NewStateMachine() *StateMachine {
m := &StateMachine{
transitions: make(map[transitionKey]Transition),
}
m.registerTransitions()
return m
}
func (m *StateMachine) register(
from Status,
event Event,
transition Transition,
) {
key := transitionKey{
From: from,
Event: event,
}
m.transitions[key] = transition
}
注册流转规则
func (m *StateMachine) registerTransitions() {
m.registerOpenTransitions()
m.registerBanTransitions()
m.registerUnbanTransitions()
}
这一部分通常由业务层实现,然后注册到状态机中。
func (m *StateMachine) registerOpenTransitions() {
openTransitions := Transition{
To: StatusOpen,
Guard: func(
permission *ConversationPermission,
input Input,
) error {
if input.OperatorID == "" {
return errors.New("operator_id is required")
}
if input.OperatorID == permission.InitiatorID {
return errors.New(
"initiator cannot open conversation by replying",
)
}
return nil
},
Action: func(
permission *ConversationPermission,
input Input,
) []DomainEvent {
at := input.At
permission.OpenedAt = &at
permission.BannedAt = nil
permission.BannedBy = ""
return []DomainEvent{
ConversationOpened{
ConversationID: permission.ConversationID,
OpenedAt: at,
},
}
},
}
// 定义流转路径
// Pending -> Event(Reply) -> Transition(StatusOpen)
m.register(StatusPending, EventReply, openTransitions)
}
func (m *StateMachine) registerBanTransitions() {
banTransition := Transition{
To: StatusBanned,
// 过滤非法数据
Guard: func(
permission *ConversationPermission,
input Input,
) error {
if input.OperatorID == "" {
return errors.New("operator_id is required")
}
return nil
},
// 由业务定义,这里只是做演示
Action: func(
permission *ConversationPermission,
input Input,
) []DomainEvent {
at := input.At
permission.BannedAt = &at
permission.BannedBy = input.OperatorID
return []DomainEvent{
ConversationBanned{
ConversationID: permission.ConversationID,
OperatorID: input.OperatorID,
BannedAt: at,
},
}
},
}
// 定义流转路径
// Pending/Open -> Event(Ban) -> Transition(StatusBanned)
m.register(StatusPending, EventBan, banTransition)
m.register(StatusOpen, EventBan, banTransition)
}
type ConversationUnbanned struct {
ConversationID string
OperatorID string
UnbannedAt time.Time
}
func (ConversationUnbanned) EventName() string {
return "conversation.permission.unbanned"
}
func (m *StateMachine) registerUnbanTransitions() {
unbanTransition := Transition{
To: StatusOpen,
Guard: func(
permission *ConversationPermission,
input Input,
) error {
if input.OperatorID == "" {
return errors.New("operator_id is required")
}
if permission.BannedBy != input.OperatorID {
return errors.New(
"only the user who banned the conversation can unban it",
)
}
return nil
},
Action: func(
permission *ConversationPermission,
input Input,
) []DomainEvent {
permission.BannedAt = nil
permission.BannedBy = ""
return []DomainEvent{
ConversationUnbanned{
ConversationID: permission.ConversationID,
OperatorID: input.OperatorID,
UnbannedAt: input.At,
},
}
},
}
// 定义流转路径
// Banned -> Event(Unban) -> Transition(StatusOpen)
m.register(StatusBanned, EventUnban, unbanTransition)
}
实现 Fire
之前通过 register 方法将 Status + Event 组成的 Key 注册 Transitions,现在就能通过 Key 找到对应的 Transitions。Key 表示来源,Transitions 表示流转路径,从而完成状态转移。
type Result struct {
From Status
To Status
Event Event
Events []DomainEvent
}
func (m *StateMachine) Fire(
permission *ConversationPermission,
input Input,
) (Result, error) {
if permission == nil {
return Result{}, errors.New(
"conversation permission is nil",
)
}
if input.At.IsZero() {
return Result{}, errors.New(
"transition time is required",
)
}
// 通过来源构造 Transitions Key
key := transitionKey{
From: permission.Status,
Event: input.Event,
}
// 通过 key 找到之前注册的流转路径
transition, ok := m.transitions[key]
if !ok {
return Result{}, fmt.Errorf(
"%w: status=%s event=%s",
ErrInvalidTransition,
permission.Status,
input.Event,
)
}
// 流转前验证
if transition.Guard != nil {
if err := transition.Guard(permission, input); err != nil {
return Result{}, fmt.Errorf(
"transition guard rejected: %w",
err,
)
}
}
from := permission.Status
// 开始流转,并返回领域事件
// 领域事件可选,主要用于广播通知
var events []DomainEvent
if transition.Action != nil {
events = transition.Action(permission, input)
}
permission.Status = transition.To
// 返回流转结果
return Result{
From: from,
To: transition.To,
Event: input.Event,
Events: events,
}, nil
}
总体流程如下:
- 找到流转规则
- Guard 校验
- 执行 Action
- 更新状态
- 返回状态变化和(领域事件)
业务调用
业务方只需要在事务中调用状态机 Fire 方法,传入当前状态和事件,状态机会构造 Key 获取到正确的流转路径,完成状态转移,然后业务方根据状态机结果调用 Repo 接口完成数据落库或其他事件广播。
// 初始状态
permission, err :=
s.repo.FindByConversationIDForUpdate(
ctx,
tx,
conversationID,
)
if err != nil {
return fmt.Errorf(
"find conversation permission: %w",
err,
)
}
if permission == nil {
return errors.New(
"conversation permission not found",
)
}
// 调用状态机
// 状态机会根据状态和事件构造 Key 的到 Transitions 修改初始状态并返回流转结果
result, err := s.machine.Fire(
permission,
Input{
Event: EventReply,
OperatorID: senderID,
At: time.Now(),
},
)
if err != nil {
return err
}
// 新状态落库
if err := s.repo.Save(
ctx,
tx,
permission,
); err != nil {
return fmt.Errorf(
"save conversation permission: %w",
err,
)
}
// 可选领域事件广播/Outbox
for _, event := range result.Events {
if err := s.outboxRepo.Append(
ctx,
tx,
event,
); err != nil {
return fmt.Errorf(
"append permission event: %w",
err,
)
}
}
在这里不得不吐槽一下 macOS 输入法,我想输入“初始状态”,非要给我联想出一个“出事状态”,我看怕是 🐒 猴王出事还差不多!
状态机选型
| 状态机规模 | 推荐写法 |
|---|---|
| 2~5 个状态、规则简单 | switch |
| 状态和事件较多 | 转换表map[key]Transition |
| 涉及多服务长事务 | Saga / Workflow |
| 仅单表简单状态更新 | 条件 UPDATE |
对于简单一些的状态机,那么定义一个流转路径 map 或者直接使用 switch 即可:
func (m StateMachine) Fire(
permission *ConversationPermission,
input Input,
) (Result, error) {
if permission == nil {
return Result{}, errors.New(
"conversation permission is nil",
)
}
switch input.Event {
case EventReply:
return m.reply(permission, input)
case EventBan:
return m.ban(permission, input)
case EventUnban:
return m.unban(permission, input)
default:
return Result{}, fmt.Errorf(
"unknown permission event: %d",
input.Event,
)
}
}
结束!
评论区