【ETCD】【源码阅读】深入分析 storeTxnWrite.Put方法源码

该方法是 storeTxnWrite 类型中的核心方法，负责将键值对存储到数据库，同时处理键的元数据（如版本、修订号、租约）并管理租约关联。

一、完整代码

func (tw *storeTxnWrite) put(key, value []byte, leaseID lease.LeaseID) {rev := tw.beginRev + 1c := revoldLease := lease.NoLease// if the key exists before, use its previous created and// get its previous leaseID_, created, ver, err := tw.s.kvindex.Get(key, rev)if err == nil {c = created.mainoldLease = tw.s.le.GetLease(lease.LeaseItem{Key: string(key)})tw.trace.Step("get key's previous created_revision and leaseID")}ibytes := newRevBytes()idxRev := revision{main: rev, sub: int64(len(tw.changes))}revToBytes(idxRev, ibytes)ver = ver + 1kv := mvccpb.KeyValue{Key:            key,Value:          value,CreateRevision: c,ModRevision:    rev,Version:        ver,Lease:          int64(leaseID),}d, err := kv.Marshal()if err != nil {tw.storeTxnRead.s.lg.Fatal("failed to marshal mvccpb.KeyValue",zap.Error(err),)}tw.trace.Step("marshal mvccpb.KeyValue")tw.tx.UnsafeSeqPut(buckets.Key, ibytes, d)tw.s.kvindex.Put(key, idxRev)tw.changes = append(tw.changes, kv)tw.trace.Step("store kv pair into bolt db")if oldLease != lease.NoLease {if tw.s.le == nil {panic("no lessor to detach lease")}err = tw.s.le.Detach(oldLease, []lease.LeaseItem{{Key: string(key)}})if err != nil {tw.storeTxnRead.s.lg.Error("failed to detach old lease from a key",zap.Error(err),)}}if leaseID != lease.NoLease {if tw.s.le == nil {panic("no lessor to attach lease")}err = tw.s.le.Attach(leaseID, []lease.LeaseItem{{Key: string(key)}})if err != nil {panic("unexpected error from lease Attach")}}tw.trace.Step("attach lease to kv pair")
}

---

二、方法详解

方法签名

func (tw *storeTxnWrite) put(key, value []byte, leaseID lease.LeaseID)

• key, value []byte：键值对的字节切片。
• leaseID lease.LeaseID：与键关联的租约 ID。如果为 lease.NoLease，表示没有租约。

1. 计算修订号并初始化变量

rev := tw.beginRev + 1
c := rev
oldLease := lease.NoLease

• rev：新键值对的修订号，比事务起始修订号 beginRev 大 1。
• c：表示键的 CreateRevision，初始值为当前修订号。
• oldLease：存储键之前的租约 ID，初始值设为 lease.NoLease（无租约）。

2. 检查键是否已存在

_, created, ver, err := tw.s.kvindex.Get(key, rev)
if err == nil {c = created.mainoldLease = tw.s.le.GetLease(lease.LeaseItem{Key: string(key)})tw.trace.Step("get key's previous created_revision and leaseID")
}

• tw.s.kvindex.Get：
  • 在 kvindex 中查找当前键是否存在（按当前修订号 rev）。
  • 如果键存在，获取：
    • created.main：键的初始创建修订号。
    • ver：键的当前版本号。
• c = created.main：如果键已存在，将其初始创建修订号 created.main 赋值给 c。
• oldLease = tw.s.le.GetLease：查找该键关联的旧租约（如果有）。
• tw.trace.Step：记录 “获取键的创建修订号和租约 ID” 这一步操作。

3. 生成索引修订号与字节表示

ibytes := newRevBytes()
idxRev := revision{main: rev, sub: int64(len(tw.changes))}
revToBytes(idxRev, ibytes)

• idxRev：为键生成索引修订号：
  • main：当前修订号 rev。
  • sub：当前事务中已变更的键值对数量（len(tw.changes)）。
• revToBytes：将修订号 idxRev 转换成字节数组 ibytes，用于存储到底层数据库。

---

4. 更新版本号并构建新的键值对

ver = ver + 1
kv := mvccpb.KeyValue{Key:            key,Value:          value,CreateRevision: c,ModRevision:    rev,Version:        ver,Lease:          int64(leaseID),
}

• ver = ver + 1：增加键的版本号。
• 构建新的 mvccpb.KeyValue 对象：
  • Key：键。
  • Value：值。
  • CreateRevision：键的创建修订号 c。
  • ModRevision：键的最新修改修订号 rev。
  • Version：键的版本号。
  • Lease：关联的租约 ID。

---

5. 序列化键值对并存储

d, err := kv.Marshal()
if err != nil {tw.storeTxnRead.s.lg.Fatal("failed to marshal mvccpb.KeyValue",zap.Error(err),)
}tw.trace.Step("marshal mvccpb.KeyValue")
tw.tx.UnsafeSeqPut(buckets.Key, ibytes, d)
tw.s.kvindex.Put(key, idxRev)
tw.changes = append(tw.changes, kv)
tw.trace.Step("store kv pair into bolt db")

• 序列化：将 kv 通过 Marshal 序列化为字节数组 d。
• 存储：
  • 调用 UnsafeSeqPut 将键值对存储到 Bolt DB 中。
  • 使用 kvindex.Put 更新键的索引，记录索引修订号 idxRev。
  • 将 kv 添加到事务变更列表 tw.changes 中。
• tw.trace.Step：记录 “序列化键值对” 和 “存储到 Bolt DB” 的步骤。

6. 处理旧租约

if oldLease != lease.NoLease {if tw.s.le == nil {panic("no lessor to detach lease")}err = tw.s.le.Detach(oldLease, []lease.LeaseItem{{Key: string(key)}})if err != nil {tw.storeTxnRead.s.lg.Error("failed to detach old lease from a key",zap.Error(err),)}
}

• 如果键之前绑定了租约 oldLease，调用 Detach 方法解除该键与旧租约的关联。
• 如果 Detach 失败，记录错误日志。

7. 绑定新租约

if leaseID != lease.NoLease {if tw.s.le == nil {panic("no lessor to attach lease")}err = tw.s.le.Attach(leaseID, []lease.LeaseItem{{Key: string(key)}})if err != nil {panic("unexpected error from lease Attach")}
}
tw.trace.Step("attach lease to kv pair")

• 如果指定了新租约 leaseID，调用 Attach 将键绑定到新租约。
• 如果 Attach 失败，触发 panic。

---

三、总结

方法的主要功能：

1. 检查键是否存在：如果存在，获取其创建修订号和旧租约。
2. 生成修订号和版本号：为新键值对生成修订号和版本号。
3. 存储键值对：
   • 序列化键值对。
   • 存储到 Bolt DB。
   • 更新索引。
4. 处理租约：
   • 解除旧租约绑定。
   • 绑定新租约。
5. 事务追踪：通过 tw.trace 记录关键步骤，便于调试和性能分析。

核心步骤：

1. 查找旧数据与租约。
2. 构建新的键值对（包括版本、修订号）。
3. 存储键值对并更新索引。
4. 管理租约绑定/解绑。