Golang 狀態機設計模式,你知道多少?
導言
在我們開發的許多項目中,都需要依賴某種運行狀態從而實現連續操作。
這方面的例子包括:
- 解析配置語言、編程語言等
- 在系統、路由器、集群上執行操作...
- ETL(Extract Transform Load,提取轉換加載)
很久以前,Rob Pike 有一個關于 Go 中詞法掃描[2]的演講,內容很講座,我看了好幾遍才真正理解。但演講中介紹的最基本知識之一就是某個版本的 Go 狀態機。
該狀態機利用了 Go 的能力,即從函數中創建類型并將函數賦值給變量。
他在演講中介紹的狀態機功能強大,打破了讓函數執行 if/else 并調用下一個所需函數的邏輯。取而代之的是,每個狀態都會返回下一個需要調用的函數。
這樣就能將調用鏈分成更容易測試的部分。
調用鏈
下面是一個用簡單的調用鏈來完成任務的例子:
func Caller(args Args) {
callA(args)
callB(args)
}
或
func Caller(args Args) {
callA(args)
}
func callA(args Args) {
callB(args)
}
func callB(args Args) {
return
}
兩種方法都表示調用鏈,其中 Caller() 調用 callA(),并最終調用 callB(),從中可以看到這一系列調用是如何執行的。
當然,這種設計沒有任何問題,但當調用者遠程調用其他系統時,必須對這些遠程調用進行模擬/打樁,以提供密封測試。
你可能還想實現條件調用鏈,即根據某些參數或狀態,在特定條件下通過 if/else 調用不同函數。
這就意味著,要對 Caller() 進行密封測試,可能需要處理整個調用鏈中的樁函數。如果有 50 個調用層級,則可能需要對被測函數下面每個層級的所有函數進行模擬/打樁。
這正是 Pike 的狀態機設計大顯身手的地方。
狀態機模式
首先定義狀態:
type State[T any] func(ctx context.Context, args T) (T, State[T], error)
狀態表示為函數/方法,接收一組參數(任意類型 T),并返回下一個狀態及其參數或錯誤信息。
如果返回的狀態為 nil,那么狀態機將停止運行。如果設置了 error,狀態機也將停止運行。因為返回的是下一個要運行的狀態,所以根據不同的條件,會有不同的下一個狀態。
這個版本與 Pike 的狀態機的不同之處在于這里包含了泛型并返回 T。這樣我們就可以創建純粹的函數式狀態機(如果需要的話),可以返回某個類型,并將其傳遞給下一個狀態。Pike 最初實現狀態機設計時還沒有使用泛型。
為了實現這一目標,需要一個狀態驅動程序:
func Run[T any](ctx context.Context, args T, start State[T] "T any") (T, error) {
var err error
current := start
for {
if ctx.Err() != nil {
return args, ctx.Err()
}
args, current, err = current(ctx, args)
if err != nil {
return args, err
}
if current == nil {
return args, nil
}
}
}
寥寥幾行代碼,我們就有了一個功能強大的狀態驅動程序。
下面來看一個例子,在這個例子中,我們為集群中的服務關閉操作編寫了狀態機:
package remove
...
// storageClient provides the methods on a storage service
// that must be provided to use Remove().
type storageClient interface {
RemoveBackups(ctx context.Context, service string, mustKeep int) error
RemoveContainer(ctx context.Context, service string) error
}
// serviceClient provides methods to do operations for services
// within a cluster.
type servicesClient interface {
Drain(ctx context.Context, service string) error
Remove(ctx context.Context, service string) error
List(ctx context.Context) ([]string, error)
HasStorage(ctx context.Context, service string) (bool, error)
}
這里定義了幾個需要客戶實現的私有接口,以便從集群中移除服務。
我們定義了私有接口,以防止他人使用我們的定義,但會通過公有變量公開這些接口。這樣,我們就能與客戶保持松耦合,保證只使用我們需要的方法。
// Args are arguments to Service().
type Args struct {
// Name is the name of the service.
Name string
// Storage is a client that can remove storage backups and storage
// containers for a service.
Storage storageClient
// Services is a client that allows the draining and removal of
// a service from the cluster.
Services servicesClient
}
func (a Args) validate(ctx context.Context) error {
if a.Name == "" {
return fmt.Errorf("Name cannot be an empty string")
}
if a.Storage == nil {
return fmt.Errorf("Storage cannot be nil")
}
if a.Services == nil {
return fmt.Errorf("Services cannot be nil")
}
return nil
}
這里設置了要通過狀態傳遞的參數,可以將在一個狀態中設置并傳遞到另一個狀態的私有字段包括在內。
請注意,Args 并非指針。
由于我們修改了 Args 并將其傳遞給每個狀態,因此不需要給垃圾回收器增加負擔。對于像這樣操作來說,這點節約微不足道,但在工作量大的 ETL 管道中,節約的時間可能就很明顯了。
實現中包含 validate() 方法,用于測試參數是否滿足使用的最低基本要求。
// Service removes a service from a cluster and associated storage.
// The last 3 storage backups are retained for whatever the storage retainment
// period is.
func Service(ctx context.Context, args Args) error {
if err := args.validate(); err != nil {
return err
}
start := drainService
_, err := Run[Args](ctx, args, start "Args")
if err != nil {
return fmt.Errorf("problem removing service %q: %w", args.Name, err)
}
return nil
}
用戶只需調用 Service(),傳入 Args,如果出錯就會收到錯誤信息。用戶不需要看到狀態機模式,也不需要理解狀態機模式就能執行操作。
我們只需驗證 Args 是否正確,將狀態機的起始狀態設置為名為 drainService 的函數,然后調用上面定義的 Run() 函數即可。
func drainService(ctx context.Context, args Args) (Args, State[Args], error) {
l, err := args.Services.List(ctx)
if err != nil {
return args, nil, err
}
found := false
for _, entry := range l {
if entry == args.Name {
found = true
break
}
}
if !found {
return args, nil, fmt.Errorf("the service was not found")
}
if err := args.Services.Drain(ctx, args.Name); err != nil {
return args, nil, fmt.Errorf("problem draining the service: %w", err)
}
return args, removeService, nil
}
我們的第一個狀態叫做 drainService(),實現了上面定義的狀態類型。
它使用 Args 中定義的 Services 客戶端列出集群中的所有服務,如果找不到服務,就會返回錯誤并結束狀態機。
如果找到服務,就會對服務執行關閉。一旦完成,就進入下一個狀態,即 removeService()。
func removeService(ctx context.Context, args Args) (Args, State[Args], error) {
if err := args.Services.Remove(ctx, args.Name); err != nil {
return args, nil, fmt.Errorf("could not remove the service: %w", err)
}
hasStorage, err := args.Services.HasStorage(ctx, args.Name)
if err != nil {
return args, nil, fmt.Errorf("HasStorage() failed: %w", err)
}
if hasStorage{
return args, removeBackups, nil
}
return args, nil, nil
}
removeService() 使用我們的 Services 客戶端將服務從群集中移除。
調用 HasStorage() 方法確定是否有存儲,如果有,就會進入 removeBackups() 狀態,否則就會返回 args, nil, nil,這將導致狀態機在無錯誤的情況下退出。
這個示例說明如何根據 Args 中的信息或代碼中的遠程調用在狀態機中創建分支。
其他狀態調用由你自行決定。我們看看這種設計如何更適合測試此類操作。
測試優勢
這種模式首先鼓勵的是小塊的可測試代碼,模塊變得很容易分割,這樣當代碼塊變得太大時,只需創建新的狀態來隔離代碼塊。
但更大的優勢在于無需進行大規模端到端測試。由于操作流程中的每個階段都需要調用下一階段,因此會出現以下情況:
- 頂層調用者按一定順序調用所有子函數
- 每個調用者都會調用下一個函數
- 兩者的某種混合
兩者都會導致某種類型的端到端測試,而這種測試本不需要。
如果我們對頂層調用者方法進行編碼,可能看起來像這樣:
func Service(ctx context.Context, args Args) error {
...
if err := drainService(ctx, args); err != nil {
return err
}
if err := removeService(ctx, args); err != nil {
return err
}
hasStorage, err := args.Services.HasStorage(ctx, args.Name)
if err != nil {
return err
}
if hasStorage{
if err := removeBackups(ctx, args); err != nil {
return err
}
if err := removeStorage(ctx, args); err != nil {
return err
}
}
return nil
}
如你所見,可以為所有子函數編寫單獨的測試,但要測試 Service(),現在必須對調用的所有客戶端或方法打樁。這看起來就像是端到端測試,而對于這類代碼來說,通常不是好主意。
如果轉到功能調用鏈,情況也不會好到哪里去:
func Service(ctx context.Context, args Args) error {
...
return drainService(ctx, args)
}
func drainService(ctx context.Context, args Args) (Args, error) {
...
return removeService(ctx, args)
}
func removeService(ctx context.Context, args Args) (Args, error) {
...
hasStorage, err := args.Services.HasStorage(ctx, args.Name)
if err != nil {
return args, fmt.Errorf("HasStorage() failed: %w", err)
}
if hasStorage{
return removeBackups(ctx, args)
}
return nil
}
...
當我們測試時,越接近調用鏈的頂端,測試的實現就變得越困難。在 Service() 中,必須測試 drainService()、removeService() 以及下面所有調用。
有幾種方法可以做到,但都不太好。
如果使用狀態機,只需測試每個階段是否按要求運行,并返回想要的下一階段。
頂層調用者甚至不需要測試,它只是調用 validate() 方法,并調用應該能夠被測試的 Run() 函數。
我們為 drainService() 編寫一個表驅動測試,這里會拷貝一份 drainService() 代碼,這樣就不用返回到前面看代碼了。
func drainService(ctx context.Context, args Args) (Args, State[Args], error) {
l, err := args.Services.List(ctx)
if err != nil {
return args, nil, err
}
found := false
for _, entry := range l {
if entry == args.Name {
found = true
break
}
}
if !found {
return args, nil, fmt.Errorf("the service was not found")
}
if err := args.Services.Drain(ctx, args.Name); err != nil {
return args, nil, fmt.Errorf("problem draining the service: %w", err)
}
return args, removeService, nil
}
func TestDrainSerivce(t *testing.T) {
t.Parallel()
tests := []struct {
name string
args Args
wantErr bool
wantState State[Args]
}{
{
name: "Error: Services.List() returns an error",
args: Args{
Services: &fakeServices{
list: fmt.Errorf("error"),
},
},
wantErr: true,
},
{
name: "Error: Services.List() didn't contain our service name",
args: Args{
Name: "myService",
Services: &fakeServices{
list: []string{"nope", "this", "isn't", "it"},
},
},
wantErr: true,
},
{
name: "Error: Services.Drain() returned an error",
args: Args{
Name: "myService",
Services: &fakeServices{
list: []string{"yes", "mySerivce", "is", "here"},
drain: fmt.Errorf("error"),
},
},
wantErr: true,
},
{
name: "Success",
args: Args{
Name: "myService",
Services: &fakeServices{
list: []string{"yes", "myService", "is", "here"},
drain: nil,
},
},
wantState: removeService,
},
}
for _, test := range tests {
_, nextState, err := drainService(context.Background(), test.args)
switch {
case err == nil && test.wantErr:
t.Errorf("TestDrainService(%s): got err == nil, want err != nil", test.name)
continue
case err != nil && !test.wantErr:
t.Errorf("TestDrainService(%s): got err == %s, want err == nil", test.name, err)
continue
case err != nil:
continue
}
gotState := methodName(nextState)
wantState := methodName(test.wantState)
if gotState != wantState {
t.Errorf("TestDrainService(%s): got next state %s, want %s", test.name, gotState, wantState)
}
}
}
可以在 Go Playground[3]玩一下。
如你所見,這避免了測試整個調用鏈,同時還能確保測試調用鏈中的下一個函數。
這些測試很容易劃分,維護人員也很容易遵循。
其他可能性
這種模式也有變種,即根據 Args 中設置的字段確定狀態,并跟蹤狀態的執行以防止循環。
在第一種情況下,狀態機軟件包可能是這樣的:
type State[T any] func(ctx context.Context, args T) (T, State[T], error)
type Args[T] struct {
Data T
Next State
}
func Run[T any](ctx context.Context, args Args[T], start State[T] "T any") (T, error) {
var err error
current := start
for {
if ctx.Err() != nil {
return args, ctx.Err()
}
args, current, err = current(ctx, args)
if err != nil {
return args, err
}
current = args.Next // Set our next stage
args.Next = nil // Clear this so to prevent infinite loops
if current == nil {
return args, nil
}
}
}
可以很容易的將分布式跟蹤或日志記錄集成到這種設計中。
如果希望推送大量數據并利用并發優勢,不妨試試 stagedpipe 軟件包[4],其內置了大量高級功能,可以看視頻和 README 學習如何使用。
希望這篇文章能讓你充分了解 Go 狀態機設計模式,現在你的工具箱里多了一個強大的新工具。