写一个所谓的通用状态机

2026/7/24 17:00:00Henry2 阅读0 点赞0 评论

技术背景

当状态转移关系比较繁杂的时候,通常需要引入状态机,让业务层不用每次都去维护繁杂的状态,工程里最常见的写法是:

强类型状态和事件 + 显式转换表 + Guard 校验 + 内存状态变更 + Application Service 负责事务和持久化。

GO
permission := repo.GetForUpdate(...)

result, err := machine.Fire(permission, input)
if err != nil {
	return err
}
repo.Save(permission)
outbox.Append(result.Events...)

最近哥们儿在做即时通讯的权限部分,需要实现类似抖音、小红书的陌生人消息功能,很显然需要引入一手状态机,其实 Outbox 部分就已经需要状态机了,不过也是瞎搞的,招笑。

这次 AI 哥 A 明白了之后,把之前的 💩 修改一下。

基本结构

对于发信权限表,状态其实并不复杂,包含几种关系:没有打过招呼的透明关系、好友关系、打过招呼的陌生人关系、互相打过招呼的开放关系以及经典的红色感叹号❗️拉黑关系!

要写一个状态机,首先要理清楚状态有哪些,事件有哪些,状态的流转链路是怎样的。

我们可以简化假设存在以下关系:

状态:

TEXT
Pending:已经发送招呼,等待对方回复
Open:会话开放,双方可以正常聊天
Banned:会话被禁止

事件:

TEXT
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).

关系变更状态由好友服务实时下发,保证即时通讯服务不会出现异常的发信权限。

(为确保关系状态的实时性,目前必须这样设计,并没有想到更好的办法!)

GO
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"
	}
}

定义事件

GO
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

用文字描述就是,某一个用户和对方打了招呼的状态 + 对方回复事件 = 开放聊天状态

即:状态 + 事件 = 新的状态。

定义状态机输入

上面只是简单描述了有什么状态,有什么事件,但是不管是状态还是事件,其实都是和实体要产生联系的,所以我们在使用状态机时,需要告诉状态机我们操作的是哪个实体的状态,例如:

  • 谁回复;
  • 谁拉黑;
  • 什么时候。
GO
type Input struct {
	Event      Event
	OperatorID string
	At         time.Time
}

定义转换规则

如果没有状态机约束,那业务层可以随便流转状态,比如我突发性的从 Open 状态流转到 Pending 状态,这显然是不符合预期的异常流转,导致流转链路混乱,同时难以维护。

一个流转规则通常包含:

  • 当前状态
  • 事件
  • 目标状态
  • Guard 条件
  • Action 附加操作
GO
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 事件,在此处定义:

GO
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 组成。

GO
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
}

注册流转规则

GO
func (m *StateMachine) registerTransitions() {
	m.registerOpenTransitions()
	m.registerBanTransitions()
	m.registerUnbanTransitions()
}

这一部分通常由业务层实现,然后注册到状态机中。

GO
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 表示流转路径,从而完成状态转移。

GO
type Result struct {
	From   Status
	To     Status
	Event  Event
	Events []DomainEvent
}
GO
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
}

总体流程如下:

  1. 找到流转规则
  2. Guard 校验
  3. 执行 Action
  4. 更新状态
  5. 返回状态变化和(领域事件)

业务调用

业务方只需要在事务中调用状态机 Fire 方法,传入当前状态和事件,状态机会构造 Key 获取到正确的流转路径,完成状态转移,然后业务方根据状态机结果调用 Repo 接口完成数据落库或其他事件广播。

GO
// 初始状态
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 即可:

GO
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,
		)
	}
}

结束!

评论区