一致性哈希

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

// Consistent holds the information about the members of the consistent hash circle.
type Consistent struct {
	mu sync.RWMutex

	config         Config
	hasher         Hasher
	sortedSet      []uint64           //节点下标，已排好序，便于对某个 parttion 哈希得到相应下标后找到其邻近的节点
	partitionCount uint64             //分区数量，如果 key 比较多，可以设置大一些
	loads          map[string]float64 // 每个节点对应的对应的负载
	members        map[string]*Member // 节点
	partitions     map[int]*Member    // 通过 partition id 找到其对应节点
	ring           map[uint64]*Member // 哈希环
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

// New creates and returns a new Consistent object.
func New(members []Member, config Config) *Consistent {
	c := &Consistent{
		config:         config,
		members:        make(map[string]*Member),
		partitionCount: uint64(config.PartitionCount),
		ring:           make(map[uint64]*Member),
	}
	if config.Hasher == nil {
		panic("Hasher cannot be nil")
	}
	// TODO: Check configuration here
	c.hasher = config.Hasher
	for _, member := range members {
		c.add(member)
	}
	if members != nil {
		c.distributePartitions()
	}
	return c
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

func (c *Consistent) add(member Member) {
	// 多少个副本构造出多少个节点
	for i := 0; i < c.config.ReplicationFactor; i++ {
		key := []byte(fmt.Sprintf("%s%d", member.String(), i))
		h := c.hasher.Sum64(key)
		c.ring[h] = &member
		// 将哈希值记录下来 便于后面排序
		c.sortedSet = append(c.sortedSet, h)
	}
	// 对哈希值排序
	sort.Slice(c.sortedSet, func(i int, j int) bool {
		return c.sortedSet[i] < c.sortedSet[j]
	})
	// parttion 可以通过节点名找到真正的节点
	c.members[member.String()] = &member
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

func (c *Consistent) distributePartitions() {
	loads := make(map[string]float64)
	partitions := make(map[int]*Member)

	bs := make([]byte, 8)
	for partID := uint64(0); partID < c.partitionCount; partID++ {
		binary.LittleEndian.PutUint64(bs, partID)
		key := c.hasher.Sum64(bs)
		// 二分查找邻近节点的下标
		idx := sort.Search(len(c.sortedSet), func(i int) bool {
			return c.sortedSet[i] >= key
		})
		if idx >= len(c.sortedSet) {
			idx = 0
		}
		// 将 partition 映射到查找的节点上，考虑 load 因素
		c.distributeWithLoad(int(partID), idx, partitions, loads)
	}
	c.partitions = partitions
	c.loads = loads
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

// AverageLoad exposes the current average load.
// 负载上限，config.Load 知道一个节点最多可以容纳多少个 partition
func (c *Consistent) AverageLoad() float64 {
	avgLoad := float64(c.partitionCount/uint64(len(c.members))) * c.config.Load
	return math.Ceil(avgLoad)
}

func (c *Consistent) distributeWithLoad(partID, idx int, partitions map[int]*Member, loads map[string]float64) {
	avgLoad := c.AverageLoad()
	var count int
	for {
		count++
		// 所有的节点都已超出负载
		if count >= len(c.sortedSet) {
			// User needs to decrease partition count, increase member count or increase load factor.
			panic("not enough room to distribute partitions")
		}
		i := c.sortedSet[idx]
		member := *c.ring[i]
		load := loads[member.String()]

		if load+1 <= avgLoad { //当前节点还可以未超载
			partitions[partID] = &member
			loads[member.String()]++
			return
		}
		// 当前节点已超载，尝试下一个节点
		idx++
		if idx >= len(c.sortedSet) {
			idx = 0
		}
	}
}