《游戏服务端编程实践》21.1 云原生游戏后端趋势

// file: ch21/cloudnative/cloudnative.go
// tier: L1
// verify: go test ./ch21/cloudnative/

package cloudnative

import (
	"context"
	"errors"
	"sync"
)

var (
	ErrNotReady    = errors.New("cloudnative: node is not ready")
	ErrNotDraining = errors.New("cloudnative: node is not draining")
	ErrDraining    = errors.New("cloudnative: node is draining")
	ErrStopped     = errors.New("cloudnative: node is stopped")
	ErrBusy        = errors.New("cloudnative: node still has inflight requests")
)

// Phase 是节点在发布生命周期中的位置。摘流（draining）不是“正在退出”，
// 而是“不再接新请求，但旧请求还要跑完”，探针必须能区分这几个阶段。
type Phase string

const (
	PhaseStarting Phase = "starting"
	PhaseReady    Phase = "ready"
	PhaseDraining Phase = "draining"
	PhaseStopped  Phase = "stopped"
)

// Node 用一个互斥锁维护阶段和进行中请求数。idle 在请求数归零时关闭，
// WaitDrained 等待它即可，不需要定时轮询。
type Node struct {
	mu         sync.Mutex
	phase      Phase
	inflight   int
	idle       chan struct{}
	idleClosed bool
}

func NewNode() *Node {
	return &Node{
		phase:      PhaseStarting,
		idle:       closedChan(),
		idleClosed: true,
	}
}

func closedChan() chan struct{} {
	ch := make(chan struct{})
	close(ch)
	return ch
}

func (n *Node) resetIdleLocked() {
	n.idle = make(chan struct{})
	n.idleClosed = false
}

// MarkReady 由启动检查完成后调用。重复调用是幂等的，方便启动脚本重试。
func (n *Node) MarkReady() error {
	n.mu.Lock()
	defer n.mu.Unlock()

	switch n.phase {
	case PhaseStarting:
		n.phase = PhaseReady
		return nil
	case PhaseReady:
		return nil
	case PhaseDraining:
		return ErrDraining
	default:
		return ErrStopped
	}
}

// Admit 是业务入口的闸门：只有 ready 阶段放行。返回的 release 必须调用，
// 内部用 sync.Once 保证重复释放不会把计数减成负数。
func (n *Node) Admit() (func(), error) {
	n.mu.Lock()
	defer n.mu.Unlock()

	switch n.phase {
	case PhaseReady:
	case PhaseDraining:
		return nil, ErrDraining
	case PhaseStopped:
		return nil, ErrStopped
	default:
		return nil, ErrNotReady
	}

	if n.inflight == 0 && n.idleClosed {
		n.resetIdleLocked()
	}
	n.inflight++

	var once sync.Once
	return func() {
		once.Do(n.release)
	}, nil
}

func (n *Node) release() {
	n.mu.Lock()
	defer n.mu.Unlock()

	if n.inflight > 0 {
		n.inflight--
	}
	if n.inflight == 0 && !n.idleClosed {
		close(n.idle)
		n.idleClosed = true
	}
}

// BeginDrain 先让就绪探针失败，K8s 才会把该实例从 Service 端点摘掉。
// 必须等端点摘除生效后再等待存量请求，否则仍可能有新连接到达。
func (n *Node) BeginDrain() error {
	n.mu.Lock()
	defer n.mu.Unlock()

	switch n.phase {
	case PhaseReady:
		n.phase = PhaseDraining
		return nil
	case PhaseDraining:
		return nil
	case PhaseStopped:
		return ErrStopped
	default:
		return ErrNotReady
	}
}

// WaitDrained 等待进行中的请求归零；ctx 用于给优雅退出设上限。
func (n *Node) WaitDrained(ctx context.Context) error {
	n.mu.Lock()
	if n.phase != PhaseDraining && n.phase != PhaseStopped {
		n.mu.Unlock()
		return ErrNotDraining
	}
	if n.inflight == 0 {
		n.mu.Unlock()
		return nil
	}
	idle := n.idle
	n.mu.Unlock()

	select {
	case <-idle:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

// Stop 只允许在没有存量请求时执行，避免进程退出把正在写的状态截断。
func (n *Node) Stop() error {
	n.mu.Lock()
	defer n.mu.Unlock()

	if n.phase == PhaseStopped {
		return nil
	}
	if n.phase != PhaseDraining {
		return ErrNotDraining
	}
	if n.inflight != 0 {
		return ErrBusy
	}
	n.phase = PhaseStopped
	return nil
}

// Shutdown 是终止流程的推荐顺序：先摘流，再等存量请求，最后退出。
func (n *Node) Shutdown(ctx context.Context) error {
	if err := n.BeginDrain(); err != nil {
		return err
	}
	if err := n.WaitDrained(ctx); err != nil {
		return err
	}
	return n.Stop()
}

// Liveness 在摘流期间仍返回 true：进程还活着，只是不接新流量。
// 如果存活探针在摘流时失败，编排器可能直接杀掉进程，存量请求就白等了。
func (n *Node) Liveness() bool {
	return n.Phase() != PhaseStopped
}

// Readiness 只反映“能否接新请求”，业务依赖未就绪时必须返回 false。
func (n *Node) Readiness() bool {
	return n.Phase() == PhaseReady
}

func (n *Node) Phase() Phase {
	n.mu.Lock()
	defer n.mu.Unlock()
	return n.phase
}

func (n *Node) Inflight() int {
	n.mu.Lock()
	defer n.mu.Unlock()
	return n.inflight
}

cd .local/checks/gs-practice-lab && gofmt -l ch21/cloudnative && go vet ./ch21/cloudnative/ && go test ./ch21/cloudnative/ -count=1

ok  	gspractice/ch21/cloudnative	0.232s