《后端架构设计·高性能与高可用》5.2 舱壁隔离与资源配额

// file: internal/platform/bulkhead/bulkhead.go
// tier: L1
// verify: go test ./internal/platform/bulkhead/
package bulkhead

import (
	"context"
	"errors"
	"sync"
	"sync/atomic"
	"time"
)

var (
	ErrOverloaded = errors.New("bulkhead: queue full or wait timeout")
	ErrQuota      = errors.New("bulkhead: quota exceeded")
)

type Config struct {
	Name          string
	MaxConcurrent int
	MaxQueue      int
	WaitTimeout   time.Duration
	QPS           int // tokens per second; 0 means unlimited
	Burst         int
}

type Stats struct {
	InFlight, Queued, Completed, Rejected, QuotaRejected, MaxInFlight int64
}

// Bulkhead is one downstream's admission boundary: a semaphore caps concurrent
// calls, a bounded queue caps waiters, and an optional token bucket caps rate.
// The three limits protect different things and must not be collapsed into one.
type Bulkhead struct {
	cfg         Config
	sem         chan struct{}
	mu          sync.Mutex
	tokens      float64
	last        time.Time
	queued      atomic.Int64
	inFlight    atomic.Int64
	maxInFlight atomic.Int64
	completed   atomic.Int64
	rejected    atomic.Int64
	quotaDenied atomic.Int64
}

func New(cfg Config) *Bulkhead {
	if cfg.MaxConcurrent < 1 {
		cfg.MaxConcurrent = 1
	}
	if cfg.Burst < 1 {
		cfg.Burst = cfg.MaxConcurrent
	}
	return &Bulkhead{
		cfg: cfg, sem: make(chan struct{}, cfg.MaxConcurrent),
		tokens: float64(cfg.Burst), last: time.Now(),
	}
}

// Do admits one call. Quota is charged at admission; concurrency and queue
// bounds are enforced after it, so a rejected call has already spent its token.
func (b *Bulkhead) Do(ctx context.Context, fn func(context.Context) error) error {
	if !b.allowQuota() {
		b.quotaDenied.Add(1)
		return ErrQuota
	}
	select {
	case b.sem <- struct{}{}:
	default:
		if q := b.queued.Add(1); q > int64(b.cfg.MaxQueue) {
			b.queued.Add(-1)
			b.rejected.Add(1)
			return ErrOverloaded
		}
		timer := time.NewTimer(b.cfg.WaitTimeout)
		defer timer.Stop()
		select {
		case b.sem <- struct{}{}:
			b.queued.Add(-1)
		case <-timer.C:
			b.queued.Add(-1)
			b.rejected.Add(1)
			return ErrOverloaded
		case <-ctx.Done():
			b.queued.Add(-1)
			b.rejected.Add(1)
			return ctx.Err()
		}
	}
	bump(&b.maxInFlight, b.inFlight.Add(1))
	defer func() {
		b.inFlight.Add(-1)
		<-b.sem
		b.completed.Add(1)
	}()
	return fn(ctx)
}

func (b *Bulkhead) allowQuota() bool {
	if b.cfg.QPS <= 0 {
		return true
	}
	b.mu.Lock()
	defer b.mu.Unlock()
	now := time.Now()
	b.tokens += now.Sub(b.last).Seconds() * float64(b.cfg.QPS)
	if max := float64(b.cfg.Burst); b.tokens > max {
		b.tokens = max
	}
	b.last = now
	if b.tokens < 1 {
		return false
	}
	b.tokens--
	return true
}

func (b *Bulkhead) Stats() Stats {
	return Stats{
		InFlight: b.inFlight.Load(), Queued: b.queued.Load(),
		Completed: b.completed.Load(), Rejected: b.rejected.Load(),
		QuotaRejected: b.quotaDenied.Load(), MaxInFlight: b.maxInFlight.Load(),
	}
}

func bump(dst *atomic.Int64, v int64) {
	for cur := dst.Load(); v > cur; cur = dst.Load() {
		if dst.CompareAndSwap(cur, v) {
			return
		}
	}
}

// file: internal/platform/bulkhead/bulkhead_test.go
// tier: L1
// verify: go test ./internal/platform/bulkhead/ -count=1 -v
package bulkhead

import (
	"context"
	"errors"
	"sync"
	"sync/atomic"
	"testing"
	"time"
)

func call(ctx context.Context, b *Bulkhead, work time.Duration) error {
	return b.Do(ctx, func(context.Context) error {
		time.Sleep(work)
		return nil
	})
}

