《游戏服务端编程实践》11.4 微服务间通信模式（同步/异步）

// file: ch11/pattern/pattern.go
// tier: L1
// verify: go test ./ch11/pattern/

// Package pattern 对照同一笔充值订单的同步调用与异步事件两种实现。
package pattern

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"time"
)

var (
	ErrEmptyOrderID = errors.New("pattern: empty order id")
	ErrEmptyProduct = errors.New("pattern: empty product id")
	ErrNoPlayer     = errors.New("pattern: missing player")
	ErrBadAmount    = errors.New("pattern: non-positive diamonds or amount")
)

// Order 是一笔充值订单。OrderID 由支付侧生成，同时作为业务幂等键。
type Order struct {
	OrderID   string
	PlayerID  uint64
	ProductID string
	Diamonds  int64
	AmountFen int64
}

func (o Order) Validate() error {
	switch {
	case o.OrderID == "":
		return ErrEmptyOrderID
	case o.PlayerID == 0:
		return ErrNoPlayer
	case o.ProductID == "":
		return ErrEmptyProduct
	case o.Diamonds <= 0 || o.AmountFen <= 0:
		return ErrBadAmount
	}
	return nil
}

// Receipt 是一次发放的回执。Applied 为 false 表示这笔订单此前已经发过。
type Receipt struct {
	OrderID  string
	PlayerID uint64
	Diamonds int64
	Balance  int64
	Applied  bool
}

// Wallet 用订单号做幂等，重复 Apply 不会增加余额。
type Wallet struct {
	mu       sync.Mutex
	balances map[uint64]int64
	receipts map[string]Receipt
}

func NewWallet() *Wallet {
	return &Wallet{balances: make(map[uint64]int64), receipts: make(map[string]Receipt)}
}

func (w *Wallet) Balance(playerID uint64) int64 {
	w.mu.Lock()
	defer w.mu.Unlock()
	return w.balances[playerID]
}

// Apply 首次入账返回 Applied=true，重复订单返回历史回执与 Applied=false。
func (w *Wallet) Apply(o Order) (Receipt, error) {
	if err := o.Validate(); err != nil {
		return Receipt{}, err
	}
	w.mu.Lock()
	defer w.mu.Unlock()
	if r, ok := w.receipts[o.OrderID]; ok {
		r.Applied = false
		return r, nil
	}
	w.balances[o.PlayerID] += o.Diamonds
	r := Receipt{OrderID: o.OrderID, PlayerID: o.PlayerID, Diamonds: o.Diamonds, Balance: w.balances[o.PlayerID], Applied: true}
	w.receipts[o.OrderID] = r
	return r, nil
}

// Backend 是支付/发货下游的最小接口，Status 用于超时后的对账。
type Backend interface {
	Grant(ctx context.Context, o Order) (Receipt, error)
	Status(ctx context.Context, orderID string) (Receipt, bool, error)
}

type SyncStatus string

const (
	SyncApplied  SyncStatus = "applied"
	SyncDegraded SyncStatus = "degraded"
	SyncRejected SyncStatus = "rejected"
)

type SyncResult struct {
	Status  SyncStatus
	Receipt Receipt
	Err     error
}

// UncertainLog 保存“已经发出、但没拿到确定结果”的订单。
type UncertainLog struct {
	mu     sync.Mutex
	orders map[string]Order
}

func NewUncertainLog() *UncertainLog {
	return &UncertainLog{orders: make(map[string]Order)}
}

func (u *UncertainLog) Add(o Order) {
	u.mu.Lock()
	defer u.mu.Unlock()
	u.orders[o.OrderID] = o
}

func (u *UncertainLog) Remove(orderID string) {
	u.mu.Lock()
	defer u.mu.Unlock()
	delete(u.orders, orderID)
}

func (u *UncertainLog) Pending() []Order {
	u.mu.Lock()
	defer u.mu.Unlock()
	out := make([]Order, 0, len(u.orders))
	for _, o := range u.orders {
		out = append(out, o)
	}
	return out
}

// SyncService 走同步请求-响应：调用方等待下游结果，超时进入待对账状态。
type SyncService struct {
	backend   Backend
	timeout   time.Duration
	uncertain *UncertainLog
}

func NewSyncService(b Backend, timeout time.Duration, log *UncertainLog) *SyncService {
	if log == nil {
		log = NewUncertainLog()
	}
	return &SyncService{backend: b, timeout: timeout, uncertain: log}
}

// Purchase 发起同步发放。超时不等于失败，只表示结果未知。
func (s *SyncService) Purchase(ctx context.Context, o Order) SyncResult {
	if err := o.Validate(); err != nil {
		return SyncResult{Status: SyncRejected, Err: err}
	}
	ctx, cancel := context.WithTimeout(ctx, s.timeout)
	defer cancel()
	r, err := s.backend.Grant(ctx, o)
	if err == nil {
		s.uncertain.Remove(o.OrderID)
		return SyncResult{Status: SyncApplied, Receipt: r}
	}
	if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) {
		s.uncertain.Add(o)
		return SyncResult{Status: SyncDegraded, Err: err}
	}
	return SyncResult{Status: SyncRejected, Err: err}
}

// Event 是异步路径上的领域事件。EventID 用于传输层去重，OrderID 用于业务幂等。
type Event struct {
	EventID string
	Type    string
	Order   Order
	At      time.Time
}

// Sink 是事件出口，生产环境通常由消息队列实现。
type Sink interface {
	Publish(Event) error
}

type AsyncStats struct {
	Published  int
	Applied    int
	Duplicates int
	Rejected   int
}

// AsyncService 先发布事件再异步入账，调用方拿到的是“已受理”而不是“已到账”。
type AsyncService struct {
	wallet *Wallet
	sink   Sink

	mu    sync.Mutex
	seen  map[string]struct{}
	stats AsyncStats
}

func NewAsyncService(w *Wallet, sink Sink) *AsyncService {
	return &AsyncService{wallet: w, sink: sink, seen: make(map[string]struct{})}
}

// Purchase 校验订单并发布事件；本地余额此时不会变化。
func (a *AsyncService) Purchase(ctx context.Context, o Order) (Event, error) {
	if err := ctx.Err(); err != nil {
		return Event{}, err
	}
	if err := o.Validate(); err != nil {
		return Event{}, err
	}
	ev := Event{
		EventID: fmt.Sprintf("evt-%s-%d", o.OrderID, time.Now().UnixNano()),
		Type:    "order.paid",
		Order:   o,
		At:      time.Now(),
	}
	if err := a.sink.Publish(ev); err != nil {
		return Event{}, err
	}
	a.mu.Lock()
	a.stats.Published++
	a.mu.Unlock()
	return ev, nil
}

// Handle 消费事件。事件号去重挡重复投递，账本订单号幂等挡语义重复。
func (a *AsyncService) Handle(ctx context.Context, ev Event) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	a.mu.Lock()
	defer a.mu.Unlock()
	if _, ok := a.seen[ev.EventID]; ok {
		a.stats.Duplicates++
		return nil
	}
	r, err := a.wallet.Apply(ev.Order)
	if err != nil {
		a.stats.Rejected++
		return err
	}
	a.seen[ev.EventID] = struct{}{}
	if r.Applied {
		a.stats.Applied++
	} else {
		a.stats.Duplicates++
	}
	return nil
}

func (a *AsyncService) Stats() AsyncStats {
	a.mu.Lock()
	defer a.mu.Unlock()
	return a.stats
}

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

ok  	gspractice/ch11/pattern	0.272s