etcd源码分析 - 4.【打通核心流程】processInternalRaftRequestOnce四个细节

简介: 在上一讲,我们继续梳理了`PUT`请求到`EtcdServer`这一层的逻辑,并大概阅读了其中的关键函数`processInternalRaftRequestOnce`。

在上一讲,我们继续梳理了PUT请求到EtcdServer这一层的逻辑,并大概阅读了其中的关键函数processInternalRaftRequestOnce

这个方法里面有不少细节,我们今天就选择其中有价值的四点来看看。

1. entry索引 - appliedIndex与committedIndex

在etcd中,我们将每个客户端的操作(如PUT)抽象为一个日志项(entry)。如果这个操作生效,etcd就将这个entry项同步给其它etcd server,作为数据同步。

操作有顺序之分,于是服务端就保存了一个长entry数组,用一个关键的索引index来进行区分entry数组(即一个分界的标志),对entry状态进行分类:

  • entry处于状态A - 小于等于索引的entry项
  • entry处于状态B - 大于索引的entry项

一般状态A和B都是互补的,即是一种二分类状态。

而由于分布式的特性,entry不能立刻完成执行的,于是这里就区分出了两种状态,它们复用一个entry数组:

  • 已应用 - applied
  • 已提交 - committed

对应索引appliedIndexcommittedIndex

// 函数用atomic保证原子性
ai := s.getAppliedIndex()
ci := s.getCommittedIndex()
// 两者的差值,表示已应用但是未提交的entry数,不能太多
if ci > ai+maxGapBetweenApplyAndCommitIndex {
   
  return nil, ErrTooManyRequests
}

entry数组中的索引的一致性非常重要,尤其是在并发的场景下。而示例中的原子操作,其实是一种乐观锁的实现。

更多的细节就涉及到分布式相关了,这里就不展开。

2.id生成器 - idutil.Generator

Generator数据结构不复杂,它的设计详情都放在了备注里,我们可以自行阅读:

// Generator generates unique identifiers based on counters, timestamps, and
// a node member ID.
//
// The initial id is in this format:
// High order 2 bytes are from memberID, next 5 bytes are from timestamp,
// and low order one byte is a counter.
// | prefix   | suffix              |
// | 2 bytes  | 5 bytes   | 1 byte  |
// | memberID | timestamp | cnt     |

在很多分布式系统中,都需要有一套唯一id生成器。etcd的这个方案相对简单,就是 成员id+时间戳 的组合方案。

关于分布式唯一id,更全面的设计可以参考Snowflake,如 https://segmentfault.com/a/1190000020899379

3.认证模块 - auth.AuthStore

authInfo, err := s.AuthInfoFromCtx(ctx)

认证功能在成熟软件中非常常见。在etcd,被独立到了etcd/auth模块中。这个模块的内部调用不复杂,功能就是从context中提取出 用户名+版本信息

这个提取过程中值得注意的是,AuthStore是从grpcmetadata提取出想要的认证信息,而metadata类似于HTTP1协议中的header,是一种用KV形式保存和提取数据的结构。

串联一下我们之前的思路,etcd通过grpc-gateway将HTTP1转化成了gRPC,那么就有一个 HTTP header到grpc metadata的映射过程,有兴趣的可以去研究一下。

总体来说,etcd的认证模块做得很简单,也方便其接入service-mesh。

4.多协程小工具 - wait.Wait

wait.Wait是一个很精巧的小工具,使用起来非常简单:

// 示例代码
ch := s.w.Register(id)
s.w.Trigger(id, nil)

我们可以在etcd/pkg/wait目录下看到它的具体实现,我提取了重点

// 通过id,来等待和触发对应的事件。
// 注意使用的顺序:先等待,再触发。
type Wait interface {
   
  // 等待,即注册一个id
    Register(id uint64) <-chan interface{
   }
    // 触发,用一个id
    Trigger(id uint64, x interface{
   })
    IsRegistered(id uint64) bool
}

// 实现:读写锁+map数据结构
type list struct {
   
    l sync.RWMutex
    m map[uint64]chan interface{
   }
}

// 注册一个id
func (w *list) Register(id uint64) <-chan interface{
   } {
   
    w.l.Lock()
    defer w.l.Unlock()
    ch := w.m[id]
    if ch == nil {
   
    // go官方建议带buffer的channel尽量设置大小为1
        ch = make(chan interface{
   }, 1)
        w.m[id] = ch
    } else {
   
    // 不允许重复
        log.Panicf("dup id %x", id)
    }
    return ch
}

// 触发id的channel
func (w *list) Trigger(id uint64, x interface{
   }) {
   
    w.l.Lock()
    ch := w.m[id]
    delete(w.m, id)
  // 取出ch后直接Unlock(可以思考一下与defer的区别)
    w.l.Unlock()
  // 如果触发的id不存在map里,就直接跳过这个判断
    if ch != nil {
   
        ch <- x
        close(ch)
    }
}

了解Wait的实现之后,我们就知道在正常情况下,RegisterTrigger必须一一对应。

