TiDB 中的全局唯一 ID

作者：Ed Huang链接：https://www.zhihu.com/question/52823076/answer/132331104来源：知乎著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。

// Generate a random cluster ID.ts := uint64(time.Now().Unix())clusterID := (ts << 32) + uint64(rand.Uint32())value := typeutil.Uint64ToBytes(clusterID)
// Multiple PDs may try to init the cluster ID at the same time.// Only one PD can commit this transaction, then other PDs can get// the committed cluster ID.resp, err := c.Txn(ctx).  If(clientv3.Compare(clientv3.CreateRevision(key), "=", 0)).  Then(clientv3.OpPut(key, string(value))).  Else(clientv3.OpGet(key)).  Commit()
// Txn commits ok, return the generated cluster ID.if resp.Succeeded {  return clusterID, nil}

service PD {  ...  rpc AllocID(AllocIDRequest) returns (AllocIDResponse) {}  ...}
message AllocIDRequest {    RequestHeader header = 1;}
message AllocIDResponse {    ResponseHeader header = 1;
    uint64 id = 2;}
message RequestHeader {    // cluster_id is the ID of the cluster which be sent to.    uint64 cluster_id = 1;}
message ResponseHeader {    // cluster_id is the ID of the cluster which sent the response.    uint64 cluster_id = 1;    Error error = 2;}enum ErrorType {    OK = 0;    UNKNOWN = 1;    NOT_BOOTSTRAPPED = 2;    STORE_TOMBSTONE = 3;    ALREADY_BOOTSTRAPPED = 4;    INCOMPATIBLE_VERSION = 5;    REGION_NOT_FOUND = 6;}
message Error {    ErrorType type = 1;    string message = 2;}

// AllocID implements gRPC PDServer.func (s *Server) AllocID(ctx context.Context, request *pdpb.AllocIDRequest) (*pdpb.AllocIDResponse, error) {  // 1. 校验请求数据，这里是在 pdpb.proto 中封装成 RequestHeader  if err := s.validateRequest(request.GetHeader()); err != nil {    return nil, err  }
  // 2. 我们用一个 idAllocator 来分配 ID  // We can use an allocator for all types ID allocation.  id, err := s.idAllocator.Alloc()  if err != nil {    return nil, status.Errorf(codes.Unknown, err.Error())  }
  // 3. 返回请求时的 header 和生成的 ID  return &pdpb.AllocIDResponse{    Header: s.header(),    Id:     id,  }, nil}

// Server is the pd server.type Server struct {
  // Server services.  // for id allocator, we can use one allocator for  // store, region and peer, because we just need  // a unique ID.  idAllocator *id.AllocatorImpl}
func (s *Server) startServer() error {}

func (s *Server) startServer() error {  s.idAllocator = id.NewAllocatorImpl(s.client, s.rootPath, s.member.MemberValue())}

// Allocator is the allocator to generate unique ID.type Allocator interface {  Alloc() (uint64, error)}// 步长 1000const allocStep = uint64(1000)
// AllocatorImpl 是对 Allocator 的实现，用于分配 ID// AllocatorImpl is used to allocate ID.type AllocatorImpl struct {  mu   sync.Mutex  base uint64  end  uint64
  // etcd client  client   *clientv3.Client  rootPath string  member   string}
// NewAllocatorImpl creates a new IDAllocator.func NewAllocatorImpl(client *clientv3.Client, rootPath string, member string) *AllocatorImpl {  return &AllocatorImpl{client: client, rootPath: rootPath, member: member}}

// Alloc returns a new id.func (alloc *AllocatorImpl) Alloc() (uint64, error) {  // 给分配增加锁，使用 defer 在函数结束时进行释放  alloc.mu.Lock()  defer alloc.mu.Unlock()
  // 第一次的时候 base 和 end 都为 0，所以会执行 generate()  // 否则直接返回 alloc.base++  if alloc.base == alloc.end {    end, err := alloc.generate()    if err != nil {      return 0, err    }
    alloc.end = end    alloc.base = alloc.end - allocStep  }
  alloc.base++
  return alloc.base, nil}
func (alloc *AllocatorImpl) generate() (uint64, error) {  // 获取要给 XXPath 分配 ID 的 key  key := alloc.getAllocIDPath()  // 从 etcd 中读取 key 所对应的值  value, err := etcdutil.GetValue(alloc.client, key)  if err != nil {    return 0, err  }
  var (    cmp clientv3.Cmp    end uint64  )
  if value == nil {    // create the key    cmp = clientv3.Compare(clientv3.CreateRevision(key), "=", 0)  } else {    // update the key    end, err = typeutil.BytesToUint64(value)    if err != nil {      return 0, err    }
    cmp = clientv3.Compare(clientv3.Value(key), "=", string(value))  }  // 如果以前不存在，则 end 被赋值为 1000（分配的步长），否则，就是原有的值+步长  end += allocStep  // 将 uint64 转为 bytes  value = typeutil.Uint64ToBytes(end)  // 从 etcd 获取一个事务，然后将值提交到 etcd 中  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()  if err != nil {    return 0, err  }  if !resp.Succeeded {    return 0, errors.New("generate id failed, we may not leader")  }
  log.Info("idAllocator allocates a new id", zap.Uint64("alloc-id", end))  idGauge.WithLabelValues("idalloc").Set(float64(end))  return end, nil}
func (alloc *AllocatorImpl) getAllocIDPath() string {  return path.Join(alloc.rootPath, "alloc_id")}

newRegionID, err := c.s.idAllocator.Alloc()
...
peerIDs := make([]uint64, len(request.Region.Peers))for i := 0; i < len(peerIDs); i++ {  if peerIDs[i], err = c.s.idAllocator.Alloc(); err != nil {    return nil, err  }}

// prometheusPushClient pushs metrics to Prometheus Pushgateway.func prometheusPushClient(job, addr string, interval time.Duration) {  for {    err := push.FromGatherer(      job, push.HostnameGroupingKey(),      addr,      prometheus.DefaultGatherer,    )    if err != nil {      log.Error("could not push metrics to Prometheus Pushgateway", zap.Error(err))    }
    time.Sleep(interval)  }}
// Push metircs in background.func Push(cfg *MetricConfig) {  if cfg.PushInterval.Duration == zeroDuration || len(cfg.PushAddress) == 0 {    log.Info("disable Prometheus push client")    return  }
  log.Info("start Prometheus push client")
  interval := cfg.PushInterval.Duration  go prometheusPushClient(cfg.PushJob, cfg.PushAddress, interval)}