func TestIsolationKeepsHealthyDownstreamsMoving(t *testing.T) {
	ctx := context.Background()
	payment := New(Config{Name: "payment", MaxConcurrent: 2, MaxQueue: 1000, WaitTimeout: 2 * time.Second})
	inventory := New(Config{Name: "inventory", MaxConcurrent: 8, MaxQueue: 1000, WaitTimeout: 2 * time.Second})
	notify := New(Config{Name: "notify", MaxConcurrent: 8, MaxQueue: 1000, WaitTimeout: 2 * time.Second})

	var payWG, fastWG sync.WaitGroup
	payBegin := time.Now()
	for i := 0; i < 60; i++ {
		payWG.Add(1)
		go func() { defer payWG.Done(); _ = call(ctx, payment, 25*time.Millisecond) }()
	}
	time.Sleep(5 * time.Millisecond)

	invBegin := time.Now()
	for i := 0; i < 60; i++ {
		fastWG.Add(1)
		go func() { defer fastWG.Done(); _ = call(ctx, inventory, 2*time.Millisecond) }()
	}
	for i := 0; i < 60; i++ {
		fastWG.Add(1)
		go func() { defer fastWG.Done(); _ = call(ctx, notify, time.Millisecond) }()
	}
	fastWG.Wait()
	fastElapsed := time.Since(invBegin)
	paymentStillRunning := time.Since(payBegin) < 500*time.Millisecond
	payWG.Wait()

	if !paymentStillRunning {
		t.Fatalf("payment finished too early; test did not overlap")
	}
	if fastElapsed > 300*time.Millisecond {
		t.Fatalf("healthy downstreams took %s", fastElapsed)
	}
	t.Logf("payment_latency=25ms payment_calls=60 fast_calls=120 fast_elapsed=%s payment_still_running=%v",
		fastElapsed.Round(time.Millisecond), paymentStillRunning)

	shared := New(Config{Name: "shared", MaxConcurrent: 2, MaxQueue: 1000, WaitTimeout: 50 * time.Millisecond})
	var slowWG sync.WaitGroup
	for i := 0; i < 10; i++ {
		slowWG.Add(1)
		go func() { defer slowWG.Done(); _ = call(ctx, shared, 100*time.Millisecond) }()
	}
	time.Sleep(5 * time.Millisecond)
	var rejected atomic.Int64
	var healthy sync.WaitGroup
	for i := 0; i < 10; i++ {
		healthy.Add(1)
		go func() {
			defer healthy.Done()
			if err := call(ctx, shared, time.Millisecond); errors.Is(err, ErrOverloaded) {
				rejected.Add(1)
			}
		}()
	}
	healthy.Wait()
	slowWG.Wait()
	if rejected.Load() == 0 {
		t.Fatalf("shared bulkhead did not starve healthy calls")
	}
	t.Logf("shared_bulkhead_healthy_rejected=%d", rejected.Load())
}

func TestQueueBoundAndWaitTimeout(t *testing.T) {
	ctx := context.Background()
	b := New(Config{MaxConcurrent: 1, MaxQueue: 1, WaitTimeout: 20 * time.Millisecond})
	started := make(chan struct{})
	go func() {
		_ = b.Do(ctx, func(context.Context) error { close(started); time.Sleep(100 * time.Millisecond); return nil })
	}()
	<-started
	var wg sync.WaitGroup
	for i := 0; i < 2; i++ {
		wg.Add(1)
		go func() { defer wg.Done(); _ = call(ctx, b, time.Millisecond) }()
	}
	wg.Wait()
	if s := b.Stats(); s.Rejected != 2 {
		t.Fatalf("rejected = %d, want 2 (one queue-full, one wait-timeout)", s.Rejected)
	}
	t.Logf("queue_full_or_timeout_rejected=2")
}

func TestQuota(t *testing.T) {
	ctx := context.Background()
	b := New(Config{MaxConcurrent: 4, MaxQueue: 100, QPS: 10, Burst: 10})
	ok, denied := 0, 0
	for i := 0; i < 50; i++ {
		if err := call(ctx, b, 0); errors.Is(err, ErrQuota) {
			denied++
		} else {
			ok++
		}
	}
	if ok != 10 || denied != 40 {
		t.Fatalf("ok=%d denied=%d, want 10/40", ok, denied)
	}
	t.Logf("quota_qps=10 burst=10 allowed=%d quota_rejected=%d", ok, denied)
}

func TestConcurrencySweep(t *testing.T) {
	const calls = 40
	const latency = 15 * time.Millisecond
	for _, limit := range []int{1, 2, 4, 8} {
		b := New(Config{MaxConcurrent: limit, MaxQueue: calls, WaitTimeout: 2 * time.Second})
		var wg sync.WaitGroup
		begin := time.Now()
		for i := 0; i < calls; i++ {
			wg.Add(1)
			go func() { defer wg.Done(); _ = call(context.Background(), b, latency) }()
		}
		wg.Wait()
		elapsed := time.Since(begin)
		s := b.Stats()
		t.Logf("max_concurrent=%d calls=%d elapsed=%s throughput=%.0f/s max_in_flight=%d rejected=%d",
			limit, calls, elapsed.Round(time.Millisecond), float64(calls)/elapsed.Seconds(),
			s.MaxInFlight, s.Rejected)
		if s.Completed != calls || s.Rejected != 0 {
			t.Fatalf("completed=%d rejected=%d", s.Completed, s.Rejected)
		}
	}
}

$ go test ./internal/platform/bulkhead/ -count=1 -v
payment_latency=25ms payment_calls=60 fast_calls=120 fast_elapsed=18ms payment_still_running=true
shared_bulkhead_healthy_rejected=10
queue_full_or_timeout_rejected=2
quota_qps=10 burst=10 allowed=10 quota_rejected=40
max_concurrent=1 calls=40 elapsed=658ms throughput=61/s  max_in_flight=1 rejected=0
max_concurrent=2 calls=40 elapsed=326ms throughput=123/s max_in_flight=2 rejected=0
max_concurrent=4 calls=40 elapsed=167ms throughput=240/s max_in_flight=4 rejected=0
max_concurrent=8 calls=40 elapsed=82ms  throughput=490/s max_in_flight=8 rejected=0