写一个所谓的通用状态机

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

技术背景

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

强类型状态和事件 + 显式转换表 + 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 明白了之后，把之前的 💩 修改一下。

基本结构

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

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

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

状态：

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

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

（为确保关系状态的实时性，目前必须这样设计，并没有想到更好的办法！）

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

结束！

推荐订阅源

Blog Feed