但是,我们再往下看processInternalRaftRequestOnce这部分代码,发现了一个异常点:

select {
   
  // 异常:没有找到Trigger,难道忘了?
    case x := <-ch:
        return x.(*applyResult), nil
  // 正常:用Trigger退出
    case <-cctx.Done():
        proposalsFailed.Inc()
        s.w.Trigger(id, nil) 
        return nil, s.parseProposeCtxErr(cctx.Err(), start)
  // 正常:整个server停止,此时不用关心单个Trigger了
    case <-s.done:
        return nil, ErrStopped
}

这里,我们可以做个简单的猜测:在另一个goroutine中,这个etcd server进行了一个操作,包括下面两步:

  1. ch这个channel里发送了一个*applyResult结构的消息
  2. 对wait进行了Trigger操作

小结

今天我们进一步阅读了processInternalRaftRequestOnce中的四个细节,加强了etcd server对请求处理的印象。

etcd作为一款优秀的开源项目,其模块设计比较精巧,而阅读源码的同学也要掌握一个技巧:适当控制阅读深度。比如,在阅读PUT请求时,第一阶段阅读到EtcdServerprocessInternalRaftRequestOnce这层即可:

  • 如果继续深入看raftNode等实现,很容易导致你的整体思路变成过程性的调用,学习不成体系
  • 这时,回过头来巩固一下当前学习的部分,通过串联细节来加深印象,会对你梳理整体更有帮助
目录
相关文章
|
XML JSON Go
etcd源码分析 - 3.【打通核心流程】PUT键值对的执行链路
在上一讲,我们一起看了etcd server是怎么匹配到对应的处理函数的,如果忘记了请回顾一下。 今天,我们再进一步,看看`PUT`操作接下来是怎么执行的。
94 0
|
7月前
|
Kubernetes 开发工具 Docker
微服务实践k8s与dapr开发部署实验(2)状态管理
微服务实践k8s与dapr开发部署实验(2)状态管理
113 3
微服务实践k8s与dapr开发部署实验(2)状态管理
|
1月前
|
数据管理 Nacos 开发者
"Nacos架构深度解析:一篇文章带你掌握业务层四大核心功能,服务注册、配置管理、元数据与健康检查一网打尽!"
【10月更文挑战第23天】Nacos 是一个用于服务注册发现和配置管理的平台,支持动态服务发现、配置管理、元数据管理和健康检查。其业务层包括服务注册与发现、配置管理、元数据管理和健康检查四大核心功能。通过示例代码展示了如何在业务层中使用Nacos,帮助开发者构建高可用、动态扩展的微服务生态系统。
107 0
|
7月前
|
消息中间件 Java 调度
【深度挖掘RocketMQ底层源码】「底层源码挖掘系列」透彻剖析贯穿RocketMQ的消费者端的运行调度的流程(Pull模式)
【深度挖掘RocketMQ底层源码】「底层源码挖掘系列」透彻剖析贯穿RocketMQ的消费者端的运行调度的流程(Pull模式)
68 1
|
7月前
|
消息中间件 存储 NoSQL
【深度挖掘 RocketMQ底层源码】「底层源码挖掘系列」透彻剖析贯穿RocketMQ的消费者端的运行核心的流程(Pull模式-上)
【深度挖掘 RocketMQ底层源码】「底层源码挖掘系列」透彻剖析贯穿RocketMQ的消费者端的运行核心的流程(Pull模式-上)
62 1
|
7月前
|
消息中间件 Java RocketMQ
【深度挖掘 RocketMQ底层源码】「底层源码挖掘系列」抽丝剥茧贯穿RocketMQ的消费者端的运行核心的流程(Pull模式-下)
【深度挖掘 RocketMQ底层源码】「底层源码挖掘系列」抽丝剥茧贯穿RocketMQ的消费者端的运行核心的流程(Pull模式-下)
50 1
|
7月前
|
负载均衡 Dubbo Java
最简最快了解RPC核心流程
本文主要以最简易最快速的方式介绍RPC调用核心流程,文中以Dubbo为例。同时,会写一个简易的RPC调用代码,方便理解和记忆核心组件和核心流程。
最简最快了解RPC核心流程
|
7月前
|
监控 安全 数据处理
了解阿里云 RPA:如何实现流程自动化
机器人流程自动化(RPA)是一种快速发展的技术,它可以帮助企业实现重复性任务的自动化,提高工作效率和准确性。阿里云 RPA 作为一款强大的 RPA 解决方案,为用户提供了一种简单而高效的方式来实现流程自动化。本文将介绍阿里云 RPA 的功能和特点,以及如何使用它来实现流程自动化。
|
7月前
|
存储 监控 安全
插件机制详解:原理、设计与最佳实践
插件机制详解:原理、设计与最佳实践
374 0
|
7月前
|
前端开发 网络协议 Java
Netty | 工作流程图分析 & 核心组件说明 & 代码案例实践
Netty | 工作流程图分析 & 核心组件说明 & 代码案例实践
410 0