PD 关于ID分配的源码分析

ID分配模块初始化

ID模块的实例化
s.idAllocator = id.NewAllocator(s.client, s.rootPath, s.member.MemberValue())

// NewAllocator creates a new ID Allocator.
func NewAllocator(client *clientv3.Client, rootPath string, member string) Allocator {
return &allocatorImpl{client: client, rootPath: rootPath, member: member}
}

ID分配模块的定义，

// allocatorImpl is used to allocate ID.
type allocatorImpl struct {
mu sync.Mutex
base uint64
end uint64
client *clientv3.Client
rootPath string
member string
}

ID模块核心算法就是分配ID，原理非常简单，内存维护一个id范围[base,end). etcd会保存上次分配的end大小（主要防止PD异常导致ID分配不线性唯一）。 每次调用函数Alloc（）分配ID的时候，流程如下：
1.如果base不等于end,.说明上次预分配还有多余，base++得到的值,返回当作
分配的ID
2.如果base等于end,说明上次预分配的所有的ID已经使用完，或者是PD重新启动（不管上次还有多少ID没用，我们依然从上次的end重新批量分配一些ID），无论那种情况，我们重新基于上次的end，批量分配一批ID:
2.1.调用etcd 查看上次end值是多少，key等于alloc.getAllocIDPath()
2.2.上次的end要么为0（pd 第一次启动），pd不为0(重新启动),取出end值，这个end值 作为下次分配ID起点，并计算新的END值end += allocStep，allocStep表示每次分配的ID数目
2.3.利用ETCD的事务机制，保存END值到ETCD

// Alloc returns a new id.
func (alloc *allocatorImpl) Alloc() (uint64, error) {
if alloc.base == alloc.end {
if err := alloc.rebaseLocked(); err != nil {
return 0, err
}
}
alloc.base++
return alloc.base, nil
}

// Rebase resets the base for the allocator from the persistent window boundary,
// which also resets the end of the allocator. (base, end) is the range that can
// be allocated in memory.
func (alloc *allocatorImpl) Rebase() error {
return alloc.rebaseLocked()
}
func (alloc *allocatorImpl) rebaseLocked() error {
key := alloc.getAllocIDPath()
value, err := etcdutil.GetValue(alloc.client, key)
if value == nil {
// create the key
cmp = clientv3.Compare(clientv3.CreateRevision(key), “=”, 0)
} else {
// update the key
end, err = typeutil.BytesToUint64(value)
cmp = clientv3.Compare(clientv3.Value(key), “=”, string(value))
}
end += allocStep
value = typeutil.Uint64ToBytes(end)
txn := kv.NewSlowLogTxn(alloc.client)
leaderPath := path.Join(alloc.rootPath, “leader”)
t := txn.If(append([]clientv3.Cmp{cmp}, clientv3.Compare(clientv3.Value(leaderPath), “=”, alloc.member))…)
resp, err := t.Then(clientv3.OpPut(key, string(value))).Commit()
alloc.end = end
alloc.base = end - allocStep
return nil
}