《后端架构设计·分布式与数据一致性》5.1 分片键、热点与一致性哈希

// file: internal/platform/sharding/ring.go
// tier: L1
// verify: go test ./internal/platform/sharding/ -count=1
package sharding

import (
	"fmt"
	"hash/crc32"
	"sort"
)

// Node identifies one shard.
type Node string

type point struct {
	hash uint32
	node Node
}

// Ring is an immutable consistent-hash ring. Build it once and route
// concurrently; Route has no mutable state.
type Ring struct {
	points   []point
	nodes    []Node
	replicas int
}

// NewRing places replicas virtual nodes for every shard on a CRC32 ring.
func NewRing(nodes []Node, replicas int) (*Ring, error) {
	if len(nodes) == 0 {
		return nil, fmt.Errorf("sharding: no nodes")
	}
	if replicas <= 0 {
		return nil, fmt.Errorf("sharding: replicas must be positive, got %d", replicas)
	}
	r := &Ring{replicas: replicas}
	seen := make(map[Node]bool, len(nodes))
	for _, n := range nodes {
		if n == "" {
			return nil, fmt.Errorf("sharding: empty node name")
		}
		if seen[n] {
			return nil, fmt.Errorf("sharding: duplicate node %q", n)
		}
		seen[n] = true
		r.nodes = append(r.nodes, n)
		for i := 0; i < replicas; i++ {
			r.points = append(r.points, point{
				hash: hashKey(fmt.Sprintf("%s#%d", n, i)),
				node: n,
			})
		}
	}
	sort.Slice(r.points, func(i, j int) bool {
		if r.points[i].hash != r.points[j].hash {
			return r.points[i].hash < r.points[j].hash
		}
		return r.points[i].node < r.points[j].node
	})
	return r, nil
}

func hashKey(key string) uint32 {
	return crc32.ChecksumIEEE([]byte(key))
}

// Route returns the shard that owns key.
func (r *Ring) Route(key string) Node {
	h := hashKey(key)
	i := sort.Search(len(r.points), func(i int) bool {
		return r.points[i].hash >= h
	})
	if i == len(r.points) {
		i = 0
	}
	return r.points[i].node
}

// Distribution counts keys per shard.
func Distribution(r *Ring, keys []string) map[Node]int {
	out := make(map[Node]int, len(r.nodes))
	for _, k := range keys {
		out[r.Route(k)]++
	}
	return out
}

// MaxDeviation returns the largest distance between a shard's key share and
// the ideal 1/len(nodes).
func MaxDeviation(r *Ring, keys []string) float64 {
	if len(keys) == 0 {
		return 0
	}
	ideal := 1.0 / float64(len(r.nodes))
	dist := Distribution(r, keys)
	var worst float64
	for _, n := range r.nodes {
		d := float64(dist[n])/float64(len(keys)) - ideal
		if d < 0 {
			d = -d
		}
		if d > worst {
			worst = d
		}
	}
	return worst
}

// Moved counts keys whose owner changes between two rings.
func Moved(oldRing, newRing *Ring, keys []string) int {
	moved := 0
	for _, k := range keys {
		if oldRing.Route(k) != newRing.Route(k) {
			moved++
		}
	}
	return moved
}

// HotKey is one logical key observed above the hotspot threshold.
type HotKey struct {
	Key   string
	Count int
}

// HotspotWindow counts accesses per logical key. It catches business hotspots
// that a uniform physical shard key cannot see.
type HotspotWindow struct {
	minCount int
	counts   map[string]int
}

// NewHotspotWindow flags keys with at least minCount observations.
func NewHotspotWindow(minCount int) *HotspotWindow {
	if minCount < 1 {
		minCount = 1
	}
	return &HotspotWindow{minCount: minCount, counts: make(map[string]int)}
}

// Observe adds n accesses for key.
func (w *HotspotWindow) Observe(key string, n int) {
	if n > 0 {
		w.counts[key] += n
	}
}

// HotKeys returns flagged keys by count descending, then key ascending.
func (w *HotspotWindow) HotKeys() []HotKey {
	var hot []HotKey
	for k, c := range w.counts {
		if c >= w.minCount {
			hot = append(hot, HotKey{Key: k, Count: c})
		}
	}
	sort.Slice(hot, func(i, j int) bool {
		if hot[i].Count != hot[j].Count {
			return hot[i].Count > hot[j].Count
		}
		return hot[i].Key < hot[j].Key
	})
	return hot
}

$ go test ./internal/platform/sharding/ -v -count=1
=== RUN   TestRingDistribution
    ring_test.go:26: shard=shard-0 keys=28139 share=0.2814
    ring_test.go:26: shard=shard-1 keys=22112 share=0.2211
    ring_test.go:26: shard=shard-2 keys=23075 share=0.2308
    ring_test.go:26: shard=shard-3 keys=26674 share=0.2667
--- PASS: TestRingDistribution (0.02s)
=== RUN   TestVirtualNodesReduceSkew
    ring_test.go:46: replicas=1 max_deviation=0.7359
    ring_test.go:47: replicas=256 max_deviation=0.0331
--- PASS: TestVirtualNodesReduceSkew (0.00s)
=== RUN   TestMigrationRatio
    ring_test.go:67: moved=21981/100000 share=0.2198
--- PASS: TestMigrationRatio (0.02s)
=== RUN   TestHotspotWindow
--- PASS: TestHotspotWindow (0.00s)
=== RUN   TestNewRingRejectsInvalidInput
--- PASS: TestNewRingRejectsInvalidInput (0.00s)
PASS
ok  	archlab/internal/platform/sharding	0.430s