Golang 狀態(tài)機(jī)設(shè)計(jì)模式，你知道多少？
導(dǎo)言
在我們開發(fā)的許多項(xiàng)目中，都需要依賴某種運(yùn)行狀態(tài)從而實(shí)現(xiàn)連續(xù)操作。
這方面的例子包括：
- 解析配置語言、編程語言等
- 在系統(tǒng)、路由器、集群上執(zhí)行操作...
- ETL（Extract Transform Load，提取轉(zhuǎn)換加載）
很久以前，Rob Pike 有一個(gè)關(guān)于 Go 中詞法掃描[2]的演講，內(nèi)容很講座，我看了好幾遍才真正理解。但演講中介紹的最基本知識之一就是某個(gè)版本的 Go 狀態(tài)機(jī)。
該狀態(tài)機(jī)利用了 Go 的能力，即從函數(shù)中創(chuàng)建類型并將函數(shù)賦值給變量。
他在演講中介紹的狀態(tài)機(jī)功能強(qiáng)大，打破了讓函數(shù)執(zhí)行 if/else 并調(diào)用下一個(gè)所需函數(shù)的邏輯。取而代之的是，每個(gè)狀態(tài)都會返回下一個(gè)需要調(diào)用的函數(shù)。
這樣就能將調(diào)用鏈分成更容易測試的部分。
調(diào)用鏈
下面是一個(gè)用簡單的調(diào)用鏈來完成任務(wù)的例子：
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
}兩種方法都表示調(diào)用鏈，其中 Caller() 調(diào)用 callA()，并最終調(diào)用 callB()，從中可以看到這一系列調(diào)用是如何執(zhí)行的。
當(dāng)然，這種設(shè)計(jì)沒有任何問題，但當(dāng)調(diào)用者遠(yuǎn)程調(diào)用其他系統(tǒng)時(shí)，必須對這些遠(yuǎn)程調(diào)用進(jìn)行模擬/打樁，以提供密封測試。
你可能還想實(shí)現(xiàn)條件調(diào)用鏈，即根據(jù)某些參數(shù)或狀態(tài)，在特定條件下通過 if/else 調(diào)用不同函數(shù)。
這就意味著，要對 Caller() 進(jìn)行密封測試，可能需要處理整個(gè)調(diào)用鏈中的樁函數(shù)。如果有 50 個(gè)調(diào)用層級，則可能需要對被測函數(shù)下面每個(gè)層級的所有函數(shù)進(jìn)行模擬/打樁。
這正是 Pike 的狀態(tài)機(jī)設(shè)計(jì)大顯身手的地方。
狀態(tài)機(jī)模式
首先定義狀態(tài)：
type State[T any] func(ctx context.Context, args T) (T, State[T], error)狀態(tài)表示為函數(shù)/方法，接收一組參數(shù)（任意類型 T），并返回下一個(gè)狀態(tài)及其參數(shù)或錯(cuò)誤信息。
如果返回的狀態(tài)為 nil，那么狀態(tài)機(jī)將停止運(yùn)行。如果設(shè)置了 error，狀態(tài)機(jī)也將停止運(yùn)行。因?yàn)榉祷氐氖窍乱粋€(gè)要運(yùn)行的狀態(tài)，所以根據(jù)不同的條件，會有不同的下一個(gè)狀態(tài)。
這個(gè)版本與 Pike 的狀態(tài)機(jī)的不同之處在于這里包含了泛型并返回 T。這樣我們就可以創(chuàng)建純粹的函數(shù)式狀態(tài)機(jī)（如果需要的話），可以返回某個(gè)類型，并將其傳遞給下一個(gè)狀態(tài)。Pike 最初實(shí)現(xiàn)狀態(tài)機(jī)設(shè)計(jì)時(shí)還沒有使用泛型。
為了實(shí)現(xiàn)這一目標(biāo)，需要一個(gè)狀態(tài)驅(qū)動(dòng)程序：
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
}
}
}寥寥幾行代碼，我們就有了一個(gè)功能強(qiáng)大的狀態(tài)驅(qū)動(dòng)程序。
下面來看一個(gè)例子，在這個(gè)例子中，我們?yōu)榧褐械姆?wù)關(guān)閉操作編寫了狀態(tài)機(jī)：
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)
}這里定義了幾個(gè)需要客戶實(shí)現(xiàn)的私有接口，以便從集群中移除服務(wù)。
我們定義了私有接口，以防止他人使用我們的定義，但會通過公有變量公開這些接口。這樣，我們就能與客戶保持松耦合，保證只使用我們需要的方法。
// 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
}這里設(shè)置了要通過狀態(tài)傳遞的參數(shù)，可以將在一個(gè)狀態(tài)中設(shè)置并傳遞到另一個(gè)狀態(tài)的私有字段包括在內(nèi)。
請注意，Args 并非指針。
由于我們修改了 Args 并將其傳遞給每個(gè)狀態(tài)，因此不需要給垃圾回收器增加負(fù)擔(dān)。對于像這樣操作來說，這點(diǎn)節(jié)約微不足道，但在工作量大的 ETL 管道中，節(jié)約的時(shí)間可能就很明顯了。
實(shí)現(xiàn)中包含 validate() 方法，用于測試參數(shù)是否滿足使用的最低基本要求。
// 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
}用戶只需調(diào)用 Service()，傳入 Args，如果出錯(cuò)就會收到錯(cuò)誤信息。用戶不需要看到狀態(tài)機(jī)模式，也不需要理解狀態(tài)機(jī)模式就能執(zhí)行操作。
我們只需驗(yàn)證 Args 是否正確，將狀態(tài)機(jī)的起始狀態(tài)設(shè)置為名為 drainService 的函數(shù)，然后調(diào)用上面定義的 Run() 函數(shù)即可。
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
}我們的第一個(gè)狀態(tài)叫做 drainService()，實(shí)現(xiàn)了上面定義的狀態(tài)類型。
它使用 Args 中定義的 Services 客戶端列出集群中的所有服務(wù)，如果找不到服務(wù)，就會返回錯(cuò)誤并結(jié)束狀態(tài)機(jī)。
如果找到服務(wù)，就會對服務(wù)執(zhí)行關(guān)閉。一旦完成，就進(jìn)入下一個(gè)狀態(tài)，即 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 客戶端將服務(wù)從群集中移除。
調(diào)用 HasStorage() 方法確定是否有存儲，如果有，就會進(jìn)入 removeBackups() 狀態(tài)，否則就會返回 args, nil, nil，這將導(dǎo)致狀態(tài)機(jī)在無錯(cuò)誤的情況下退出。
這個(gè)示例說明如何根據(jù) Args 中的信息或代碼中的遠(yuǎn)程調(diào)用在狀態(tài)機(jī)中創(chuàng)建分支。
其他狀態(tài)調(diào)用由你自行決定。我們看看這種設(shè)計(jì)如何更適合測試此類操作。
測試優(yōu)勢
這種模式首先鼓勵(lì)的是小塊的可測試代碼，模塊變得很容易分割，這樣當(dāng)代碼塊變得太大時(shí)，只需創(chuàng)建新的狀態(tài)來隔離代碼塊。
但更大的優(yōu)勢在于無需進(jìn)行大規(guī)模端到端測試。由于操作流程中的每個(gè)階段都需要調(diào)用下一階段，因此會出現(xiàn)以下情況：
- 頂層調(diào)用者按一定順序調(diào)用所有子函數(shù)
- 每個(gè)調(diào)用者都會調(diào)用下一個(gè)函數(shù)
- 兩者的某種混合
兩者都會導(dǎo)致某種類型的端到端測試，而這種測試本不需要。
如果我們對頂層調(diào)用者方法進(jìn)行編碼，可能看起來像這樣：
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
} 如你所見，可以為所有子函數(shù)編寫單獨(dú)的測試，但要測試 Service()，現(xiàn)在必須對調(diào)用的所有客戶端或方法打樁。這看起來就像是端到端測試，而對于這類代碼來說，通常不是好主意。
如果轉(zhuǎn)到功能調(diào)用鏈，情況也不會好到哪里去：
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
}
...當(dāng)我們測試時(shí)，越接近調(diào)用鏈的頂端，測試的實(shí)現(xiàn)就變得越困難。在 Service() 中，必須測試 drainService()、removeService() 以及下面所有調(diào)用。
有幾種方法可以做到，但都不太好。
如果使用狀態(tài)機(jī)，只需測試每個(gè)階段是否按要求運(yùn)行，并返回想要的下一階段。
頂層調(diào)用者甚至不需要測試，它只是調(diào)用 validate() 方法，并調(diào)用應(yīng)該能夠被測試的 Run() 函數(shù)。
我們?yōu)?nbsp;drainService() 編寫一個(gè)表驅(qū)動(dòng)測試，這里會拷貝一份 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]玩一下。
如你所見，這避免了測試整個(gè)調(diào)用鏈，同時(shí)還能確保測試調(diào)用鏈中的下一個(gè)函數(shù)。
這些測試很容易劃分，維護(hù)人員也很容易遵循。
其他可能性
這種模式也有變種，即根據(jù) Args 中設(shè)置的字段確定狀態(tài)，并跟蹤狀態(tài)的執(zhí)行以防止循環(huán)。
在第一種情況下，狀態(tài)機(jī)軟件包可能是這樣的:
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
}
}
}可以很容易的將分布式跟蹤或日志記錄集成到這種設(shè)計(jì)中。
如果希望推送大量數(shù)據(jù)并利用并發(fā)優(yōu)勢，不妨試試 stagedpipe 軟件包[4]，其內(nèi)置了大量高級功能，可以看視頻和 README 學(xué)習(xí)如何使用。
希望這篇文章能讓你充分了解 Go 狀態(tài)機(jī)設(shè)計(jì)模式，現(xiàn)在你的工具箱里多了一個(gè)強(qiáng)大的新工具。
