TLS-Golang实现

转载:

前言

在进行项目总结的时候,领导提出有关数据安全的问题。总结会议过后,自己查阅了一下资料,发现基于CA的TLS证书认证方案是一个很好的选择,虽然项目本身也有关于数据安全的处理,但是从远不及TLS的处理方式。

本文只介绍tls的开发,采用go语言,不会涉及到太多专业的词语。


制作自签名证书

初始目录如下:





CA

为了保证证书的可靠性和有效性,在这里可引入 CA 颁发的根证书的概念。CA就是专门用自己的私钥给别人进行签名的单位或者机构,其遵守 X.509 标准,即无论是客户端还是服务端都是使用CA来签发证书。

根证书

根证书(root certificate)是属于根证书颁发机构(CA)的公钥证书。我们可以通过验证 CA 的签名从而信任 CA ,任何人都可以得到 CA 的证书(含公钥),用以验证它所签发的证书(客户端、服务端)。

它包含了公钥和密钥。

CA公钥

进入cert目录

openssl genrsa -out ca.key 2048

  • openssl genrsa:生成RSA私钥,命令的最后一个参数,将指定生成密钥的位数,如果没有指定,默认512

CA秘钥(证书)

openssl req -new -x509 -days 365 -key ca.key -out ca.pem

  • -x509:证书文件格式为x509,目前TLS默认只支持这种格式的证书
  • -days 365:证书有效期1年
  • -out ca.pem:生成的私钥保存到ca.pem

要填写的信息:

Country Name (2 letter code) [XX]: State or Province Name (full name) []: Locality Name (eg, city) [Default City]: Organization Name (eg, company) [Default Company Ltd]: Organizational Unit Name (eg, section) []: Common Name (eg, your name or your server's hostname) []:ca.com Email Address []:
  • 生成的过程中会要求填一些信息,除了Common Name要取一个容易区分的名字之外,其它都可以随便填写,我们在这里将它填为ca.com.

目录





服务器证书相关

服务器key

# openssl genrsa -out server.key 2048
或者
openssl ecparam -genkey -name secp384r1 -out server.key
  • openssl genrsa:生成RSA私钥,命令的最后一个参数,将指定生成密钥的位数,如果没有指定,默认512
  • openssl ecparam:生成ECC私钥,命令为椭圆曲线密钥参数生成及操作,本文中ECC曲线选择的是secp384r1

生成 CSR(证书申请文件)

CSR 是Cerificate Signing Request 的英文缩写,为证书申请文件,在服务器私钥的基础上加上一些申请人的属性信息,比如我是谁,来自哪里,名字叫什么,证书适用于什么场景等的信息,然后带上进行的签名,发给CA(私下安全的方式发送),带上自己签名的目的是为了防止别人篡改文件。

openssl req -new -key server.key -out server.csr

要填写的信息:

Country Name (2 letter code) [AU]: State or Province Name (full name) [Some-State]: Locality Name (eg, city) []: Organization Name (eg, company) [Internet Widgits Pty Ltd]: Organizational Unit Name (eg, section) []: Common Name (e.g. server FQDN or YOUR name) []:domain.com Email Address []: Please enter the following 'extra' attributes to be sent with your certificate request A challenge password []:123456 An optional company name []:
  • 生成的过程中会要求填一些信息,除了Common Name要取一个容易区分的名字之外,其它都可以随便填写,我们在这里将它填为domain.com;
  • 密码也是建议填写;
  • 注意和ca证书的不同。

基于 CA 签发

使用CA的公钥对申请文件进行签名

openssl x509 -req -sha256 -CA ca.pem -CAkey ca.key -CAcreateserial -days 3650 -in server.csr -out server.pem
  • -sha256:生成的证书里面使用sha256作为摘要算法
  • 由于需要往生成的证书里写入签名者的信息,所以这里需要ca.pem,因为只有这里有CA的描述信息,ca.key里面只有公钥的信息。

目录

grpc-tls/ ├── configs │   ├── cert # 存放证书相关的目录 │       ├── ca.key │       └── ca.pem │       ├── server.csr │       └── server.key │       └── server.pem ├── cmd

客户端证书相关

此生成的证书可用于浏览器、java、tomcat、golang等。

客户端key

openssl ecparam -genkey -name secp384r1 -out client.key

生成CSR(证书申请书)

openssl req -new -key client.key -out client.csr

要填写的信息:

  • Common Name要取一个容易区分的名字之外,填为domain.com,和服务器的CSR保持一致;
  • 密码也是建议填写;
  • 注意和ca证书的不同。

基于CA签发

openssl x509 -req -sha256 -CA ca.pem -CAkey ca.key -CAcreateserial -days 3650 -in client.csr -out client.pem

生成客户端p12格式根证书

该证书用于导入浏览器使用。

openssl pkcs12 -export -clcerts -in client.pem -inkey client.key -out client.p12

目录

grpc-tls/
├── configs
│   ├── cert # 存放证书相关的目录
│       ├── ca.key
│       └── ca.pem # 导入浏览器使用
│       ├── server.csr
│       └── server.key
│       └── server.pem
│       ├── client.csr
│       └── client.key
│       └── client.pem
│       └── client.p12 # 导入浏览器使用
├── cmd

证书如何验证

下面以浏览器为例,说明证书的验证过程:

  • 在TLS握手的过程中,浏览器得到了网站的证书(.p12)
  • 打开证书,查看是哪个CA签名的这个证书(.p12)
  • 在自己信任的CA库中,找相应CA的证书(ca.pem),
  • 用CA证书里面的公钥解密网站证书上的签名,取出网站证书的校验码(指纹),然后用CA证书中摘要算法(比如sha256)算出出网站证书的校验码,如果校验码和签名中的校验码对的上,说明这个证书是合法的,且没被人篡改过
  • 读出里面的CN,对于网站的证书,里面一般包含的是域名
  • 检查里面的域名和自己访问网站的域名对不对的上,对的上的话,就说明这个证书确实是颁发给这个网站的
  • 到此为止检查通过

如果浏览器发现证书有问题,一般是证书里面的签名者不是浏览器认为值得信任的CA,浏览器就会给出警告页面,这时候需要谨慎,有可能证书被掉包了。如访问12306网站,由于12306的证书是自己签的名,并且浏览器不认为12306是受信的CA,所以就会给警告,但是一旦你把12306的根证书安装到了你的浏览器中,那么下次就不会警告了,因为你配置了浏览器让它相信12306是一个受信的CA。


在浏览器中导入证书

导入证书

  • 详细步骤百度即可….
  • 个人导入client.p12证书
  • 受信任的根证书颁发机构导入ca.pem证书

修改域名

ca.pem这个证书是发给domain.com的,而不是127.0.0.1,所以在C:\Windows\System32\drivers\etc\hosts添加一记录:

测试完成之后记得手动将127.0.0.1 domain.comC:\Windows\System32\drivers\etc\hosts里面删掉。


golang服务端

目录

grpc-tls/
├── configs
│   ├── cert # 存放证书相关的目录
│       ├── ca.key
│       └── ca.pem # 导入浏览器使用
│       ├── server.csr
│       └── server.key
│       └── server.pem
│       ├── client.csr
│       └── client.key
│       └── client.pem
│       └── client.p12 # 导入浏览器使用
├── cmd
│   ├── main.go

main.go

package main

import (
    "crypto/tls"
    "crypto/x509"
    "github.com/gin-gonic/gin"
    "io/ioutil"
    "log"
    "net/http"
    "time"
)

func main() {
    router := gin.Default()
    router.GET("/test", func(c *gin.Context) {
        c.JSON(200, gin.H{
            "message": "test",
        })
    })

    // 启动https方式访问
    cert, err := tls.LoadX509KeyPair("./configs/cert/server.pem", "./configs/cert/server.key")
    if err != nil {
        log.Fatalf("tls.LoadX509KeyPair err: %v", err)
    }
    certPool := x509.NewCertPool()
    ca, err := ioutil.ReadFile("./configs/cert/ca.pem")
    if err != nil {
        log.Fatalf("ioutil.ReadFile err: %v", err)
    }
    if ok := certPool.AppendCertsFromPEM(ca); !ok {
        log.Fatalf("certPool.AppendCertsFromPEM err")
    }
    ReadTimeout := time.Duration(60) * time.Second
    WriteTimeout := time.Duration(60) * time.Second

    s := &http.Server{
        Addr:          ":8090",
        Handler:        router,
        ReadTimeout:    ReadTimeout,
        WriteTimeout:   WriteTimeout,
        MaxHeaderBytes: 1 << 20,
        TLSConfig:&tls.Config{
            Certificates: []tls.Certificate{cert},
            MinVersion: tls.VersionTLS12,
            ClientAuth:   tls.RequireAndVerifyClientCert,
            ClientCAs:    certPool,
        },
    }

    s.ListenAndServeTLS("./configs/cert/server.pem","./configs/cert/server.key")
}

测试

在浏览器输入https://domain.com:8090/test

https://img2018.cnblogs.com/blog/1481607/201907/1481607-20190718144205177-1562071036.png

参考

SSL/TLS及证书概述

带入gRPC:TLS 证书认证

Golang并发编程与同步原语

原文地址:https://draveness.me/golang-sync-primitives.html

当提到并发编程、多线程编程时,我们往往都离不开『锁』这一概念,Go 语言作为一个原生支持用户态进程 Goroutine 的语言,也一定会为开发者提供这一功能,锁的主要作用就是保证多个线程或者 Goroutine 在访问同一片内存时不会出现混乱的问题,锁其实是一种并发编程中的同步原语(Synchronization Primitives)。

在这一节中我们就会介绍 Go 语言中常见的同步原语 MutexRWMutexWaitGroupOnce 和 Cond 以及扩展原语 ErrGroupSemaphore和 SingleFlight 的实现原理,同时也会涉及互斥锁、信号量等并发编程中的常见概念。

基本原语

Go 语言在 sync 包中提供了用于同步的一些基本原语,包括常见的互斥锁 Mutex 与读写互斥锁 RWMutex 以及 OnceWaitGroup

golang-basic-sync-primitives

这些基本原语的主要作用是提供较为基础的同步功能,我们应该使用 Channel 和通信来实现更加高级的同步机制,我们在这一节中并不会介绍标准库中全部的原语,而是会介绍其中比较常见的 MutexRWMutexOnceWaitGroup 和 Cond,我们并不会涉及剩下两个用于存取数据的结构体 Map 和 Pool

Mutex

Go 语言中的互斥锁在 sync 包中,它由两个字段 state 和 sema 组成,state 表示当前互斥锁的状态,而 sema 真正用于控制锁状态的信号量,这两个加起来只占 8 个字节空间的结构体就表示了 Go 语言中的互斥锁。

type Mutex struct {
	state int32
	sema  uint32
}

状态

互斥锁的状态是用 int32 来表示的,但是锁的状态并不是互斥的,它的最低三位分别表示 mutexLockedmutexWoken 和 mutexStarving,剩下的位置都用来表示当前有多少个 Goroutine 等待互斥锁被释放:

golang-mutex-state

互斥锁在被创建出来时,所有的状态位的默认值都是 0,当互斥锁被锁定时 mutexLocked 就会被置成 1、当互斥锁被在正常模式下被唤醒时 mutexWoken 就会被被置成 1mutexStarving 用于表示当前的互斥锁进入了状态,最后的几位是在当前互斥锁上等待的 Goroutine 个数。

饥饿模式

在了解具体的加锁和解锁过程之前,我们需要先简单了解一下 Mutex 在使用过程中可能会进入的饥饿模式,饥饿模式是在 Go 语言 1.9 版本引入的特性,它的主要功能就是保证互斥锁的获取的『公平性』(Fairness)。

互斥锁可以同时处于两种不同的模式,也就是正常模式和饥饿模式,在正常模式下,所有锁的等待者都会按照先进先出的顺序获取锁,但是如果一个刚刚被唤起的 Goroutine 遇到了新的 Goroutine 进程也调用了 Lock 方法时,大概率会获取不到锁,为了减少这种情况的出现,防止 Goroutine 被『饿死』,一旦 Goroutine 超过 1ms 没有获取到锁,它就会将当前互斥锁切换饥饿模式。

golang-mutex-mode

在饥饿模式中,互斥锁会被直接交给等待队列最前面的 Goroutine,新的 Goroutine 在这时不能获取锁、也不会进入自旋的状态,它们只会在队列的末尾等待,如果一个 Goroutine 获得了互斥锁并且它是队列中最末尾的协程或者它等待的时间少于 1ms,那么当前的互斥锁就会被切换回正常模式。

相比于饥饿模式,正常模式下的互斥锁能够提供更好地性能,饥饿模式的主要作用就是避免一些 Goroutine 由于陷入等待无法获取锁而造成较高的尾延时,这也是对 Mutex 的一个优化。

加锁

互斥锁 Mutex 的加锁是靠 Lock 方法完成的,最新的 Go 语言源代码中已经将 Lock 方法进行了简化,方法的主干只保留了最常见、简单并且快速的情况;当锁的状态是 0 时直接将 mutexLocked 位置成 1

func (m *Mutex) Lock() {
	if atomic.CompareAndSwapInt32(&m.state, 0, mutexLocked) {
		return
	}
	m.lockSlow()
}

但是当 Lock 方法被调用时 Mutex 的状态不是 0 时就会进入 lockSlow 方法尝试通过自旋或者其他的方法等待锁的释放并获取互斥锁,该方法的主体是一个非常大 for 循环,我们会将该方法分成几个部分介绍获取锁的过程:

func (m *Mutex) lockSlow() {
	var waitStartTime int64
	starving := false
	awoke := false
	iter := 0
	old := m.state
	for {
		if old&(mutexLocked|mutexStarving) == mutexLocked && runtime_canSpin(iter) {
			if !awoke && old&mutexWoken == 0 && old>>mutexWaiterShift != 0 &&
				atomic.CompareAndSwapInt32(&m.state, old, old|mutexWoken) {
				awoke = true
			}
			runtime_doSpin()
			iter++
			old = m.state
			continue
		}	

在这段方法的第一部分会判断当前方法能否进入自旋来等待锁的释放,自旋(Spinnig)其实是在多线程同步的过程中使用的一种机制,当前的进程在进入自旋的过程中会一直保持 CPU 的占用,持续检查某个条件是否为真,在多核的 CPU 上,自旋的优点是避免了 Goroutine 的切换,所以如果使用恰当会对性能带来非常大的增益。

在 Go 语言的 Mutex 互斥锁中,只有在普通模式下才可能进入自旋,除了模式的限制之外,runtime_canSpin方法中会判断当前方法是否可以进入自旋,进入自旋的条件非常苛刻:

  1. 运行在多 CPU 的机器上;
  2. 当前 Goroutine 为了获取该锁进入自旋的次数小于四次;
  3. 当前机器上至少存在一个正在运行的处理器 P 并且处理的运行队列是空的;

一旦当前 Goroutine 能够进入自旋就会调用 runtime_doSpin,它最终调用汇编语言编写的方法 procyield 并执行指定次数的 PAUSE 指令,PAUSE 指令什么都不会做,但是会消耗 CPU 时间,每次自旋都会调用 30 次 PAUSE,下面是该方法在 386 架构的机器上的实现:

TEXT runtime·procyield(SB),NOSPLIT,$0-0
	MOVL	cycles+0(FP), AX
again:
	PAUSE
	SUBL	$1, AX
	JNZ	again
	RET

处理了自旋相关的特殊逻辑之后,互斥锁接下来就根据上下文计算当前互斥锁最新的状态了,几个不同的条件分别会更新 state 中存储的不同信息 mutexLockedmutexStarvingmutexWoken 和 mutexWaiterShift

		new := old
		if old&mutexStarving == 0 {
			new |= mutexLocked
		}
		if old&(mutexLocked|mutexStarving) != 0 {
			new += 1 << mutexWaiterShift
		}
		if starving && old&mutexLocked != 0 {
			new |= mutexStarving
		}
		if awoke {
			new &^= mutexWoken
		}

计算了新的互斥锁状态之后,我们就会使用 atomic 包提供的 CAS 函数修改互斥锁的状态,如果当前的互斥锁已经处于饥饿和锁定的状态,就会跳过当前步骤,调用 runtime_SemacquireMutex 方法:

		if atomic.CompareAndSwapInt32(&m.state, old, new) {
			if old&(mutexLocked|mutexStarving) == 0 {
				break // locked the mutex with CAS
			}
			queueLifo := waitStartTime != 0
			if waitStartTime == 0 {
				waitStartTime = runtime_nanotime()
			}
			runtime_SemacquireMutex(&m.sema, queueLifo, 1)
			starving = starving || runtime_nanotime()-waitStartTime > starvationThresholdNs
			old = m.state
			if old&mutexStarving != 0 {
				delta := int32(mutexLocked - 1<<mutexWaiterShift)
				if !starving || old>>mutexWaiterShift == 1 {
					delta -= mutexStarving
				}
				atomic.AddInt32(&m.state, delta)
				break
			}
			awoke = true
			iter = 0
		} else {
			old = m.state
		}
	}
}

runtime_SemacquireMutex 方法的主要作用就是通过 Mutex 的使用互斥锁中的信号量保证资源不会被两个 Goroutine 获取,从这里我们就能看出 Mutex 其实就是对更底层的信号量进行封装,对外提供更加易用的 API,runtime_SemacquireMutex 会在方法中不断调用 goparkunlock 将当前 Goroutine 陷入休眠等待信号量可以被获取。

一旦当前 Goroutine 可以获取信号量,就证明互斥锁已经被解锁,该方法就会立刻返回,Lock 方法的剩余代码也会继续执行下去了,当前互斥锁处于饥饿模式时,如果该 Goroutine 是队列中最后的一个 Goroutine 或者等待锁的时间小于 starvationThresholdNs(1ms),当前 Goroutine 就会直接获得互斥锁并且从饥饿模式中退出并获得锁。

解锁

互斥锁的解锁过程相比之下就非常简单,Unlock 方法会直接使用 atomic 包提供的 AddInt32,如果返回的新状态不等于 0 就会进入 unlockSlow 方法:

func (m *Mutex) Unlock() {
	new := atomic.AddInt32(&m.state, -mutexLocked)
	if new != 0 {
		m.unlockSlow(new)
	}
}

unlockSlow 方法首先会对锁的状态进行校验,如果当前互斥锁已经被解锁过了就会直接抛出异常 sync: unlock of unlocked mutex 中止当前程序,在正常情况下会根据当前互斥锁的状态是正常模式还是饥饿模式进入不同的分支:

func (m *Mutex) unlockSlow(new int32) {
	if (new+mutexLocked)&mutexLocked == 0 {
		throw("sync: unlock of unlocked mutex")
	}
	if new&mutexStarving == 0 {
		old := new
		for {
			if old>>mutexWaiterShift == 0 || old&(mutexLocked|mutexWoken|mutexStarving) != 0 {
				return
			}
			new = (old - 1<<mutexWaiterShift) | mutexWoken
			if atomic.CompareAndSwapInt32(&m.state, old, new) {
				runtime_Semrelease(&m.sema, false, 1)
				return
			}
			old = m.state
		}
	} else {
		runtime_Semrelease(&m.sema, true, 1)
	}
}

如果当前互斥锁的状态是饥饿模式就会直接调用 runtime_Semrelease 方法直接将当前锁交给下一个正在尝试获取锁的等待者,等待者会在被唤醒之后设置 mutexLocked 状态,由于此时仍然处于 mutexStarving,所以新的 Goroutine 也无法获得锁。

在正常模式下,如果当前互斥锁不存在等待者或者最低三位表示的状态都为 0,那么当前方法就不需要唤醒其他 Goroutine 可以直接返回,当有 Goroutine 正在处于等待状态时,还是会通过 runtime_Semrelease 唤醒对应的 Goroutine 并移交锁的所有权。

小结

通过对互斥锁 Mutex 加锁和解锁过程的分析,我们能够得出以下的一些结论,它们能够帮助我们更好地理解互斥锁的工作原理,互斥锁的加锁的过程比较复杂,涉及自旋、信号量以及 Goroutine 调度等概念:

  • 如果互斥锁处于初始化状态,就会直接通过置位 mutexLocked 加锁;
  • 如果互斥锁处于 mutexLocked 并且在普通模式下工作,就会进入自旋,执行 30 次 PAUSE 指令消耗 CPU 时间等待锁的释放;
  • 如果当前 Goroutine 等待锁的时间超过了 1ms,互斥锁就会被切换到饥饿模式;
  • 互斥锁在正常情况下会通过 runtime_SemacquireMutex 方法将调用 Lock 的 Goroutine 切换至休眠状态,等待持有信号量的 Goroutine 唤醒当前协程;
  • 如果当前 Goroutine 是互斥锁上的最后一个等待的协程或者等待的时间小于 1ms,当前 Goroutine 会将互斥锁切换回正常模式;

互斥锁的解锁过程相对来说就比较简单,虽然对于普通模式和饥饿模式的处理有一些不同,但是由于代码行数不多,所以逻辑清晰,也非常容易理解:

  • 如果互斥锁已经被解锁,那么调用 Unlock 会直接抛出异常;
  • 如果互斥锁处于饥饿模式,会直接将锁的所有权交给队列中的下一个等待者,等待者会负责设置 mutexLocked 标志位;
  • 如果互斥锁处于普通模式,并且没有 Goroutine 等待锁的释放或者已经有被唤醒的 Goroutine 获得了锁就会直接返回,在其他情况下回通过 runtime_Semrelease 唤醒对应的 Goroutine;

RWMutex

读写互斥锁也是 Go 语言 sync 包为我们提供的接口之一,一个常见的服务对资源的读写比例会非常高,如果大多数的请求都是读请求,它们之间不会相互影响,那么我们为什么不能将对资源读和写操作分离呢?这也就是 RWMutex 读写互斥锁解决的问题,不限制对资源的并发读,但是读写、写写操作无法并行执行。

 
YN
NN

读写互斥锁在 Go 语言中的实现是 RWMutex,其中不仅包含一个互斥锁,还持有两个信号量,分别用于写等待读和读等待写:

type RWMutex struct {
	w           Mutex
	writerSem   uint32
	readerSem   uint32
	readerCount int32
	readerWait  int32
}

readerCount 存储了当前正在执行的读操作的数量,最后的 readerWait 表示当写操作被阻塞时等待的读操作个数。

读锁

读锁的加锁非常简单,我们通过 atomic.AddInt32 方法为 readerCount 加一,如果该方法返回了负数说明当前有 Goroutine 获得了写锁,当前 Goroutine 就会调用 runtime_SemacquireMutex 陷入休眠等待唤醒:

func (rw *RWMutex) RLock() {
	if atomic.AddInt32(&rw.readerCount, 1) < 0 {
		runtime_SemacquireMutex(&rw.readerSem, false, 0)
	}
}

如果没有写操作获取当前互斥锁,当前方法就会在 readerCount 加一后返回;当 Goroutine 想要释放读锁时会调用 RUnlock 方法:

func (rw *RWMutex) RUnlock() {
	if r := atomic.AddInt32(&rw.readerCount, -1); r < 0 {
		rw.rUnlockSlow(r)
	}
}

该方法会在减少正在读资源的 readerCount,当前方法如果遇到了返回值小于零的情况,说明有一个正在进行的写操作,在这时就应该通过 rUnlockSlow 方法减少当前写操作等待的读操作数 readerWait 并在所有读操作都被释放之后触发写操作的信号量 writerSem

func (rw *RWMutex) rUnlockSlow(r int32) {
	if r+1 == 0 || r+1 == -rwmutexMaxReaders {
		throw("sync: RUnlock of unlocked RWMutex")
	}
	if atomic.AddInt32(&rw.readerWait, -1) == 0 {
		runtime_Semrelease(&rw.writerSem, false, 1)
	}
}

writerSem 在被触发之后,尝试获取读写锁的进程就会被唤醒并获得锁。

读写锁

当资源的使用者想要获取读写锁时,就需要通过 Lock 方法了,在 Lock 方法中首先调用了读写互斥锁持有的 Mutex 的 Lock 方法保证其他获取读写锁的 Goroutine 进入等待状态,随后的 atomic.AddInt32(&rw.readerCount, -rwmutexMaxReaders) 其实是为了阻塞后续的读操作:

func (rw *RWMutex) Lock() {
	rw.w.Lock()
	r := atomic.AddInt32(&rw.readerCount, -rwmutexMaxReaders) + rwmutexMaxReaders
	if r != 0 && atomic.AddInt32(&rw.readerWait, r) != 0 {
		runtime_SemacquireMutex(&rw.writerSem, false, 0)
	}
}

如果当前仍然有其他 Goroutine 持有互斥锁的读锁,该 Goroutine 就会调用 runtime_SemacquireMutex 进入休眠状态,等待读锁释放时触发 writerSem 信号量将当前协程唤醒。

对资源的读写操作完成之后就会将通过 atomic.AddInt32(&rw.readerCount, rwmutexMaxReaders) 变回正数并通过 for 循环触发所有由于获取读锁而陷入等待的 Goroutine:

func (rw *RWMutex) Unlock() {
	r := atomic.AddInt32(&rw.readerCount, rwmutexMaxReaders)
	if r >= rwmutexMaxReaders {
		throw("sync: Unlock of unlocked RWMutex")
	}
	for i := 0; i < int(r); i++ {
		runtime_Semrelease(&rw.readerSem, false, 0)
	}
	rw.w.Unlock()
}

在方法的最后,RWMutex 会释放持有的互斥锁让其他的协程能够重新获取读写锁。

小结

相比状态复杂的互斥锁 Mutex 来说,读写互斥锁 RWMutex 虽然提供的功能非常复杂,但是由于站在了 Mutex的『肩膀』上,所以整体的实现上会简单很多。

  1. readerSem — 读写锁释放时通知由于获取读锁等待的 Goroutine;
  2. writerSem — 读锁释放时通知由于获取读写锁等待的 Goroutine;
  3. w 互斥锁 — 保证写操作之间的互斥;
  4. readerCount — 统计当前进行读操作的协程数,触发写锁时会将其减少 rwmutexMaxReaders 阻塞后续的读操作;
  5. readerWait — 当前读写锁等待的进行读操作的协程数,在触发 Lock 之后的每次 RUnlock 都会将其减一,当它归零时该 Goroutine 就会获得读写锁;
  6. 当读写锁被释放 Unlock 时首先会通知所有的读操作,然后才会释放持有的互斥锁,这样能够保证读操作不会被连续的写操作『饿死』;

RWMutex 在 Mutex 之上提供了额外的读写分离功能,能够在读请求远远多于写请求时提供性能上的提升,我们也可以在场景合适时选择读写互斥锁。

WaitGroup

WaitGroup 是 Go 语言 sync 包中比较常见的同步机制,它可以用于等待一系列的 Goroutine 的返回,一个比较常见的使用场景是批量执行 RPC 或者调用外部服务:

requests := []*Request{...}

wg := &sync.WaitGroup{}
wg.Add(len(requests))

for _, request := range requests {
    go func(r *Request) {
        defer wg.Done()
        
        // res, err := service.call(r)
    }(request)
}

wg.Wait()

通过 WaitGroup 我们可以在多个 Goroutine 之间非常轻松地同步信息,原本顺序执行的代码也可以在多个 Goroutine 中并发执行,加快了程序处理的速度,在上述代码中只有在所有的 Goroutine 都执行完毕之后 Wait方法才会返回,程序可以继续执行其他的逻辑。

golang-syncgroup

总而言之,它的作用就像它的名字一样,,通过 Done 来传递任务完成的信号,比较常用于等待一组 Goroutine 中并发执行的任务全部结束。

结构体

WaitGroup 结构体中的成员变量非常简单,其中的 noCopy 的主要作用就是保证 WaitGroup 不会被开发者通过再赋值的方式进行拷贝,进而导致一些诡异的行为:

type WaitGroup struct {
	noCopy noCopy

	state1 [3]uint32
}

copylock 包就是一个用于检查类似错误的分析器,它的原理就是在 编译期间 检查被拷贝的变量中是否包含 noCopy 或者 sync 关键字,如果包含当前关键字就会报出以下的错误:

package main

import (
	"fmt"
	"sync"
)

func main() {
	wg := sync.Mutex{}
	yawg := wg
	fmt.Println(wg, yawg)
}

$ go run proc.go
./prog.go:10:10: assignment copies lock value to yawg: sync.Mutex
./prog.go:11:14: call of fmt.Println copies lock value: sync.Mutex
./prog.go:11:18: call of fmt.Println copies lock value: sync.Mutex

这段代码会在赋值和调用 fmt.Println 时发生值拷贝导致分析器报错,你可以通过访问 链接 尝试运行这段代码。

除了 noCopy 之外,WaitGroup 结构体中还包含一个总共占用 12 字节大小的数组,这个数组中会存储当前结构体持有的状态和信号量,在 64 位与 32 位的机器上表现也非常不同。

golang-waitgroup-state

WaitGroup 提供了私有方法 state 能够帮助我们从 state1 字段中取出它的状态和信号量。

操作

WaitGroup 对外暴露的接口只有三个 AddWait 和 Done,其中 Done 方法只是调用了 wg.Add(-1) 本身并没有什么特殊的逻辑,我们来了解一下剩余的两个方法:

func (wg *WaitGroup) Add(delta int) {
	statep, semap := wg.state()
	state := atomic.AddUint64(statep, uint64(delta)<<32)
	v := int32(state >> 32)
	w := uint32(state)
	if v < 0 {
		panic("sync: negative WaitGroup counter")
	}
	if v > 0 || w == 0 {
		return
	}
	*statep = 0
	for ; w != 0; w-- {
		runtime_Semrelease(semap, false, 0)
	}
}

Add 方法的主要作用就是更新 WaitGroup 中持有的计数器 counter,64 位状态的高 32 位,虽然 Add 方法传入的参数可以为负数,但是一个 WaitGroup 的计数器只能是非负数,当调用 Add 方法导致计数器归零并且还有等待的 Goroutine 时,就会通过 runtime_Semrelease 唤醒处于等待状态的所有 Goroutine。

另一个 WaitGroup 的方法 Wait 就会在当前计数器中保存的数据大于 0 时修改等待 Goroutine 的个数 waiter 并调用 runtime_Semacquire 陷入睡眠状态。

func (wg *WaitGroup) Wait() {
	statep, semap := wg.state()
	for {
		state := atomic.LoadUint64(statep)
		v := int32(state >> 32)
		if v == 0 {
			return
		}
		if atomic.CompareAndSwapUint64(statep, state, state+1) {
			runtime_Semacquire(semap)
			if +statep != 0 {
				panic("sync: WaitGroup is reused before previous Wait has returned")
			}
			return
		}
	}
}

陷入睡眠的 Goroutine 就会等待 Add 方法在计数器为 0 时唤醒。

小结

通过对 WaitGroup 的分析和研究,我们能够得出以下的一些结论:

  • Add 不能在和 Wait 方法在 Goroutine 中并发调用,一旦出现就会造成程序崩溃;
  • WaitGroup 必须在 Wait 方法返回之后才能被重新使用;
  • Done 只是对 Add 方法的简单封装,我们可以向 Add 方法传入任意负数(需要保证计数器非负)快速将计数器归零以唤醒其他等待的 Goroutine;
  • 可以同时有多个 Goroutine 等待当前 WaitGroup 计数器的归零,这些 Goroutine 也会被『同时』唤醒;

Once

Go 语言在标准库的 sync 同步包中还提供了 Once 语义,它的主要功能其实也很好理解,保证在 Go 程序运行期间 Once 对应的某段代码只会执行一次。

在如下所示的代码中,Do 方法中传入的函数只会被执行一次,也就是我们在运行如下所示的代码时只会看见一次 only once 的输出结果:

func main() {
    o := &sync.Once{}
    for i := 0; i < 10; i++ {
        o.Do(func() {
            fmt.Println("only once")
        })
    }
}

$ go run main.go
only once

作为 sync 包中的结构体,Once 有着非常简单的数据结构,每一个 Once 结构体中都只包含一个用于标识代码块是否被执行过的 done 以及一个互斥锁 Mutex

type Once struct {
	done uint32
	m    Mutex
}

Once 结构体对外唯一暴露的方法就是 Do,该方法会接受一个入参为空的函数,如果使用 atomic.LoadUint32检查到已经执行过函数了,就会直接返回,否则就会进入 doSlow 运行传入的函数:

func (o *Once) Do(f func()) {
	if atomic.LoadUint32(&o.done) == 0 {
		o.doSlow(f)
	}
}

func (o *Once) doSlow(f func()) {
	o.m.Lock()
	defer o.m.Unlock()
	if o.done == 0 {
		defer atomic.StoreUint32(&o.done, 1)
		f()
	}
}

doSlow 的实现也非常简单,我们先为当前的 Goroutine 获取互斥锁,然后通过 defer 关键字将 done 成员变量设置成 1 并运行传入的函数,无论当前函数是正常运行还是抛出 panic,当前方法都会将 done 设置成 1 保证函数不会执行第二次。

小结

作为用于保证函数执行次数的 Once 结构体,它使用互斥锁和 atomic 提供的方法实现了某个函数在程序运行期间只能执行一次的语义,在使用的过程中我们也需要注意以下的内容:

  • Do 方法中传入的函数只会被执行一次,哪怕函数中发生了 panic
  • 两次调用 Do 方法传入不同的函数时只会执行第一次调用的函数;

Cond

Go 语言在标准库中提供的 Cond 其实是一个条件变量,通过 Cond 我们可以让一系列的 Goroutine 都在触发某个事件或者条件时才被唤醒,每一个 Cond 结构体都包含一个互斥锁 L,我们先来看一下 Cond 是如何使用的:

func main() {
	c := sync.NewCond(&sync.Mutex{})

	for i := 0; i < 10; i++ {
		go listen(c)
	}

	go broadcast(c)

	ch := make(chan os.Signal, 1)
	signal.Notify(ch, os.Interrupt)
	<-ch
}

func broadcast(c *sync.Cond) {
	c.L.Lock()
	c.Broadcast()
	c.L.Unlock()
}

func listen(c *sync.Cond) {
	c.L.Lock()
	c.Wait()
	fmt.Println("listen")
	c.L.Unlock()
}

$ go run main.go
listen
listen
...
listen

在上述代码中我们同时运行了 11 个 Goroutine,其中的 10 个 Goroutine 会通过 Wait 等待期望的信号或者事件,而剩下的一个 Goroutine 会调用 Broadcast 方法通知所有陷入等待的 Goroutine,当调用 Boardcast方法之后,就会打印出 10 次 "listen" 并结束调用。

golang-cond-broadcast

结构体

Cond 的结构体中包含 noCopy 和 copyChecker 两个字段,前者用于保证 Cond 不会再编译期间拷贝,后者保证在运行期间发生拷贝会直接 panic,持有的另一个锁 L 其实是一个接口 Locker,任意实现 Lock 和 Unlock方法的结构体都可以作为 NewCond 方法的参数:

type Cond struct {
	noCopy noCopy

	L Locker

	notify  notifyList
	checker copyChecker
}

结构体中最后的变量 notifyList 其实也就是为了实现 Cond 同步机制,该结构体其实就是一个 Goroutine 的链表:

type notifyList struct {
	wait uint32
	notify uint32

	lock mutex
	head *sudog
	tail *sudog
}

在这个结构体中,head 和 tail 分别指向的就是整个链表的头和尾,而 wait 和 notify 分别表示当前正在等待的 Goroutine 和已经通知到的 Goroutine,我们通过这两个变量就能确认当前待通知和已通知的 Goroutine。

操作

Cond 对外暴露的 Wait 方法会将当前 Goroutine 陷入休眠状态,它会先调用 runtime_notifyListAdd 将等待计数器 +1,然后解锁并调用 runtime_notifyListWait 等待其他 Goroutine 的唤醒:

func (c *Cond) Wait() {
	c.checker.check()
	t := runtime_notifyListAdd(&c.notify)
	c.L.Unlock()
	runtime_notifyListWait(&c.notify, t)
	c.L.Lock()
}

func notifyListAdd(l *notifyList) uint32 {
	return atomic.Xadd(&l.wait, 1) - 1
}

notifyListWait 方法的主要作用就是获取当前的 Goroutine 并将它追加到 notifyList 链表的最末端:


func notifyListWait(l *notifyList, t uint32) {
	lock(&l.lock)

	if less(t, l.notify) {
		unlock(&l.lock)
		return
	}

	s := acquireSudog()
	s.g = getg()
	s.ticket = t
	if l.tail == nil {
		l.head = s
	} else {
		l.tail.next = s
	}
	l.tail = s
	goparkunlock(&l.lock, waitReasonSyncCondWait, traceEvGoBlockCond, 3)
	releaseSudog(s)
}

除了将当前 Goroutine 追加到链表的末端之外,我们还会调用 goparkunlock 陷入休眠状态,该函数也是在 Go 语言切换 Goroutine 时经常会使用的方法,它会直接让出当前处理器的使用权并等待调度器的唤醒。

golang-cond-notifylist

Cond 对外提供的 Signal 和 Broadcast 方法就是用来唤醒调用 Wait 陷入休眠的 Goroutine,从两个方法的名字来看,前者会唤醒队列最前面的 Goroutine,后者会唤醒队列中全部的 Goroutine:

func (c *Cond) Signal() {
	c.checker.check()
	runtime_notifyListNotifyOne(&c.notify)
}

func (c *Cond) Broadcast() {
	c.checker.check()
	runtime_notifyListNotifyAll(&c.notify)
}

notifyListNotifyAll 方法会从链表中取出全部的 Goroutine 并为它们依次调用 readyWithTime,该方法会通过 goready 将目标的 Goroutine 唤醒:

func notifyListNotifyAll(l *notifyList) {
	s := l.head
	l.head = nil
	l.tail = nil

	atomic.Store(&l.notify, atomic.Load(&l.wait))

	for s != nil {
		next := s.next
		s.next = nil
		readyWithTime(s, 4)
		s = next
	}
}

虽然它会依次唤醒全部的 Goroutine,但是这里唤醒的顺序其实也是按照加入队列的先后顺序,先加入的会先被 goready 唤醒,后加入的 Goroutine 可能就需要等待调度器的调度。

而 notifyListNotifyOne 函数就只会从 sudog 构成的链表中满足 sudog.ticket == l.notify 的 Goroutine 并通过 readyWithTime 唤醒:

func notifyListNotifyOne(l *notifyList) {
	t := l.notify
	atomic.Store(&l.notify, t+1)

	for p, s := (*sudog)(nil), l.head; s != nil; p, s = s, s.next {
		if s.ticket == t {
			n := s.next
			if p != nil {
				p.next = n
			} else {
				l.head = n
			}
			if n == nil {
				l.tail = p
			}
			s.next = nil
			readyWithTime(s, 4)
			return
		}
	}
}

在一般情况下我们都会选择在不满足特定条件时调用 Wait 陷入休眠,当某些 Goroutine 检测到当前满足了唤醒的条件,就可以选择使用 Signal 通知一个或者 Broadcast 通知全部的 Goroutine 当前条件已经满足,可以继续完成工作了。

小结

与 Mutex 相比,Cond 还是一个不被所有人都清楚和理解的同步机制,它提供了类似队列的 FIFO 的等待机制,同时也提供了 Signal 和 Broadcast 两种不同的唤醒方法,相比于使用 for {} 忙碌等待,使用 Cond 能够在遇到长时间条件无法满足时将当前处理器让出的功能,如果我们合理使用还是能够在一些情况下提升性能,在使用的过程中我们需要注意:

  • Wait 方法在调用之前一定要使用 L.Lock 持有该资源,否则会发生 panic 导致程序崩溃;
  • Signal 方法唤醒的 Goroutine 都是队列最前面、等待最久的 Goroutine;
  • Broadcast 虽然是广播通知全部等待的 Goroutine,但是真正被唤醒时也是按照一定顺序的;

扩展原语

除了这些标准库中提供的同步原语之外,Go 语言还在子仓库 x/sync 中提供了额外的四种同步原语,ErrGroupSemaphoreSingleFlight 和 SyncMap,其中的 SyncMap 其实就是 sync 包中的 sync.Map,它在 1.9 版本的 Go 语言中被引入了 x/sync 包,随着 API 的成熟和稳定最后被移到了标准库 sync 包中。

golang-extension-sync-primitives

我们在这一节中就会介绍 Go 语言目前在扩展包中提供的三种原语,也就是 ErrGroupSemaphore 和 SingleFlight

ErrGroup

子仓库 x/sync 中的包 errgroup 其实就为我们在一组 Goroutine 中提供了同步、错误传播以及上下文取消的功能,我们可以使用如下所示的方式并行获取网页的数据:

var g errgroup.Group
var urls = []string{
    "http://www.golang.org/",
    "http://www.google.com/",
    "http://www.somestupidname.com/",
}
for i := range urls {
    url := urls[i]
    g.Go(func() error {
        resp, err := http.Get(url)
        if err == nil {
            resp.Body.Close()
        }
        return err
    })
}
if err := g.Wait(); err == nil {
    fmt.Println("Successfully fetched all URLs.")
}

Go 方法能够创建一个 Goroutine 并在其中执行传入的函数,而 Wait 方法会等待 Go 方法创建的 Goroutine 全部返回后返回第一个非空的错误,如果所有的 Goroutine 都没有返回错误,该函数就会返回 nil

结构体

errgroup 包中的 Group 结构体同时由三个比较重要的部分组成:

  1. 创建 Context 时返回的 cancel 函数,主要用于通知使用 context 的 Goroutine 由于某些子任务出错,可以停止工作让出资源了;
  2. 用于等待一组 Goroutine 完成子任务的 WaitGroup 同步原语;
  3. 用于接受子任务返回错误的 err 和保证 err 只会被赋值一次的 errOnce
type Group struct {
	cancel func()

	wg sync.WaitGroup

	errOnce sync.Once
	err     error
}

这些字段共同组成了 Group 结构体并为我们提供同步、错误传播以及上下文取消等功能。

操作

errgroup 对外唯一暴露的构造器就是 WithContext 方法,我们只能从一个 Context 中创建一个新的 Group 变量,WithCancel 返回的取消函数也仅会在 Group 结构体内部使用:

func WithContext(ctx context.Context) (*Group, context.Context) {
	ctx, cancel := context.WithCancel(ctx)
	return &Group{cancel: cancel}, ctx
}

创建新的并行子任务需要使用 Go 方法,这个方法内部会对 WaitGroup 加一并创建一个新的 Goroutine,在 Goroutine 内部运行子任务并在返回错误时及时调用 cancel 并对 err 赋值,只有最早返回的错误才会被上游感知到,后续的错误都会被舍弃:

func (g *Group) Go(f func() error) {
	g.wg.Add(1)

	go func() {
		defer g.wg.Done()

		if err := f(); err != nil {
			g.errOnce.Do(func() {
				g.err = err
				if g.cancel != nil {
					g.cancel()
				}
			})
		}
	}()
}

func (g *Group) Wait() error {
	g.wg.Wait()
	if g.cancel != nil {
		g.cancel()
	}
	return g.err
}

Wait 方法其实就只是调用了 WaitGroup 的同步方法,在子任务全部完成时取消 Context 并返回可能出现的错误。

小结

errgroup 包中的 Group 同步原语的实现原理还是非常简单的,它没有涉及非常底层和运行时包中的 API,只是对基本同步语义进行了简单的封装提供了更加复杂的功能,在使用时我们也需要注意以下的几个问题:

  • 出现错误或者等待结束后都会调用 Context 的 cancel 方法取消上下文;
  • 只有第一个出现的错误才会被返回,剩余的错误都会被直接抛弃;

Semaphore

信号量是在并发编程中比较常见的一种同步机制,它会保证持有的计数器在 0 到初始化的权重之间,每次获取资源时都会将信号量中的计数器减去对应的数值,在释放时重新加回来,当遇到计数器大于信号量大小时就会进入休眠等待其他进程释放信号,我们常常会在控制访问资源的进程数量时用到。

Golang 的扩展包中就提供了带权重的信号量,我们可以按照不同的权重对资源的访问进行管理,这个包对外也只提供了四个方法:

  • NewWeighted 用于创建新的信号量;
  • Acquire 获取了指定权重的资源,如果当前没有『空闲资源』,就会陷入休眠等待;
  • TryAcquire 也用于获取指定权重的资源,但是如果当前没有『空闲资源』,就会直接返回 false
  • Release 用于释放指定权重的资源;

结构体

NewWeighted 方法的主要作用创建一个新的权重信号量,传入信号量最大的权重就会返回一个新的 Weighted 结构体指针:

func NewWeighted(n int64) *Weighted {
	w := &Weighted{size: n}
	return w
}

type Weighted struct {
	size    int64
	cur     int64
	mu      sync.Mutex
	waiters list.List
}

Weighted 结构体中包含一个 waiters 列表其中存储着等待获取资源的『用户』,除此之外它还包含当前信号量的上限以及一个计数器 cur,这个计数器的范围就是 [0, size]

golang-semaphore

信号量中的计数器会随着用户对资源的访问和释放进行改变,引入的权重概念能够帮助我们更好地对资源的访问粒度进行控制,尽可能满足所有常见的用例。

获取

在上面我们已经提到过 Acquire 方法就是用于获取指定权重资源的方法,这个方法总共由三个不同的情况组成:

  1. 当信号量中剩余的资源大于获取的资源并且没有等待的 Goroutine 时就会直接获取信号量;
  2. 当需要获取的信号量大于 Weighted 的大小时,由于不可能满足条件就会直接返回;
  3. 遇到其他情况时会将当前 Goroutine 加入到等待列表并通过 select 等待当前 Goroutine 被唤醒,被唤醒后就会获取信号量;
func (s *Weighted) Acquire(ctx context.Context, n int64) error {
	s.mu.Lock()
	if s.size-s.cur >= n && s.waiters.Len() == 0 {
		s.cur += n
		s.mu.Unlock()
		return nil
	}

	if n > s.size {
		s.mu.Unlock()
		<-ctx.Done()
		return ctx.Err()
	}

	ready := make(chan struct{})
	w := waiter{n: n, ready: ready}
	elem := s.waiters.PushBack(w)
	s.mu.Unlock()

	select {
	case <-ctx.Done():
		err := ctx.Err()
		s.mu.Lock()
		select {
		case <-ready:
			err = nil
		default:
			s.waiters.Remove(elem)
		}
		s.mu.Unlock()
		return err

	case <-ready:
		return nil
	}
}

另一个用于获取信号量的方法 TryAcquire 相比之下就非常简单,它只会判断当前信号量是否有充足的资源获取,如果有充足的资源就会直接立刻返回 true 否则就会返回 false

func (s *Weighted) TryAcquire(n int64) bool {
	s.mu.Lock()
	success := s.size-s.cur >= n && s.waiters.Len() == 0
	if success {
		s.cur += n
	}
	s.mu.Unlock()
	return success
}

与 Acquire 相比,TryAcquire 由于不会等待资源的释放所以可能更适用于一些延时敏感、用户需要立刻感知结果的场景。

释放

最后要介绍的 Release 方法其实也非常简单,当我们对信号量进行释放时,Release 方法会从头到尾遍历 waiters 列表中全部的等待者,如果释放资源后的信号量有充足的剩余资源就会通过 Channel 唤起指定的 Goroutine:

func (s *Weighted) Release(n int64) {
	s.mu.Lock()
	s.cur -= n
	for {
		next := s.waiters.Front()
		if next == nil {
			break
		}

		w := next.Value.(waiter)
		if s.size-s.cur < w.n {
			break
		}

		s.cur += w.n
		s.waiters.Remove(next)
		close(w.ready)
	}
	s.mu.Unlock()
}

当然也可能会出现剩余资源无法唤起 Goroutine 的情况,在这时当前方法就会释放锁后直接返回,通过对这段代码的分析我们也能发现,如果一个信号量需要的占用的资源非常多,他可能会长时间无法获取锁,这可能也是 Acquire 方法引入另一个参数 Context 的原因,为信号量的获取设置一个超时时间。

小结

带权重的信号量确实有着更多的应用场景,这也是 Go 语言对外提供的唯一一种信号量实现,在使用的过程中我们需要注意以下的几个问题:

  • Acquire 和 TryAcquire 方法都可以用于获取资源,前者用于同步获取会等待锁的释放,后者会在无法获取锁时直接返回;
  • Release 方法会按照 FIFO 的顺序唤醒可以被唤醒的 Goroutine;
  • 如果一个 Goroutine 获取了较多地资源,由于 Release 的释放策略可能会等待比较长的时间;

SingleFlight

singleflight 是 Go 语言扩展包中提供了另一种同步原语,这其实也是作者最喜欢的一种同步扩展机制,它能够在一个服务中抑制对下游的多次重复请求,一个比较常见的使用场景是 — 我们在使用 Redis 对数据库中的一些热门数据进行了缓存并设置了超时时间,缓存超时的一瞬间可能有非常多的并行请求发现了 Redis 中已经不包含任何缓存所以大量的流量会打到数据库上影响服务的延时和质量。

golang-query-without-single-flight

但是 singleflight 就能有效地解决这个问题,它的主要作用就是对于同一个 Key 最终只会进行一次函数调用,在这个上下文中就是只会进行一次数据库查询,查询的结果会写回 Redis 并同步给所有请求对应 Key 的用户:

golang-extension-single-flight

这其实就减少了对下游的瞬时流量,在获取下游资源非常耗时,例如:访问缓存、数据库等场景下就非常适合使用 singleflight 对服务进行优化,在上述的这个例子中我们就可以在想 Redis 和数据库中获取数据时都使用 singleflight 提供的这一功能减少下游的压力;它的使用其实也非常简单,我们可以直接使用 singleflight.Group{} 创建一个新的 Group 结构体,然后通过调用 Do 方法就能对相同的请求进行抑制:

type service struct {
    requestGroup singleflight.Group
}

func (s *service) handleRequest(ctx context.Context, request Request) (Response, error) {
    v, err, _ := requestGroup.Do(request.Hash(), func() (interface{}, error) {
        rows, err := // select * from tables
        if err != nil {
            return nil, err
        }
        return rows, nil
    })
    if err != nil {
        return nil, err
    }
    
    return Response{
        rows: rows,
    }, nil
}

上述代码使用请求的哈希作为抑制相同请求的键,我们也可以选择一些比较关键或者重要的字段作为 Do 方法的第一个参数避免对下游的瞬时大量请求。

结构体

Group 结构体本身由一个互斥锁 Mutex 和一个从 Key 到 call 结构体指针的映射表组成,每一个 call 结构体都保存了当前这次调用对应的信息:

type Group struct {
	mu sync.Mutex
	m  map[string]*call
}

type call struct {
	wg sync.WaitGroup

	val interface{}
	err error

	dups  int
	chans []chan<- Result
}

call 结构体中的 val 和 err 字段都是在执行传入的函数时只会被赋值一次,它们也只会在 WaitGroup 等待结束都被读取,而 dups 和 chans 字段分别用于存储当前 singleflight 抑制的请求数量以及在结果返回时将信息传递给调用方。

操作

singleflight 包提供了两个用于抑制相同请求的方法,其中一个是同步等待的方法 Do,另一个是返回 Channel 的 DoChan,这两个方法在功能上没有太多的区别,只是在接口的表现上稍有不同。

每次 Do 方法的调用时都会获取互斥锁并尝试对 Group 持有的映射表进行懒加载,随后判断是否已经存在 key对应的函数调用:

  1. 当不存在对应的 call 结构体时:
    1. 初始化一个新的 call 结构体指针;
    2. 增加 WaitGroup 持有的计数器;
    3. 将 call 结构体指针添加到映射表;
    4. 释放持有的互斥锁 Mutex
    5. 阻塞地调用 doCall 方法等待结果的返回;
  2. 当已经存在对应的 call 结构体时;
    1. 增加 dups 计数器,它表示当前重复的调用次数;
    2. 释放持有的互斥锁 Mutex
    3. 通过 WaitGroup.Wait 等待请求的返回;
func (g *Group) Do(key string, fn func() (interface{}, error)) (v interface{}, err error, shared bool) {
	g.mu.Lock()
	if g.m == nil {
		g.m = make(map[string]*call)
	}
	if c, ok := g.m[key]; ok {
		c.dups++
		g.mu.Unlock()
		c.wg.Wait()
		return c.val, c.err, true
	}
	c := new(call)
	c.wg.Add(1)
	g.m[key] = c
	g.mu.Unlock()

	g.doCall(c, key, fn)
	return c.val, c.err, c.dups > 0
}

因为 val 和 err 两个字段都只会在 doCall 方法中被赋值,所以当 doCall 方法和 WaitGroup.Wait 方法返回时,这两个值就会返回给 Do 函数的调用者。

func (g *Group) doCall(c *call, key string, fn func() (interface{}, error)) {
	c.val, c.err = fn()
	c.wg.Done()

	g.mu.Lock()
	delete(g.m, key)
	for _, ch := range c.chans {
		ch <- Result{c.val, c.err, c.dups > 0}
	}
	g.mu.Unlock()
}

doCall 中会运行传入的函数 fn,该函数的返回值就会赋值给 c.val 和 c.err,函数执行结束后就会调用 WaitGroup.Done 方法通知所有被抑制的请求,当前函数已经执行完成,可以从 call 结构体中取出返回值并返回了;在这之后,doCall 方法会获取持有的互斥锁并通过管道将信息同步给使用 DoChan 方法的调用方。

func (g *Group) DoChan(key string, fn func() (interface{}, error)) <-chan Result {
	ch := make(chan Result, 1)
	g.mu.Lock()
	if g.m == nil {
		g.m = make(map[string]*call)
	}
	if c, ok := g.m[key]; ok {
		c.dups++
		c.chans = append(c.chans, ch)
		g.mu.Unlock()
		return ch
	}
	c := &call{chans: []chan<- Result{ch}}
	c.wg.Add(1)
	g.m[key] = c
	g.mu.Unlock()

	go g.doCall(c, key, fn)

	return ch
}

DoChan 方法和 Do的区别就是,它使用 Goroutine 异步执行 doCall 并向 call 持有的 chans 切片中追加 chan Result 变量,这也是它能够提供异步传值的原因。

小结

singleflight 包提供的 Group 接口确实非常好用,当我们需要这种抑制对下游的相同请求时就可以通过这个方法来增加吞吐量和服务质量,在使用的过程中我们也需要注意以下的几个问题:

  • Do 和 DoChan 一个用于同步阻塞调用传入的函数,一个用于异步调用传入的参数并通过 Channel 接受函数的返回值;
  • Forget 方法可以通知 singleflight 在持有的映射表中删除某个键,接下来对该键的调用就会直接执行方法而不是等待前面的函数返回;
  • 一旦调用的函数返回了错误,所有在等待的 Goroutine 也都会接收到同样的错误;

总结

我们在这一节中介绍了 Go 语言标准库中提供的基本原语以及扩展包中的扩展原语,这些并发编程的原语能够帮助我们更好地利用 Go 语言的特性构建高吞吐量、低延时的服务,并解决由于并发带来的错误,到这里我们再重新回顾一下这一节介绍的内容:

  • Mutex 互斥锁
    • 如果互斥锁处于初始化状态,就会直接通过置位 mutexLocked 加锁;
    • 如果互斥锁处于 mutexLocked 并且在普通模式下工作,就会进入自旋,执行 30 次 PAUSE 指令消耗 CPU 时间等待锁的释放;
    • 如果当前 Goroutine 等待锁的时间超过了 1ms,互斥锁就会被切换到饥饿模式;
    • 互斥锁在正常情况下会通过 runtime_SemacquireMutex 方法将调用 Lock 的 Goroutine 切换至休眠状态,等待持有信号量的 Goroutine 唤醒当前协程;
    • 如果当前 Goroutine 是互斥锁上的最后一个等待的协程或者等待的时间小于 1ms,当前 Goroutine 会将互斥锁切换回正常模式;
    • 如果互斥锁已经被解锁,那么调用 Unlock 会直接抛出异常;
    • 如果互斥锁处于饥饿模式,会直接将锁的所有权交给队列中的下一个等待者,等待者会负责设置 mutexLocked 标志位;
    • 如果互斥锁处于普通模式,并且没有 Goroutine 等待锁的释放或者已经有被唤醒的 Goroutine 获得了锁就会直接返回,在其他情况下回通过 runtime_Semrelease 唤醒对应的 Goroutine;
  • RWMutex 读写互斥锁
    • readerSem — 读写锁释放时通知由于获取读锁等待的 Goroutine;
    • writerSem — 读锁释放时通知由于获取读写锁等待的 Goroutine;
    • w 互斥锁 — 保证写操作之间的互斥;
    • readerCount — 统计当前进行读操作的协程数,触发写锁时会将其减少 rwmutexMaxReaders 阻塞后续的读操作;
    • readerWait — 当前读写锁等待的进行读操作的协程数,在触发 Lock 之后的每次 RUnlock 都会将其减一,当它归零时该 Goroutine 就会获得读写锁;
    • 当读写锁被释放 Unlock 时首先会通知所有的读操作,然后才会释放持有的互斥锁,这样能够保证读操作不会被连续的写操作『饿死』;
  • WaitGroup 等待一组 Goroutine 结束
    • Add 不能在和 Wait 方法在 Goroutine 中并发调用,一旦出现就会造成程序崩溃;
    • WaitGroup 必须在 Wait 方法返回之后才能被重新使用;
    • Done 只是对 Add 方法的简单封装,我们可以向 Add 方法传入任意负数(需要保证计数器非负)快速将计数器归零以唤醒其他等待的 Goroutine;
    • 可以同时有多个 Goroutine 等待当前 WaitGroup 计数器的归零,这些 Goroutine 也会被『同时』唤醒;
  • Once 程序运行期间仅执行一次
    • Do 方法中传入的函数只会被执行一次,哪怕函数中发生了 panic
    • 两次调用 Do 方法传入不同的函数时只会执行第一次调用的函数;
  • Cond 发生指定事件时唤醒
    • Wait 方法在调用之前一定要使用 L.Lock 持有该资源,否则会发生 panic 导致程序崩溃;
    • Signal 方法唤醒的 Goroutine 都是队列最前面、等待最久的 Goroutine;
    • Broadcast 虽然是广播通知全部等待的 Goroutine,但是真正被唤醒时也是按照一定顺序的;
  • ErrGroup 为一组 Goroutine 提供同步、错误传播以及上下文取消的功能
    • 出现错误或者等待结束后都会调用 Context 的 cancel 方法取消上下文;
    • 只有第一个出现的错误才会被返回,剩余的错误都会被直接抛弃;
  • Semaphore 带权重的信号量
    • Acquire 和 TryAcquire 方法都可以用于获取资源,前者用于同步获取会等待锁的释放,后者会在无法获取锁时直接返回;
    • Release 方法会按照 FIFO 的顺序唤醒可以被唤醒的 Goroutine;
    • 如果一个 Goroutine 获取了较多地资源,由于 Release 的释放策略可能会等待比较长的时间;
  • SingleFlight 用于抑制对下游的重复请求
    • Do 和 DoChan 一个用于同步阻塞调用传入的函数,一个用于异步调用传入的参数并通过 Channel 接受函数的返回值;
    • Forget 方法可以通知 singleflight 在持有的映射表中删除某个键,接下来对该键的调用就会直接执行方法而不是等待前面的函数返回;
    • 一旦调用的函数返回了错误,所有在等待的 Goroutine 也都会接收到同样的错误;

这些同步原语的实现不仅要考虑 API 接口的易用、解决并发编程中可能遇到的线程竞争问题,还需要对尾延时进行优化避免某些 Goroutine 无法获取锁或者资源而被饿死,对同步原语的学习也能够增强我们队并发编程的理解和认识,也是了解并发编程无法跨越的一个步骤。

Reference

关于图片和转载

知识共享许可协议

本作品采用知识共享署名 4.0 国际许可协议进行许可。 转载时请注明原文链接,图片在使用时请保留图片中的全部内容,可适当缩放并在引用处附上图片所在的文章链接,图片使用 Sketch 进行绘制。

如何写出优雅的 Golang 代码

https://draveness.me/golang-101

Go 语言是一门简单、易学的编程语言,对于有编程背景的工程师来说,学习 Go 语言并写出能够运行的代码并不是一件困难的事情,对于之前有过其他语言经验的开发者来说,写什么语言都像自己学过的语言其实是有问题的,想要真正融入生态写出优雅的代码就一定要花一些时间和精力了解语言背后的设计哲学和最佳实践。

如果你之前没有 Go 语言的开发经历,正在学习和使用 Go 语言,相信这篇文章能够帮助你更快地写出优雅的 Go 语言代码;在这篇文章中,我们并不会给一个长长地列表介绍变量、方法和结构体应该怎么命名,这些 Go 语言的代码规范可以在 Go Code Review Comments 中找到,它们非常重要但并不是这篇文章想要介绍的重点,我们将从代码结构、最佳实践以及单元测试几个不同的方面介绍如何写出优雅的 Go 语言代码。

写在前面

想要写出好的代码并不是一件容易的事情,它需要我们不断地对现有的代码进行反思 — 如何改写这段代码才能让它变得更加优雅。优雅听起来是一个非常感性、难以量化的结果,然而这却是好的代码能够带来的最直观感受,它可能隐式地包含了以下特性:

  • 容易阅读和理解;
  • 容易测试、维护和扩展;
  • 命名清晰、无歧义、注释完善清楚;

相信读完了这篇文章,我们也不能立刻写出优雅的 Go 语言代码,但是如果我们遵循这里介绍几个的容易操作并且切实可行的方法,就帮助我们走出第一步,作者写这篇文章有以下的几个目的:

  • 帮助 Go 语言的开发者了解生态中的规范与工具,写出更优雅的代码;
  • 为代码和项目的管理提供被社区广泛认同的规则、共识以及最佳实践;

代码规范

代码规范其实是一个老生常态的问题,我们也不能免俗还是要简单介绍一下相关的内容,Go 语言比较常见并且使用广泛的代码规范就是官方提供的 Go Code Review Comments,无论你是短期还是长期使用 Go 语言编程,都应该至少完整地阅读一遍这个官方的代码规范指南,它既是我们在写代码时应该遵守的规则,也是在代码审查时需要注意的规范。

学习 Go 语言相关的代码规范是一件非常重要的事情,也是让我们的项目遵循统一规范的第一步,虽然阅读代码规范相关的文档非常重要,但是在实际操作时我们并不能靠工程师自觉地遵守以及经常被当做形式的代码审查,而是需要借助工具来辅助执行。

辅助工具

使用自动化的工具保证项目遵守一些最基本的代码规范是非常容易操作和有效的事情,相比之下人肉审查代码的方式更加容易出错,也会出现一些违反规则和约定的特例,维护代码规范的最好方式就是『尽量自动化一切能够自动化的步骤,让工程师审查真正重要的逻辑和设计』

我们在这一节中就会介绍两种非常切实有效的办法帮助我们在项目中自动化地进行一些代码规范检查和静态检查保证项目的质量。

goimports

goimports 是 Go 语言官方提供的工具,它能够为我们自动格式化 Go 语言代码并对所有引入的包进行管理,包括自动增删依赖的包引用、将依赖包按字母序排序并分类。相信很多人使用的 IDE 都会将另一个官方提供的工具 gofmt 对代码进行格式化,而 goimports 就是等于 gofmt 加上依赖包管理。

golang-goimports

建议所有 Go 语言的开发者都在开发时使用 goimports,虽然 goimports 有时会引入错误的包,但是与带来的好处相比,这些偶尔出现的错误在作者看来也是可以接受的;当然,不想使用 goimports 的开发者也一定要在 IDE 或者编辑器中开启自动地 gofmt(保存时自动格式化)。

在 IDE 和 CI 检查中开启自动地 gofmt 或者 goimports 检查是没有、也不应该有讨论的必要的,这就是一件使用和开发 Go 语言必须要做的事情。

golint

另一个比较常用的静态检查工具就是 golint 了,作为官方提供的工具,它在可定制化上有着非常差的支持,我们只能通过如下所示的方式运行 golint 对我们的项目进行检查:

$ golint ./pkg/...
pkg/liquidity/liquidity_pool.go:18:2: exported var ErrOrderBookNotFound should have comment or be unexported
pkg/liquidity/liquidity_pool.go:23:6: exported type LiquidityPool should have comment or be unexported
pkg/liquidity/liquidity_pool.go:23:6: type name will be used as liquidity.LiquidityPool by other packages, and that stutters; consider calling this Pool
pkg/liquidity/liquidity_pool.go:31:1: exported function NewLiquidityPool should have comment or be unexported
...

社区上有关于 golint 定制化的 讨论golint 的开发者给出了以下的几个观点解释为什么 golint 不支持定制化的功能:

  • lint 的目的就是在 Go 语言社区中鼓励统一、一致的编程风格,某些开发者也许不会同意其中的某些规范,但是使用统一的风格对于 Go 语言社区有比较强的好处,而能够开关指定规则的功能会导致 golint 不能够有效地完成这个工作;
  • 有一些静态检查的规则会导致一些错误的警告,这些情况确实非常让人头疼,但是我会选择支持在 golint 中直接保留或者删除这些规则,而不是随提供意增删规则的能力;
  • 能够通过 min_confidence 过滤一些静态检查规则,但是需要我们选择合适的值;

golint 作者的观点在 issue 中得到了非常多的 👎,但是这件事情很难说对错;在社区中保证一致的编程规范是一件非常有益的事情,不过对于很多公司内部的服务或者项目,可能在业务服务上就会发生一些比较棘手的情况,使用这种过强的约束没有太多明显地收益。

golang-lint

更推荐的方法是在基础库或者框架中使用 golint 进行静态检查(或者同时使用 golint 和 golangci-lint),在其他的项目中使用可定制化的 golangci-lint 来进行静态检查,因为在基础库和框架中施加强限制对于整体的代码质量有着更大的收益。

作者会在自己的 Go 项目中使用 golint + golangci-lint 并开启全部的检查尽量尽早发现代码中包含文档在内的全部缺陷。

自动化

无论是用于检查代码规范和依赖包的 goimports 还是静态检查工具 glint 或者 golangci-lint,只要我们在项目中引入这些工具就一定要在代码的 CI 流程中加入对应的自动化检查:

在自建的或者其他的代码托管平台上也应该想尽办法寻找合适的工具,现代的代码托管工具应该都会对 CI/CD 有着非常不错的支持;我们需要通过这些 CI 工具将代码的自动化检查变成 PR 合并和发版的一个前置条件,减少工程师 Review 代码时可能发生的疏漏。

最佳实践

我们在上一节中介绍了一些能通过自动化工具发现的问题,这一节提到的最佳实践可能就没有办法通过自动化工具进行保证,这些最佳实践更像是 Go 语言社区内部发展过程中积累的一些工程经验和共识,遵循这些最佳实践能够帮助我们写出符合 Go 语言『味道』的代码,我们将在这一小节覆盖以下的几部分内容:

  • 目录结构;
  • 模块拆分;
  • 显式调用;
  • 面向接口;

这四部分内容是在社区中相对来说比较常见的约定,如果我们学习并遵循了这些约定,同时在 Go 语言的项目中实践这几部分内容,相信一定会对我们设计 Go 语言项目有所帮助。

目录结构

目录结构基本上就是一个项目的门面,很多时候我们从目录结构中就能够看出开发者对这门语言是否有足够的经验,所以在这里首先要介绍的最佳实践就是如何在 Go 语言的项目或者服务中组织代码。

官方并没有给出一个推荐的目录划分方式,很多项目对于目录结构的划分也非常随意,这其实也是没有什么问题的,但是社区中还是有一些比较常见的约定,例如:golang-standards/project-layout 项目中就定义了一个比较标准的目录结构。

├── LICENSE.md
├── Makefile
├── README.md
├── api
├── assets
├── build
├── cmd
├── configs
├── deployments
├── docs
├── examples
├── githooks
├── init
├── internal
├── pkg
├── scripts
├── test
├── third_party
├── tools
├── vendor
├── web
└── website

我们在这里就像简单介绍其中几个比较常见并且重要的目录和文件,帮助我们快速理解如何使用如上所示的目录结构,如果各位读者想要了解使用其他目录的原因,可以从 golang-standards/project-layout 项目中的 README 了解更详细的内容。

/pkg

/pkg 目录是 Go 语言项目中非常常见的目录,我们几乎能够在所有知名的开源项目(非框架)中找到它的身影,例如:

这个目录中存放的就是项目中可以被外部应用使用的代码库,其他的项目可以直接通过 import 引入这里的代码,所以当我们将代码放入 pkg 时一定要慎重,不过如果我们开发的是 HTTP 或者 RPC 的接口服务或者公司的内部服务,将私有和公有的代码都放到 /pkg 中也没有太多的不妥,因为作为最顶层的项目来说很少会被其他应用直接依赖,当然严格遵循公有和私有代码划分是非常好的做法,作者也建议各位开发者对项目中公有和私有的代码进行妥善的划分。

私有代码

私有代码推荐放到 /internal 目录中,真正的项目代码应该写在 /internal/app 里,同时这些内部应用依赖的代码库应该在 /internal/pkg 子目录和 /pkg 中,下图展示了一个使用 /internal 目录的项目结构:

golang-internal-app-and-pkg

当我们在其他项目引入包含 internal 的依赖时,Go 语言会在编译时报错:

An import of a path containing the element “internal” is disallowed
if the importing code is outside the tree rooted at the parent of the 
"internal" directory.

这种错误只有在被引入的 internal 包不存在于当前项目树中才会发生,如果在同一个项目中引入该项目的 internal 包并不会出现这种错误。

/src

在 Go 语言的项目最不应该有的目录结构其实就是 /src 了,社区中的一些项目确实有 /src 文件夹,但是这些项目的开发者之前大多数都有 Java 的编程经验,这在 Java 和其他语言中其实是一个比较常见的代码组织方式,但是作为一个 Go 语言的开发者,我们不应该允许项目中存在 /src 目录。

最重要的原因其实是 Go 语言的项目在默认情况下都会被放置到 $GOPATH/src 目录下,这个目录中存储着我们开发和依赖的全部项目代码,如果我们在自己的项目中使用 /src 目录,该项目的 PATH 中就会出现两个 src

$GOPATH/src/github.com/draveness/project/src/code.go

上面的目录结构看起来非常奇怪,这也是我们在 Go 语言中不建议使用 /src 目录的最重要原因。

当然哪怕我们在 Go 语言的项目中使用 /src 目录也不会导致编译不通过或者其他问题,如果坚持这种做法对于项目的可用性也没有任何的影响,但是如果想让我们『看起来』更专业,还是遵循社区中既定的约定减少其他 Go 语言开发者的理解成本,这对于社区来说是一件好事。

平铺

另一种在 Go 语言中组织代码的方式就是项目的根目录下放项目的代码,这种方式在很多框架或者库中非常常见,如果想要引入一个使用 pkg 目录结构的框架时,我们往往需要使用 github.com/draveness/project/pkg/somepkg,当代码都平铺在项目的根目录时只需要使用 github.com/draveness/project,很明显地减少了引用依赖包语句的长度。

所以对于一个 Go 语言的框架或者库,将代码平铺在根目录下也很正常,但是在一个 Go 语言的服务中使用这种代码组织方法可能就没有那么合适了。

/cmd

/cmd 目录中存储的都是当前项目中的可执行文件,该目录下的每一个子目录都应该包含我们希望有的可执行文件,如果我们的项目是一个 grpc 服务的话,可能在 /cmd/server/main.go 中就包含了启动服务进程的代码,编译后生成的可执行文件就是 server

我们不应该在 /cmd 目录中放置太多的代码,我们应该将公有代码放置到 /pkg 中并将私有代码放置到 /internal 中并在 /cmd 中引入这些包,保证 main 函数中的代码尽可能简单和少。

/api

/api 目录中存放的就是当前项目对外提供的各种不同类型的 API 接口定义文件了,其中可能包含类似 /api/protobuf-spec/api/thrift-spec 或者 /api/http-spec 的目录,这些目录中包含了当前项目对外提供的和依赖的所有 API 文件:

$ tree ./api
api
└── protobuf-spec
    └── oceanbookpb
        ├── oceanbook.pb.go
        └── oceanbook.proto

二级目录的主要作用就是在一个项目同时提供了多种不同的访问方式时,用这种办法避免可能存在的潜在冲突问题,也可以让项目结构的组织更加清晰。

Makefile

最后要介绍的 Makefile 文件也非常值得被关注,在任何一个项目中都会存在一些需要运行的脚本,这些脚本文件应该被放到 /scripts 目录中并由 Makefile 触发,将这些经常需要运行的命令固化成脚本减少『祖传命令』的出现。

小结

总的来说,每一个项目都应该按照固定的组织方式进行实现,这种约定虽然并不是强制的,但是无论是组内、公司内还是整个 Go 语言社区中,只要达成了一致,对于其他工程师快速梳理和理解项目都是很有帮助的。

这一节介绍的 Go 语言项目的组织方式也并不是强制要求的,这只是 Go 语言社区中经常出现的项目组织方式,一个大型项目在使用这种目录结构时也会对其进行微调,不过这种组织方式确实更为常见并且合理。

模块拆分

我们既然已经介绍过了如何从顶层对项目的结构进行组织,接下来就会深入到项目的内部介绍 Go 语言对模块的一些拆分方法。

Go 语言的一些顶层设计最终导致了它在划分模块上与其他的编程语言有着非常明显的不同,很多其他语言的 Web 框架都采用 MVC 的架构模式,例如 Rails 和 Spring MVC,Go 语言对模块划分的方法就与 Ruby 和 Java 完全不同。

按层拆分

无论是 Java 还是 Ruby,它们最著名的框架都深受 MVC 架构模式 的影响,我们从 Spring MVC 的名字中就能体会到 MVC 对它的影响,而 Ruby 社区的 Rails 框架也与 MVC 的关系非常紧密,这是一种 Web 框架的最常见架构方式,将服务中的不同组件分成了 Model、View 和 Controller 三层。

divide-by-laye

这种模块拆分的方式其实就是按照层级进行拆分,Rails 脚手架默认生成的代码其实就是将这三层不同的源文件放在对应的目录下:modelsviews 和 controllers,我们通过 rails new example 生成一个新的 Rails 项目后可以看到其中的目录结构:

$ tree -L 2 app
app
├── controllers
│   ├── application_controller.rb
│   └── concerns
├── models
│   ├── application_record.rb
│   └── concerns
└── views
    └── layouts

而很多 Spring MVC 的项目中也会出现类似 modeldaoview 的目录,这种按层拆分模块的设计其实有以下的几方面原因:

  1. MVC 架构模式 — MVC 本身就强调了按层划分职责的设计,所以遵循该模式设计的框架自然有着一脉相承的思路;
  2. 扁平的命名空间 — 无论是 Spring MVC 还是 Rails,同一个项目中命名空间非常扁平,跨文件夹使用其他文件夹中定义的类或者方法不需要引入新的包,使用其他文件定义的类时也不需要增加额外的前缀,多个文件定义的类被『合并』到了同一个命名空间中;
  3. 单体服务的场景 — Spring MVC 和 Rails 刚出现时,SOA 和微服务架构还不像今天这么普遍,绝大多数的场景也不需要通过拆分服务;

上面的几个原因共同决定了 Spring MVC 和 Rails 会出现 modelsviews 和 controllers 的目录并按照层级的方式对模块进行拆分。

按职责拆分

Go 语言在拆分模块时就使用了完全不同的思路,虽然 MVC 架构模式是在我们写 Web 服务时无法避开的,但是相比于横向地切分不同的层级,Go 语言的项目往往都按照职责对模块进行拆分:

divide-by-responsibility

对于一个比较常见的博客系统,使用 Go 语言的项目会按照不同的职责将其纵向拆分成 postusercomment三个模块,每一个模块都对外提供相应的功能,post 模块中就包含相关的模型和视图定义以及用于处理 API 请求的控制器(或者服务):

$ tree pkg
pkg
├── comment
├── post
│   ├── handler.go
│   └── post.go
└── user

Go 语言项目中的每一个文件目录都代表着一个独立的命名空间,也就是一个单独的包,当我们想要引用其他文件夹的目录时,首先需要使用 import 关键字引入相应的文件目录,再通过 pkg.xxx 的形式引用其他目录定义的结构体、函数或者常量,如果我们在 Go 语言中使用 modelview 和 controller 来划分层级,你会在其他的模块中看到非常多的 model.Postmodel.Comment 和 view.PostView

这种划分层级的方法在 Go 语言中会显得非常冗余,并且如果对项目依赖包的管理不够谨慎时,很容易发生引用循环,出现这些问题的最根本原因其实也非常简单:

  1. Go 语言对同一个项目中不同目录的命名空间做了隔离,整个项目中定义的类和方法并不是在同一个命名空间下的,这也就需要工程师自己维护不同包之间的依赖关系;
  2. 按照职责垂直拆分的方式在单体服务遇到瓶颈时非常容易对微服务进行拆分,我们可以直接将一个负责独立功能的 package 拆出去,对这部分性能热点单独进行扩容;

小结

项目是按照层级还是按照职责对模块进行拆分其实并没有绝对的好与不好,语言和框架层面的设计最终决定了我们应该采用哪种方式对项目和代码进行组织。

Java 和 Ruby 这些语言在框架中往往采用水平拆分的方式划分不同层级的职责,而 Go 语言项目的最佳实践就是按照职责对模块进行垂直拆分,将代码按照功能的方式分到多个 package 中,这并不是说 Go 语言中不存在模块的水平拆分,只是因为 package 作为一个 Go 语言访问控制的最小粒度,所以我们应该遵循顶层的设计使用这种方式构建高内聚的模块。

显式与隐式

从开始学习、使用 Go 语言到参与社区上一些开源的 Golang 项目,作者发现 Go 语言社区对于显式的初始化、方法调用和错误处理非常推崇,类似 Spring Boot 和 Rails 的框架其实都广泛地采纳了『约定优于配置』的中心思想,简化了开发者和工程师的工作量。

然而 Go 语言社区虽然达成了很多的共识与约定,但是从语言的设计以及工具上的使用我们就能发现显式地调用方法和错误处理是被鼓励的。

init

我们在这里先以一个非常常见的函数 init 为例,介绍 Go 语言社区对显式调用的推崇;相信很多人都在一些 package 中阅读过这样的代码:

var grpcClient *grpc.Client

func init() {
    var err error
    grpcClient, err = grpc.Dial(...)
    if err != nil {
        panic(err)
    }
}

func GetPost(postID int64) (*Post, error) {
    post, err := grpcClient.FindPost(context.Background(), &pb.FindPostRequest{PostID: postID})
    if err != nil {
        return nil, err
    }
    
    return post, nil
}

这种代码虽然能够通过编译并且正常工作,然而这里的 init 函数其实隐式地初始化了 grpc 的连接资源,如果另一个 package 依赖了当前的包,那么引入这个依赖的工程师可能会在遇到错误时非常困惑,因为在 init 函数中做这种资源的初始化是非常耗时并且容易出现问题的。

一种更加合理的做法其实是这样的,首先我们定义一个新的 Client 结构体以及一个用于初始化结构的 NewClient 函数,这个函数接收了一个 grpc 连接作为入参返回一个用于获取 Post 资源的客户端,GetPost 成为了这个结构体的方法,每当我们调用 client.GetPost 时都会用到结构体中保存的 grpc 连接:

// pkg/post/client.go
type Client struct {
    grpcClient *grpc.ClientConn    
}

func NewClient(grpcClient *grpcClientConn) Client {
    return &Client{
        grpcClient: grpcClient,
    }
}

func (c *Client) GetPost(postID int64) (*Post, error) {
    post, err := c.grpcClient.FindPost(context.Background(), &pb.FindPostRequest{PostID: postID})
    if err != nil {
        return nil, err
    }
    
    return post, nil
}

初始化 grpc 连接的代码应该放到 main 函数或者 main 函数调用的其他函数中执行,如果我们在 main 函数中显式的初始化这种依赖,对于其他的工程师来说就非常易于理解,我们从 main 函数开始就能梳理出程序启动的整个过程。

// cmd/grpc/main.go
func main() {
    grpcClient, err := grpc.Dial(...)
    if err != nil {
        panic(err)
    }
    
    postClient := post.NewClient(grpcClient)
    // ...
}

各个模块之间会构成一种树形的结构和依赖关系,上层的模块会持有下层模块中的接口或者结构体,不会存在孤立的、不被引用的对象。

golang-project-and-tree-structure

上图中出现的两个 Database 其实是在 main 函数中初始化的数据库连接,在项目运行期间,它们可能表示同一个内存中的数据库连接

当我们使用 golangci-lint 并开启 gochecknoinits 和 gochecknoglobals 静态检查时,它其实严格地限制我们对 init 函数和全局变量的使用。

当然这并不是说我们一定不能使用 init 函数,作为 Go 语言赋予开发者的能力,因为它能在包被引入时隐式地执行了一些代码,所以我们更应该慎重地使用它们。

一些框架会在 init 中判断是否满足使用的前置条件,但是对于很多的 Web 或者 API 服务来说,大量使用 init 往往意味着代码质量的下降以及不合理的设计。

func init() {
    if user == "" {
        log.Fatal("$USER not set")
    }
    if home == "" {
        home = "/home/" + user
    }
    if gopath == "" {
        gopath = home + "/go"
    }
    // gopath may be overridden by --gopath flag on command line.
    flag.StringVar(&gopath, "gopath", gopath, "override default GOPATH")
}

上述代码其实是 Effective Go 在介绍 init 方法使用是展示的实例代码,这是一个比较合理地 init 函数使用示例,我们不应该在 init 中做过重的初始化逻辑,而是做一些简单、轻量的前置条件判断。

error

另一个要介绍的就是 Go 语言的错误处理机制了,虽然 Golang 的错误处理被开发者诟病已久,但是工程师每天都在写 if err != nil { return nil, err } 的错误处理逻辑其实就是在显式地对错误处理,关注所有可能会发生错误的方法调用并在无法处理时抛给上层模块。

func ListPosts(...) ([]Post, error) {
    conn, err := gorm.Open(...)
    if err != nil {
        return []Post{}, err
    }
    
    var posts []Post
    if err := conn.Find(&posts).Error; err != nil {
        return []Post{}, err
    }
    
    return posts, nil
}

上述代码只是简单展示 Go 语言常见的错误处理逻辑,我们不应该在这种方法中初始化数据库的连接。

虽然 Golang 中也有类似 Java 或者 Ruby try/catch 关键字,但是很少有人会在代码中使用 panic 和 recover 来实现错误和异常的处理,与 init 函数一样,Go 语言对于 panic 和 recover 的使用也非常谨慎。

当我们在 Go 语言中处理错误相关的逻辑时,最重要的其实就是以下几点:

  1. 使用 error 实现错误处理 — 尽管这看起来非常啰嗦;
  2. 将错误抛给上层处理 — 对于一个方法是否需要返回 error 也需要我们仔细地思考,向上抛出错误时可以通过 errors.Wrap 携带一些额外的信息方便上层进行判断;
  3. 处理所有可能返回的错误 — 所有可能返回错误的地方最终一定会返回错误,考虑全面才能帮助我们构建更加健壮的项目;

小结

作者在使用 Go 语言的这段时间,能够深刻地体会到它对于显式方法调用与错误处理的鼓励,这不仅能够帮助项目的其他开发者快速地理解上下文,也能够帮助我们构建更加健壮、容错性与可维护性更好的工程。

面向接口

面向接口编程是一个老生常谈的话题,接口 的作用其实就是为不同层级的模块提供了一个定义好的中间层,上游不再需要依赖下游的具体实现,充分地对上下游进行了解耦。

golang-interface

这种编程方式不仅是在 Go 语言中是被推荐的,在几乎所有的编程语言中,我们都会推荐这种编程的方式,它为我们的程序提供了非常强的灵活性,想要构建一个稳定、健壮的 Go 语言项目,不使用接口是完全无法做到的。

如果一个略有规模的项目中没有出现任何 type ... interface 的定义,那么作者可以推测出这在很大的概率上是一个工程质量堪忧并且没有多少单元测试覆盖的项目,我们确实需要认真考虑一下如何使用接口对项目进行重构。

单元测试是一个项目保证工程质量最有效并且投资回报率最高的方法之一,作为静态语言的 Golang,想要写出覆盖率足够(最少覆盖核心逻辑)的单元测试本身就比较困难,因为我们不能像动态语言一样随意修改函数和方法的行为,而接口就成了我们的救命稻草,写出抽象良好的接口并通过接口隔离依赖能够帮助我们有效地提升项目的质量和可测试性,我们会在下一节中详细介绍如何写单元测试。

package post

var client *grpc.ClientConn

func init() {
    var err error
    client, err = grpc.Dial(...)
    if err != nil {
        panic(err)
    }
}

func ListPosts() ([]*Post, error) {
    posts, err := client.ListPosts(...)
    if err != nil {
        return []*Post{}, err
    }
    
    return posts, nil
}

上述代码其实就不是一个设计良好的代码,它不仅在 init 函数中隐式地初始化了 grpc 连接这种全局变量,而且没有将 ListPosts 通过接口的方式暴露出去,这会让依赖 ListPosts 的上层模块难以测试。

我们可以使用下面的代码改写原有的逻辑,使得同样地逻辑变得更容易测试和维护:

package post

type Service interface {
    ListPosts() ([]*Post, error)
}

type service struct {
    conn *grpc.ClientConn
}

func NewService(conn *grpc.ClientConn) Service {
    return &service{
        conn: conn,
    }
}

func (s *service) ListPosts() ([]*Post, error) {
    posts, err := s.conn.ListPosts(...)
    if err != nil {
        return []*Post{}, err
    }
    
    return posts, nil
}
  1. 通过接口 Service 暴露对外的 ListPosts 方法;
  2. 使用 NewService 函数初始化 Service 接口的实现并通过私有的结构体 service 持有 grpc 连接;
  3. ListPosts 不再依赖全局变量,而是依赖接口体 service 持有的连接;

当我们使用这种方式重构代码之后,就可以在 main 函数中显式的初始化 grpc 连接、创建 Service 接口的实现并调用 ListPosts 方法:

package main

import ...

func main() {
    conn, err = grpc.Dial(...)
    if err != nil {
        panic(err)
    }
    
    svc := post.NewService(conn)
    posts, err := svc.ListPosts()
    if err != nil {
        panic(err)
    }
    
    fmt.Println(posts)
}

这种使用接口组织代码的方式在 Go 语言中非常常见,我们应该在代码中尽可能地使用这种思想和模式对外提供功能:

  1. 使用大写的 Service 对外暴露方法;
  2. 使用小写的 service 实现接口中定义的方法;
  3. 通过 NewService 函数初始化 Service 接口;

当我们使用上述方法组织代码之后,其实就对不同模块的依赖进行了解耦,也正遵循了软件设计中经常被提到的一句话 — 『依赖接口,不要依赖实现』,也就是面向接口编程

小结

在这一小节中总共介绍了 Go 语言中三个经常会打交道的『元素』— init 函数、error 和接口,我们在这里主要是想通过三个不同的例子为大家传达的一个主要思想就是尽量使用显式的(explicit)的方式编写 Go 语言代码。

单元测试

一个代码质量和工程质量有保证的项目一定有比较合理的单元测试覆盖率,没有单元测试的项目一定是不合格的或者不重要的,单元测试应该是所有项目都必须有的代码,每一个单元测试都表示一个可能发生的情况,单元测试就是业务逻辑

作为软件工程师,重构现有的项目对于我们来说应该是一件比较正常的事情,如果项目中没有单元测试,我们很难在不改变已有业务逻辑的情况对项目进行重构,一些业务的边界情况很可能会在重构的过程中丢失,当时参与相应 case 开发的工程师可能已经不在团队中,而项目相关的文档可能也消失在了归档的 wiki 中(更多的项目可能完全没有文档),我们能够在重构中相信的东西其实只有当前的代码逻辑(很可能是错误的)以及单元测试(很可能是没有的)。

简单总结一下,单元测试的缺失不仅会意味着较低的工程质量,而且意味着重构的难以进行,一个有单元测试的项目尚且不能够保证重构前后的逻辑完全相同,一个没有单元测试的项目很可能本身的项目质量就堪忧,更不用说如何在不丢失业务逻辑的情况下进行重构了

可测试

写代码并不是一件多困难的事情,不过想要在项目中写出可以测试的代码并不容易,而优雅的代码一定是可以测试的,我们在这一节中需要讨论的就是什么样的代码是可以测试的。

如果想要想清楚什么样的才是可测试的,我们首先要知道测试是什么?作者对于测试的理解就是控制变量,在我们隔离了待测试方法中一些依赖之后,当函数的入参确定时,就应该得到期望的返回值。

golang-unit-test

如何控制待测试方法中依赖的模块是写单元测试时至关重要的,控制依赖也就是对目标函数的依赖进行 Mock 消灭不确定性,为了减少每一个单元测试的复杂度,我们需要:

  1. 尽可能减少目标方法的依赖,让目标方法只依赖必要的模块;
  2. 依赖的模块也应该非常容易地进行 Mock

单元测试的执行不应该依赖于任何的外部模块,无论是调用外部的 HTTP 请求还是数据库中的数据,我们都应该想尽办法模拟可能出现的情况,因为单元测试不是集成测试的,它的运行不应该依赖除项目代码外的其他任何系统。

接口

在 Go 语言中如果我们完全不使用接口,是写不出易于测试的代码的,作为静态语言的 Golang,只有我们使用接口才能脱离依赖具体实现的窘境,接口的使用能够为我们带来更清晰的抽象,帮助我们思考如何对代码进行设计,也能让我们更方便地对依赖进行 Mock

我们再来回顾一下上一节对接口进行介绍时展示的常见模式:

type Service interface { ... }

type service struct { ... }

func NewService(...) (Service, error) {
    return &service{...}, nil
}

上述代码在 Go 语言中是非常常见的,如果你不知道应不应该使用接口对外提供服务,这时就应该无脑地使用上述模式对外暴露方法了,这种模式可以在绝大多数的场景下工作,至少作者到目前还没有见到过不适用的。

函数简单

另一个建议就是保证每一个函数尽可能简单,这里的简单不止是指功能上的简单、单一,还意味着函数容易理解并且命名能够自解释。

一些语言的 lint 工具其实会对函数的理解复杂度(PerceivedComplexity)进行检查,也就是检查函数中出现的 if/elseswitch/case 分支以及方法的调用的数量,一旦超过约定的阈值就会报错,Ruby 社区中的 Rubocop 和上面提到的 golangci-lint 都有这个功能。

Ruby 社区中的 Rubocop 对于函数的长度和理解复杂度都有着非常严格的限制,在默认情况下函数的行数不能超过 10 行,理解复杂度也不能超过 7,除此之外,Rubocop 其实还有其他的复杂度限制,例如循环复杂度(CyclomaticComplexity),这些复杂度的限制都是为了保证函数的简单和容易理解。

组织方式

如何对测试进行组织也是一个值得讨论的话题,Golang 中的单元测试文件和代码都是与源代码放在同一个目录下按照 package 进行组织的,server.go 文件对应的测试代码应该放在同一目录下的 server_test.go 文件中。

如果文件不是以 _test.go 结尾,当我们运行 go test ./pkg 时就不会找到该文件中的测试用例,其中的代码也就不会被执行,这也是 Go 语言对于测试组织方法的一个约定。

Test

单元测试的最常见以及默认组织方式就是写在以 _test.go 结尾的文件中,所有的测试方法也都是以 Test 开头并且只接受一个 testing.T 类型的参数:

func TestAuthor(t *testing.T) {
    author := blog.Author()
    assert.Equal(t, "draveness", author)
}

如果我们要给函数名为 Add 的方法写单元测试,那么对应的测试方法一般会被写成 TestAdd,为了同时测试多个分支的内容,我们可以通过以下的方式组织 Add 函数相关的测试:

func TestAdd(t *testing.T) {
    assert.Equal(t, 5, Add(2, 3))
}

func TestAddWithNegativeNumber(t *testing.T) {
    assert.Equal(t, -2, Add(-1, -1))
}

除了这种将一个函数相关的测试分散到多个 Test 方法之外,我们可以使用 for 循环来减少重复的测试代码,这在逻辑比较复杂的测试中会非常好用,能够减少大量的重复代码,不过也需要我们小心地进行设计:

func TestAdd(t *testing.T) {
    tests := []struct{
        name     string
        first    int64
        second   int64
        expected int64
    } {
        {
            name:     "HappyPath":
            first:    2,
            second:   3,
            expected: 5,
        },
        {
            name:     "NegativeNumber":
            first:    -1,
            second:   -1,
            expected: -2,
        },
    }
    
    for _, test := range tests {
        t.Run(test.name, func(t *testing.T) {
            assert.Equal(t, test.expected, Add(test.first, test.second))
        })
    }
}

这种方式其实也能生成树形的测试结果,将 Add 相关的测试分成一组方便我们进行观察和理解,不过这种测试组织方法需要我们保证测试代码的通用性,当函数依赖的上下文较多时往往需要我们写很多的 if/else 条件判断语句影响我们对测试的快速理解。

作者通常会在测试代码比较简单时使用第一种组织方式,而在依赖较多、函数功能较为复杂时使用第二种方式,不过这也不是定论,我们需要根据实际情况决定如何对测试进行设计。

Suite

第二种比较常见的方式是按照簇进行组织,其实就是对 Go 语言默认的测试方式进行简单的封装,我们可以使用 stretchr/testify 中的 suite 包对测试进行组织:

import (
    "testing"
    "github.com/stretchr/testify/suite"
)

type ExampleTestSuite struct {
    suite.Suite
    VariableThatShouldStartAtFive int
}

func (suite *ExampleTestSuite) SetupTest() {
    suite.VariableThatShouldStartAtFive = 5
}

func (suite *ExampleTestSuite) TestExample() {
    suite.Equal(suite.VariableThatShouldStartAtFive, 5)
}

func TestExampleTestSuite(t *testing.T) {
    suite.Run(t, new(ExampleTestSuite))
}

我们可以使用 suite 包,以结构体的方式对测试簇进行组织,suite 提供的 SetupTest/SetupSuite 和 TearDownTest/TearDownSuite 是执行测试前后以及执行测试簇前后的钩子方法,我们能在其中完成一些共享资源的初始化,减少测试中的初始化代码。

BDD

最后一种组织代码的方式就是使用 BDD 的风格对单元测试进行组织,ginkgo 就是 Golang 社区最常见的 BDD 框架了,这里提到的行为驱动开发(BDD)和测试驱动开发(TDD)都是一种保证工程质量的方法论。想要在项目中实践这种思想还是需要一些思维上的转变和适应,也就是先通过写单元测试或者行为测试约定方法的 Spec,再实现方法让我们的测试通过,这是一种比较科学的方法,它能为我们带来比较强的信心。

我们虽然不一定要使用 BDD/TDD 的思想对项目进行开发,但是却可以使用 BDD 的风格方式组织非常易读的测试代码:

var _ = Describe("Book", func() {
    var (
        book Book
        err error
    )

    BeforeEach(func() {
        book, err = NewBookFromJSON(`{
            "title":"Les Miserables",
            "author":"Victor Hugo",
            "pages":1488
        }`)
    })

    Describe("loading from JSON", func() {
        Context("when the JSON fails to parse", func() {
            BeforeEach(func() {
                book, err = NewBookFromJSON(`{
                    "title":"Les Miserables",
                    "author":"Victor Hugo",
                    "pages":1488oops
                }`)
            })

            It("should return the zero-value for the book", func() {
                Expect(book).To(BeZero())
            })

            It("should error", func() {
                Expect(err).To(HaveOccurred())
            })
        })
    })
})

BDD 框架中一般都包含 DescribeContext 以及 It 等代码块,其中 Describe 的作用是描述代码的独立行为、Context 是在一个独立行为中的多个不同上下文,最后的 It 用于描述期望的行为,这些代码块最终都构成了类似『描述……,当……时,它应该……』的句式帮助我们快速地理解测试代码。

Mock 方法

项目中的单元测试应该是稳定的并且不依赖任何的外部项目,它只是对项目中函数和方法的测试,所以我们需要在单元测试中对所有的第三方的不稳定依赖进行 Mock,也就是模拟这些第三方服务的接口;除此之外,为了简化一次单元测试的上下文,在同一个项目中我们也会对其他模块进行 Mock,模拟这些依赖模块的返回值。

单元测试的核心就是隔离依赖并验证输入和输出的正确性,Go 语言作为一个静态语言提供了比较少的运行时特性,这也让我们在 Go 语言中 Mock 依赖变得非常困难。

Mock 的主要作用就是保证待测试方法依赖的上下文固定,在这时无论我们对当前方法运行多少次单元测试,如果业务逻辑不改变,它都应该返回完全相同的结果,在具体介绍 Mock 的不同方法之前,我们首先要清楚一些常见的依赖,一个函数或者方法的常见依赖可以有以下几种:

  1. 接口
  2. 数据库
  3. HTTP 请求
  4. Redis、缓存以及其他依赖

这些不同的场景基本涵盖了写单元测试时会遇到的情况,我们会在接下来的内容中分别介绍如何处理以上几种不同的依赖。

接口

首先要介绍的其实就是 Go 语言中最常见也是最通用的 Mock 方法,也就是能够对接口进行 Mock 的 golang/mock 框架,它能够根据接口生成 Mock 实现,假设我们有以下代码:

package blog

type Post struct {}

type Blog interface {
	ListPosts() []Post
}

type jekyll struct {}

func (b *jekyll) ListPosts() []Post {
 	return []Post{}
}

type wordpress struct{}

func (b *wordpress) ListPosts() []Post {
	return []Post{}
}

我们的博客可能使用 jekyll 或者 wordpress 作为引擎,但是它们都会提供 ListsPosts 方法用于返回全部的文章列表,在这时我们就需要定义一个 Post 接口,接口要求遵循 Blog 的结构体必须实现 ListPosts 方法。

golang-interface-blog-example

当我们定义好了 Blog 接口之后,上层 Service 就不再需要依赖某个具体的博客引擎实现了,只需要依赖 Blog接口就可以完成对文章的批量获取功能:

package service

type Service interface {
	ListPosts() ([]Post, error)
}

type service struct {
    blog blog.Blog
}

func NewService(b blog.Blog) *Service {
    return &service{
        blog: b,
    }
}

func (s *service) ListPosts() ([]Post, error) {
    return s.blog.ListPosts(), nil
}

如果我们想要对 Service 进行测试,我们就可以使用 gomock 提供的 mockgen 工具命令生成 MockBlog 结构体,使用如下所示的命令:

$ mockgen -package=mblog -source=pkg/blog/blog.go > test/mocks/blog/blog.go

$ cat test/mocks/blog/blog.go
// Code generated by MockGen. DO NOT EDIT.
// Source: blog.go

// Package mblog is a generated GoMock package.
...
// NewMockBlog creates a new mock instance
func NewMockBlog(ctrl *gomock.Controller) *MockBlog {
	mock := &MockBlog{ctrl: ctrl}
	mock.recorder = &MockBlogMockRecorder{mock}
	return mock
}

// EXPECT returns an object that allows the caller to indicate expected use
func (m *MockBlog) EXPECT() *MockBlogMockRecorder {
	return m.recorder
}

// ListPosts mocks base method
func (m *MockBlog) ListPosts() []Post {
	m.ctrl.T.Helper()
	ret := m.ctrl.Call(m, "ListPosts")
	ret0, _ := ret[0].([]Post)
	return ret0
}

// ListPosts indicates an expected call of ListPosts
func (mr *MockBlogMockRecorder) ListPosts() *gomock.Call {
	mr.mock.ctrl.T.Helper()
	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListPosts", reflect.TypeOf((*MockBlog)(nil).ListPosts))
}

这段 mockgen 生成的代码非常长的,所以我们只展示了其中的一部分,它的功能就是帮助我们验证任意接口的输入参数并且模拟接口的返回值;而在生成 Mock 实现的过程中,作者总结了一些可以分享的经验:

  1. 在 test/mocks 目录中放置所有的 Mock 实现,子目录与接口所在文件的二级目录相同,在这里源文件的位置在 pkg/blog/blog.go,它的二级目录就是 blog/,所以对应的 Mock 实现会被生成到 test/mocks/blog/ 目录中;
  2. 指定 package 为 mxxx,默认的 mock_xxx 看起来非常冗余,上述 blog 包对应的 Mock 包也就是 mblog
  3. mockgen 命令放置到 Makefile 中的 mock 下统一管理,减少祖传命令的出现; mock: rm -rf test/mocks mkdir -p test/mocks/blog mockgen -package=mblog -source=pkg/blog/blog.go > test/mocks/blog/blog.go

当我们生成了上述的 Mock 实现代码之后,就可以使用如下的方式为 Service 写单元测试了,这段代码通过 NewMockBlog 生成一个 Blog 接口的 Mock 实现,然后通过 EXPECT 方法控制该实现会在调用 ListPosts 时返回空的 Post 数组:


func TestListPosts(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

 	mockBlog := mblog.NewMockBlog(ctrl)
 	mockBlog.EXPECT().ListPosts().Return([]Post{})
  
 	service := NewService(mockBlog)
  
 	assert.Equal(t, []Post{}, service.ListPosts())
}

由于当前 Service 只依赖于 Blog 的实现,所以在这时我们就能够断言当前方法一定会返回 []Post{},这时我们的方法的返回值就只与传入的参数有关(虽然 ListPosts 方法没有入参),我们能够减少一次关注的上下文并保证测试的稳定和可信。

这是 Go 语言中最标准的单元测试写法,所有依赖的 package 无论是项目内外都应该使用这种方式处理(在有接口的情况下),如果没有接口 Go 语言的单元测试就会非常难写,这也是为什么从项目中是否有接口就能判断工程质量的原因了。

SQL

另一个项目中比较常见的依赖其实就是数据库,在遇到数据库的依赖时,我们一般都会使用 sqlmock 来模拟数据库的连接,当我们使用 sqlmock 时会写出如下所示的单元测试:

func (s *suiteServerTester) TestRemovePost() {
	entry := pb.Post{
		Id: 1,
	}

	rows := sqlmock.NewRows([]string{"id", "author"}).AddRow(1, "draveness")

	s.Mock.ExpectQuery(`SELECT (.+) FROM "posts"`).WillReturnRows(rows)
	s.Mock.ExpectExec(`DELETE FROM "posts"`).
		WithArgs(1).
		WillReturnResult(sqlmock.NewResult(1, 1))

	response, err := s.server.RemovePost(context.Background(), &entry)

	s.NoError(err)
	s.EqualValues(response, &entry)
	s.NoError(s.Mock.ExpectationsWereMet())
}

最常用的几个方法就是 ExpectQuery 和 ExpectExec,前者主要用于模拟 SQL 的查询语句,后者用于模拟 SQL 的增删,从上面的实例中我们可以看到这个这两种方法的使用方式,建议各位先阅读相关的 文档 再尝试使用。

HTTP

HTTP 请求也是我们在项目中经常会遇到的依赖,httpmock 就是一个用于 Mock 所有 HTTP 依赖的包,它使用模式匹配的方式匹配 HTTP 请求的 URL,在匹配到特定的请求时就会返回预先设置好的响应。

func TestFetchArticles(t *testing.T) {
	httpmock.Activate()
	defer httpmock.DeactivateAndReset()

	httpmock.RegisterResponder("GET", "https://api.mybiz.com/articles",
		httpmock.NewStringResponder(200, `[{"id": 1, "name": "My Great Article"}]`))

	httpmock.RegisterResponder("GET", `=~^https://api\.mybiz\.com/articles/id/\d+\z`,
		httpmock.NewStringResponder(200, `{"id": 1, "name": "My Great Article"}`))

	...
}

如果遇到 HTTP 请求的依赖时,就可以使用上述 httpmock 包模拟依赖的 HTTP 请求。

猴子补丁

最后要介绍的猴子补丁其实就是一个大杀器了,bouk/monkey 能够通过替换函数指针的方式修改任意函数的实现,所以如果上述的几种方法都不能满足我们的需求,我们就只能够通过猴子补丁这种比较 hack 的方法 Mock 依赖了:

func main() {
	monkey.Patch(fmt.Println, func(a ...interface{}) (n int, err error) {
		s := make([]interface{}, len(a))
		for i, v := range a {
			s[i] = strings.Replace(fmt.Sprint(v), "hell", "*bleep*", -1)
		}
		return fmt.Fprintln(os.Stdout, s...)
	})
	fmt.Println("what the hell?") // what the *bleep*?
}

然而这种方法的使用其实有一些限制,由于它是在运行时替换了函数的指针,所以如果遇到一些简单的函数,例如 rand.Int63n 和 time.Now,编译器可能会直接将这种函数内联到调用实际发生的代码处并不会调用原有的方法,所以使用这种方式往往需要我们在测试时额外指定 -gcflags=-l 禁止编译器的内联优化。

$ go test -gcflags=-l ./...

bouk/monkey 的 README 对于它的使用给出了一些注意事项,除了内联编译之外,我们需要注意的是不要在单元测试之外的地方使用猴子补丁,我们应该只在必要的时候使用这种方法,例如依赖的第三方库没有提供 interface 或者修改 time.Now 以及 rand.Int63n 等内置函数的返回值用于测试时。

从理论上来说,通过猴子补丁这种方式我们能够在运行时 Mock Go 语言中的一切函数,这也为我们提供了单元测试 Mock 依赖的最终解决方案。

断言

在最后,我们简单介绍一下辅助单元测试的 assert 包,它提供了非常多的断言方法帮助我们快速对期望的返回值进行测试,减少我们的工作量:

func TestSomething(t *testing.T) {
  assert.Equal(t, 123, 123, "they should be equal")

  assert.NotEqual(t, 123, 456, "they should not be equal")

  assert.Nil(t, object)

  if assert.NotNil(t, object) {
    assert.Equal(t, "Something", object.Value)
  }
}

在这里我们也是简单展示一下 assert 的示例,更详细的内容可以阅读它的相关文档,在这里也就不多做展示了。

小结

如果之前完全没有写过单元测试或者没有写过 Go 语言的单元测试,相信这篇文章已经给了足够多的上下文帮助我们开始做这件事情,我们要知道的是单元测试其实并不会阻碍我们的开发进度,它能够为我们的上线提供信心,也是质量保证上投资回报率最高的方法。

学习写好单元测试一定会有一些学习曲线和不适应,甚至会在短期内影响我们的开发效率,但是熟悉了这一套流程和接口之后,单元测试对我们的帮助会非常大,每一个单元测试都表示一个业务逻辑,每次提交时执行单元测试就能够帮助我们确定新的代码大概率上不会影响已有的业务逻辑,能够明显地降低重构的风险以及线上事故的数量

总结

在这篇文章中我们从三个方面分别介绍了如何写优雅的 Go 语言代码,作者尽可能地给出了最容易操作和最有效的方法:

  • 代码规范:使用辅助工具帮助我们在每次提交 PR 时自动化地对代码进行检查,减少工程师人工审查的工作量;
  • 最佳实践
    • 目录结构:遵循 Go 语言社区中被广泛达成共识的 目录结构,减少项目的沟通成本;
    • 模块拆分:按照职责对不同的模块进行拆分,Go 语言的项目中也不应该出现 modelcontroller 这种违反语言顶层设计思路的包名;
    • 显示与隐式:尽可能地消灭项目中的 init 函数,保证显式地进行方法的调用以及错误的处理;
    • 面向接口:面向接口是 Go 语言鼓励的开发方式,也能够为我们写单元测试提供方便,我们应该遵循固定的模式对外提供功能;
      1. 使用大写的 Service 对外暴露方法;
      2. 使用小写的 service 实现接口中定义的方法;
      3. 通过 func NewService(...) (Service, error) 函数初始化 Service 接口;
  • 单元测试:保证项目工程质量的最有效办法;
    • 可测试:意味着面向接口编程以及减少单个函数中包含的逻辑,使用『小方法』;
    • 组织方式:使用 Go 语言默认的 Test 框架、开源的 suite 或者 BDD 的风格对单元测试进行合理组织;
    • Mock 方法:四种不同的单元测试 Mock 方法;
      • gomock:最标准的也是最被鼓励的方式;
      • sqlmock:处理依赖的数据库;
      • httpmock:处理依赖的 HTTP 请求;
      • monkey:万能的方法,但是只在万不得已时使用,类似的代码写起来非常冗长而且不直观;
    • 断言:使用社区的 testify 快速验证方法的返回值;

想要写出优雅的代码本身就不是一件容易的事情,它需要我们不断地对自己的知识体系进行更新和优化,推倒之前的经验并对项目持续进行完善和重构,而只有真正经过思考和设计的代码才能够经过时间的检验(代码是需要不断重构的),随意堆砌代码的行为是不能鼓励也不应该发生的,每一行代码都应该按照最高的标准去设计和开发,这是我们保证工程质量的唯一方法。

作者也一直在努力学习如何写出更加优雅的代码,写出好的代码真的不是一件容易的事情,作者也希望能通过这篇文章帮助使用 Go 语言的工程师写出更有 Golang 风格的项目。

出镜率比较高的Golang项目。

记录用的人较多或者出镜率比较高的Golang项目。

如果我漏了你觉得重要的项目,麻烦帮我指出,我尽快补上,谢谢!

下面列表中的每一个项目都配上了Star增长趋势的图片,可以看出该项目的热度。

特别是与区块链相关的两个项目fabricgo-ethereum,能够明显看出区块链行业的起伏。

整理过程中收获:

1、了解到docker项目已经改名为moby,当前仓库为:https://github.com/moby/moby

2、了解到时间序列数据库,库中每一个数据都有时间属性。

项目列表

Gin

仓库地址:https://github.com/gin-gonic/gin

Gin是用Go语言实现的一款web框架。

它的特点和Martini类似,但是API的性能更好,大概快40倍。如果你对性能要求极高,尝试一下Gin,不会让你失望。

Beego

仓库地址:https://github.com/astaxie/beego

一个使用 Go 的思维来帮助您构建并开发 Go 应用程序的开源框架。

一个快速开发 Go 应用的 HTTP 框架,可以用来快速开发 API、Web 及后端服务等各种应用,是一个 RESTful 的框架,主要设计灵感来源于 tornado、sinatra 和 flask 这三个框架,但是结合了 Go 本身的一些特性(interface、struct 嵌入等)而设计的一个框架。

框架特性:简单化智能化模块化高性能

Caddy:

仓库地址:https://github.com/mholt/caddy

一款可以用于生产的开源服务器,具有速度快,易使用,生产效率高的特点。

当前已经可以在WindowsMacLinuxBSDSolaris, and Android使用。

具有如下特点:

  1. 使用Caddyfile方便配置
  2. Auto HTTPS Caddy 使用 Let’s Encrypt 让你的站点全自动变成全站HTTPS,无需任何配置。当然你想使用自己的证书也行。
  3. HTTP/2 全自动支持HTTP/2协议,无需任何配置。
  4. 主机虚拟化使多个站点工作
  5. 可使用插件扩展
  6. 无需依赖即可运行
  7. 为了保证安全连接,使用了TLS session ticket key rotation

Nsq

仓库地址:https://github.com/nsqio/nsq

实时分发的消息平台,用于极大规模的数据处理,处理量级10亿+。

它提升了分布式和去中心化的拓扑结构,没有单点故障,支持容错和高可用性,并保证消息传递的可靠性。

在操作上,NSQ易于配置和部署(所有参数都在命令行上指定,编译后的二进制文件没有运行时依赖项)。为了获得最大的灵活性,它与数据格式无关(消息可以是JSON、MSGPack、协议缓冲区或其他任何格式)。官方的go和python库是现成的(以及许多其他客户机库),如果您有兴趣构建自己的库,这就是一个协议规范。

Hugo

仓库地址:https://github.com/gohugoio/hugo

一个静态的,可伸缩的网页生成器,宣称世界上最快的建站框架,不过这点和wordpress怎么比呢。

Go语言编写的静态网站生成器,速度快,易用,可配置。

Hugo获取一个包含内容和模板的目录,并将其呈现为完整的HTML网站。

Gogs

仓库地址:https://github.com/gogs/gogs

Gogs是一款极易搭建的自助Git服务。

该项目旨在打造一个以最简便的方式搭建简单、稳定和可扩展的自助Git服务。

使用Go语言开发使得Gogs能够通过独立的二进制分发,并且支持Go语言支持的 所有平台,包括 Linux、macOS、Windows 以及 ARM 平台。

Frp

仓库地址:https://github.com/fatedier/frp

frp是一个可用于内网穿透的高性能的反向代理应用,支持 tcp, udp 协议,为 http 和 https 应用协议提供了额外的能力,且尝试性支持了点对点穿透。

Proxypool

仓库地址:https://github.com/henson/proxypool

采集免费的代理资源为爬虫提供有效的IP代理

设计架构:

Getter:代理获取接口,目前有9个免费代理源,每调用一次就会抓取这些网站最新的100个代理放入Channel,可自行添加额外的代理获取接口;

Channel:临时存放采集来的代理,通过访问稳定的网站去验证代理的有效性,有效则存入数据库;

Schedule:用定时的计划任务去检测数据库中代理IP的可用性,删除不可用的代理。同时也会主动通过Getter去获取最新代理;

Api:代理池的访问接口,提供get接口输出JSON,方便爬虫直接使用。

Lantern

仓库地址:https://github.com/getlantern/lantern

区别于,SS,它是分布式的,点对点的,通过蓝灯,你可以和自由上网的用户共享网络,对方可以自由上网,你也就自由了。

SS-go

仓库地址:https://github.com/SS/SS-go

该项目为SS的Go语言实现,项目名称使用SS替代,你懂吧?

Syncthing

仓库地址:https://github.com/syncthing/syncthing

Syncthing是一个持续不断的文件同步项目,能够在两台或者多台电脑上同步文件,使用了其独有的对等自由块交换协议,速度极快。

主要特点:

确保数据的安全性:保护用户的数据是责无旁贷,该项目采取所有的合理的预防措施来避免用户的文件损坏。

确保数据不被攻击:不循序任何未经授权方的窃听或修改。

易于使用

自动化

能够在大多数通用的电脑上使用

Kubernetes

仓库地址:https://github.com/kubernetes/kubernetes

容器编排工具,实现自动化部署,更新,下线,负载均衡,容错处理等。

三个特点:

优化部署:快速而有预期地部署你的应用, 极速地扩展你的应用,增加项目的实例,能够实现自动布局、自动重启、自动复制、自动伸缩,并实现应用的状态检查自我修复

优化资源利用:跨主机编排容器, 更充分地利用硬件资源来最大化地满足企业应用的需求

声明式配置etcd声明式的容器管理,保证所部署的应用按照我们部署的方式运作.

Etcd

仓库地址:https://github.com/etcd-io/etcd

etcd

分布式可靠的键值存储,尤其是分布式系统中极其重要的数据,其特点:

Simple: API设计合理,面向用户

Secure: 自动的TLS连接,支持客户定制认证

Fast: 写入能力大于1w+每秒

Reliable: 使用Raft恰当的分发

etcd当前频繁的和Kubernetes,locksmith,vulcandDoorman等项目配合使用。

Moby

仓库地址:https://github.com/moby/moby

该项目是在容器化生态中组装容器时使用,以前的大名叫做:docker,这个大家都知道。后来经过一段纠结的时刻,改名字了,原因在这儿

docker

Moby是一个开放式项目,旨在维持模块化和灵活性。

模块化:该项目包括的许多组件,优秀的函数和API共同协作。

可交换:Moby包含足够的组件来构建功能齐全的容器系统,但其模块化架构确保大部分组件可以通过不同的实现进行交换。

可用安全性:Moby提供安全的缺省值,无需特殊配置。

Traefik

仓库地址:https://github.com/containous/traefik

Traefik是一款开源的反向代理与负载均衡工具。它最大的优点是能够与常见的微服务系统直接整合,可以实现自动化动态配置。 目前支持 Docker、Swarm、Mesos/Marathon、 Mesos、Kubernetes、Consul、Etcd、Zookeeper、BoltDB、Rest API 等等后端模型。

Influxdb

仓库地址:https://github.com/influxdata/influxdb

influxdb是目前比较流行的时间序列数据库。

时间序列数据库:数据格式里包含Timestamp字段的数据,几乎所有的数据其实都可以打上一个Timestamp字段。

Influxdb是一个开源的分布式时序、时间和指标数据库,使用go语言编写,无需外部依赖。

三大特性:

时序性(Time Series):与时间相关的函数的灵活使用(诸如最大、最小、求和等);

度量(Metrics):对实时大量数据进行计算;

事件(Event):任意事件的数据我们都可以做操作。

influxdb

Prometheus

仓库地址: https://github.com/prometheus/prometheus

一个开源的服务监控系统和时间序列数据库。

Prometheus提供的是一整套监控体系, 包括数据的采集,数据存储,报警,甚至是绘图(只不过很烂,官方也推荐使用 grafana)。 而InfluxDB只是一个时序数据库。同为时间序列数据库,两者对比:prometheus和influxdb对比

Grafana

仓库地址:https://github.com/grafana/grafana

Grafana是一款开源的,具有丰富功能的度量标准仪表板和图形编辑器,用于显示Graphite,Elasticsearch,OpenTSDB,Prometheus和InfluxDB等数据,定制化高。

Go-ethereum

仓库地址:https://github.com/ethereum/go-ethereum

以太坊协议使用Go语言的官方实现。

Fabric

仓库地址:https://github.com/hyperledger/fabric

区块链超级账本Hyperledger Fabric实现,用于联盟链开发。

Drone

仓库地址:https://github.com/drone/drone

Drone是一种基于容器技术的持续交付系统。

Drone使用简单的YAML配置文件(docker-compose的超集)来定义和执行Docker容器中的Pipelines

Drone与流行的源代码管理系统无缝集成,包括GitHub,GitHub Enterprise,Bitbucket等。

欢迎关注公号:程序员的金融圈

作者:大漠胡萝卜
链接:https://juejin.im/post/5cfa2cfef265da1bcc19333e
来源:掘金
著作权归作者所有。商业转载请联系作者获得授权,非商业转载请注明出处。

高可用分布式存储 etcd 的实现原理

原始链接:https://draveness.me/etcd-introduction

在上一篇文章 详解分布式协调服务 ZooKeeper 中,我们介绍过分布式协调服务 Zookeeper 的实现原理以及应用,今天想要介绍的 etcd 其实也是在生产环境中经常被使用的协调服务,它与 Zookeeper 一样,也能够为整个集群提供服务发现、配置以及分布式协调的功能。

etcd-logo

这篇文章将会介绍 etcd 的实现原理,其中包括 Raft 协议、存储两大模块,在最后我们也会简单介绍 etcd 一些具体应用场景。

简介

etcd 的官方将它定位成一个可信赖的分布式键值存储服务,它能够为整个分布式集群存储一些关键数据,协助分布式集群的正常运转。

etcd-keywords

我们可以简单看一下 etcd 和 Zookeeper 在定义上有什么不同:

  • etcd is a distributed reliable key-value store for the most critical data of a distributed system…
  • ZooKeeper is a centralized service for maintaining configuration information, naming, providing distributed synchronization, and providing group services.

其中前者是一个用于存储关键数据的键值存储,后者是一个用于管理配置等信息的中心化服务。

etcd 的使用其实非常简单,它对外提供了 gRPC 接口,我们可以通过 Protobuf 和 gRPC 直接对 etcd 中存储的数据进行管理,也可以使用官方提供的 etcdctl 操作存储的数据。

service KV {
  rpc Range(RangeRequest) returns (RangeResponse) {
      option (google.api.http) = {
        post: "/v3beta/kv/range"
        body: "*"
    };
  }

  rpc Put(PutRequest) returns (PutResponse) {
      option (google.api.http) = {
        post: "/v3beta/kv/put"
        body: "*"
    };
  }
}

文章并不会展开介绍 etcd 的使用方法,这一小节将逐步介绍几大核心模块的实现原理,包括 etcd 使用 Raft 协议同步各个节点数据的过程以及 etcd 底层存储数据使用的结构。

Raft

在每一个分布式系统中,etcd 往往都扮演了非常重要的地位,由于很多服务配置发现以及配置的信息都存储在 etcd 中,所以整个集群可用性的上限往往就是 etcd 的可用性,而使用 3 ~ 5 个 etcd 节点构成高可用的集群往往都是常规操作。

etcd-cluste

正是因为 etcd 在使用的过程中会启动多个节点,如何处理几个节点之间的分布式一致性就是一个比较有挑战的问题了。

解决多个节点数据一致性的方案其实就是共识算法,在之前的文章中我们简单介绍过 Zookeeper 使用的 Zab 协议 以及常见的 共识算法 Paxos 和 Raft,etcd 使用的共识算法就是 Raft,这一节我们将详细介绍 Raft 以及 etcd 中 Raft 的一些实现细节。

介绍

Raft 从一开始就被设计成一个易于理解和实现的共识算法,它在容错和性能上与 Paxos 协议比较类似,区别在于它将分布式一致性的问题分解成了几个子问题,然后一一进行解决。

每一个 Raft 集群中都包含多个服务器,在任意时刻,每一台服务器只可能处于 Leader、Follower 以及 Candidate 三种状态;在处于正常的状态时,集群中只会存在一个 Leader,其余的服务器都是 Follower。

raft-server-states

上述图片修改自 In Search of an Understandable Consensus Algorithm 一文 5.1 小结中图四。

所有的 Follower 节点都是被动的,它们不会主动发出任何的请求,只会响应 Leader 和 Candidate 发出的请求,对于每一个用户的可变操作,都会被路由给 Leader 节点进行处理,除了 Leader 和 Follower 节点之外,Candidate 节点其实只是集群运行过程中的一个临时状态。

Raft 集群中的时间也被切分成了不同的几个任期(Term),每一个任期都会由 Leader 的选举开始,选举结束后就会进入正常操作的阶段,直到 Leader 节点出现问题才会开始新一轮的选择。

raft-terms

每一个服务器都会存储当前集群的最新任期,它就像是一个单调递增的逻辑时钟,能够同步各个节点之间的状态,当前节点持有的任期会随着每一个请求被传递到其他的节点上。

Raft 协议在每一个任期的开始时都会从一个集群中选出一个节点作为集群的 Leader 节点,这个节点会负责集群中的日志的复制以及管理工作。

raft-subproblems

我们将 Raft 协议分成三个子问题:节点选举、日志复制以及安全性,文章会以 etcd 为例介绍 Raft 协议是如何解决这三个子问题的。

节点选举

使用 Raft 协议的 etcd 集群在启动节点时,会遵循 Raft 协议的规则,所有节点一开始都被初始化为 Follower 状态,新加入的节点会在 NewNode 中做一些配置的初始化,包括用于接收各种信息的 Channel:

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/node.go#L190-225
func StartNode(c *Config, peers []Peer) Node {
	r := newRaft(c)
	r.becomeFollower(1, None)
	r.raftLog.committed = r.raftLog.lastIndex()
	for _, peer := range peers {
		r.addNode(peer.ID)
	}

	n := newNode()
	go n.run(r)
	return &n
}

在做完这些初始化的节点和 Raft 配置的事情之后,就会进入一个由 for 和 select 组成的超大型循环,这个循环会从 Channel 中获取待处理的事件:

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/node.go#L291-423
func (n *node) run(r *raft) {
	lead := None

	for {
		if lead != r.lead {
			lead = r.lead
		}

		select {
		case m := <-n.recvc:
			r.Step(m)
		case <-n.tickc:
			r.tick()
		case <-n.stop:
			close(n.done)
			return
		}
	}
}

作者对整个循环内的代码进行了简化,因为当前只需要关心三个 Channel 中的消息,也就是用于接受其他节点消息的 recvc、用于触发定时任务的 tickc 以及用于暂停当前节点的 stop

raft-etcd

除了 stop Channel 中介绍到的消息之外,recvc 和 tickc 两个 Channel 中介绍到事件时都会交给当前节点持有 Raft 结构体处理。

定时器与心跳

当节点从任意状态(包括启动)调用 becomeFollower 时,都会将节点的定时器设置为 tickElection

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L636-643
func (r *raft) tickElection() {
	r.electionElapsed++

	if r.promotable() && r.pastElectionTimeout() {
		r.electionElapsed = 0
		r.Step(pb.Message{From: r.id, Type: pb.MsgHup})
	}
}

如果当前节点可以成为 Leader 并且上一次收到 Leader 节点的消息或者心跳已经超过了等待的时间,当前节点就会发送 MsgHup 消息尝试开始新的选举。

但是如果 Leader 节点正常运行,就能够同样通过它的定时器 tickHeartbeat 向所有的 Follower 节点广播心跳请求,也就是 MsgBeat 类型的 RPC 消息:

func (r *raft) tickHeartbeat() {
	r.heartbeatElapsed++
	r.electionElapsed++

	if r.heartbeatElapsed >= r.heartbeatTimeout {
		r.heartbeatElapsed = 0
		r.Step(pb.Message{From: r.id, Type: pb.MsgBeat})
	}
}

上述代码段 Leader 节点中调用的 Step 函数,最终会调用 stepLeader 方法,该方法会根据消息的类型进行不同的处理:

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L931-1142
func stepLeader(r *raft, m pb.Message) error {
	switch m.Type {
	case pb.MsgBeat:
		r.bcastHeartbeat()
		return nil
	// ...
	}

	//...
}

bcastHeartbeat 方法最终会向所有的 Follower 节点发送 MsgHeartbeat 类型的消息,通知它们目前 Leader 的存活状态,重置所有 Follower 持有的超时计时器。

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L518-534
func (r *raft) sendHeartbeat(to uint64, ctx []byte) {
	commit := min(r.getProgress(to).Match, r.raftLog.committed)
	m := pb.Message{
		To:      to,
		Type:    pb.MsgHeartbeat,
		Commit:  commit,
		Context: ctx,
	}

	r.send(m)
}

作为集群中的 Follower,它们会在 stepFollower 方法中处理接收到的全部消息,包括 Leader 节点发送的心跳 RPC 消息:

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L1191-1247
func stepFollower(r *raft, m pb.Message) error {
	switch m.Type {
	case pb.MsgHeartbeat:
		r.electionElapsed = 0
		r.lead = m.From
		r.handleHeartbeat(m)
	// ...
	}
	return nil
}

当 Follower 接受到了来自 Leader 的 RPC 消息 MsgHeartbeat 时,会将当前节点的选举超时时间重置并通过 handleHeartbeat 向 Leader 节点发出响应 —— 通知 Leader 当前节点能够正常运行。

而 Candidate 节点对于 MsgHeartBeat 消息的处理会稍有不同,它会先执行 becomeFollower 设置当前节点和 Raft 协议的配置:

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L1146-1189
func stepCandidate(r *raft, m pb.Message) error {
  // ...
	switch m.Type {
	case pb.MsgHeartbeat:
		r.becomeFollower(m.Term, m.From) // always m.Term == r.Term
		r.handleHeartbeat(m)
	}
  // ...
	return nil
}

Follower 与 Candidate 会根据节点类型的不同做出不同的响应,两者收到心跳请求时都会重置节点的选举超时时间,不过后者会将节点的状态直接转变成 Follower:

raft-heartbeat

当 Leader 节点收到心跳的响应时就会将对应节点的状态设置为 Active,如果 Follower 节点在一段时间内没有收到来自 Leader 节点的消息就会尝试发起竞选。

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L636-643
func (r *raft) tickElection() {
	r.electionElapsed++

	if r.promotable() && r.pastElectionTimeout() {
		r.electionElapsed = 0
		r.Step(pb.Message{From: r.id, Type: pb.MsgHup})
	}
}

到了这里,心跳机制就起到了作用开始发送 MsgHup 尝试重置整个集群中的 Leader 节点,接下来我们就会开始分析 Raft 协议中的竞选流程了。

竞选流程

如果集群中的某一个 Follower 节点长时间内没有收到来自 Leader 的心跳请求,当前节点就会通过 MsgHup 消息进入预选举或者选举的流程。

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L785-927
func (r *raft) Step(m pb.Message) error {
  // ...

	switch m.Type {
	case pb.MsgHup:
		if r.state != StateLeader {
			if r.preVote {
				r.campaign(campaignPreElection)
			} else {
				r.campaign(campaignElection)
			}
		} else {
			r.logger.Debugf("%x ignoring MsgHup because already leader", r.id)
		}
	}
  // ...
  return nil
}

如果收到 MsgHup 消息的节点不是 Leader 状态,就会根据当前集群的配置选择进入 PreElection 或者 Election 阶段,PreElection 阶段并不会真正增加当前节点的 Term,它的主要作用是得到当前集群能否成功选举出一个 Leader 的答案,如果当前集群中只有两个节点而且没有预选举阶段,那么这两个节点的 Term 会无休止的增加,预选举阶段就是为了解决这一问题而出现的。

raft-cluster-states

在这里不会讨论预选举的过程,而是将目光主要放在选举阶段,具体了解一下使用 Raft 协议的 etcd 集群是如何从众多节点中选出 Leader 节点的。

我们可以继续来分析 campaign 方法的具体实现,下面就是删去预选举相关逻辑后的代码:

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L730-766
func (r *raft) campaign(t CampaignType) {
	r.becomeCandidate()
	
	if r.quorum() == r.poll(r.id, voteRespMsgType(voteMsg), true) {
		r.becomeLeader()
		return
	}
	for id := range r.prs {
		if id == r.id {
			continue
		}

		r.send(pb.Message{Term: r.Term, To: id, Type: pb.MsgVote, Index: r.raftLog.lastIndex(), LogTerm: r.raftLog.lastTerm(), Context: ctx})
	}
}

当前节点会立刻调用 becomeCandidate 将当前节点的 Raft 状态变成候选人;在这之后,它会将票投给自己,如果当前集群只有一个节点,该节点就会直接成为集群中的 Leader 节点。

如果集群中存在了多个节点,就会向集群中的其他节点发出 MsgVote 消息,请求其他节点投票,在 Step 函数中包含不同状态的节点接收到消息时的响应:

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L785-927
func (r *raft) Step(m pb.Message) error {
  // ...

	switch m.Type {
	case pb.MsgVote, pb.MsgPreVote:
		canVote := r.Vote == m.From || (r.Vote == None && r.lead == None)
		if canVote && r.raftLog.isUpToDate(m.Index, m.LogTerm) {
			r.send(pb.Message{To: m.From, Term: m.Term, Type: pb.MsgVoteResp})
			r.electionElapsed = 0
			r.Vote = m.From
		} else {
			r.send(pb.Message{To: m.From, Term: r.Term, Type: pb.MsgVoteResp, Reject: true})
		}

	}
  // ...
  return nil
}

如果当前节点投的票就是消息的来源或者当前节点没有投票也没有 Leader,那么就会向来源的节点投票,否则就会通知该节点当前节点拒绝投票。

raft-election

在 stepCandidate 方法中,候选人节点会处理来自其他节点的投票响应消息,也就是 MsgVoteResp

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L1146-1189
func stepCandidate(r *raft, m pb.Message) error {
	switch m.Type {
	// ...
	case pb.MsgVoteResp:
		gr := r.poll(m.From, m.Type, !m.Reject)
		switch r.quorum() {
		case gr:
			r.becomeLeader()
			r.bcastAppend()
		// ...
		}
	}
	return nil
}

每当收到一个 MsgVoteResp 类型的消息时,就会设置当前节点持有的 votes 数组,更新其中存储的节点投票状态并返回投『同意』票的人数,如果获得的票数大于法定人数 quorum,当前节点就会成为集群的 Leader 并向其他的节点发送当前节点当选的消息,通知其余节点更新 Raft 结构体中的 Term 等信息。

节点状态

对于每一个节点来说,它们根据不同的节点状态会对网络层发来的消息做出不同的响应,我们会分别介绍下面的四种状态在 Raft 中对于配置和消息究竟是如何处理的。

raft-node-states

对于每一个 Raft 的节点状态来说,它们分别有三个比较重要的区别,其中一个是在改变状态时调用 becomeLeaderbecomeCandidatebecomeFollower 和 becomePreCandidate 方法改变 Raft 状态有比较大的不同,第二是处理消息时调用 stepLeaderstepCandidate 和 stepFollower 时有比较大的不同,最后是几种不同状态的节点具有功能不同的定时任务。

对于方法的详细处理,我们在这一节中不详细介绍和分析,如果一个节点的状态是 Follower,那么当前节点切换到 Follower 一定会通过 becomeFollower 函数,在这个函数中会重置节点持有任期,并且设置处理消息的函数为 stepFollower

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L671-678
func (r *raft) becomeFollower(term uint64, lead uint64) {
	r.step = stepFollower
	r.reset(term)
	r.tick = r.tickElection
	r.lead = lead
	r.state = StateFollower
}

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L636-643
func (r *raft) tickElection() {
	r.electionElapsed++

	if r.promotable() && r.pastElectionTimeout() {
		r.electionElapsed = 0
		r.Step(pb.Message{From: r.id, Type: pb.MsgHup})
	}
}

除此之外,它还会设置一个用于在 Leader 节点宕机时触发选举的定时器 tickElection

Candidate 状态的节点与 Follower 的配置差不了太多,只是在消息处理函数 step、任期以及状态上的设置有一些比较小的区别:

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L680-691
func (r *raft) becomeCandidate() {
	r.step = stepCandidate
	r.reset(r.Term + 1)
	r.tick = r.tickElection
	r.Vote = r.id
	r.state = StateCandidate
}

最后的 Leader 就与这两者有其他的区别了,它不仅设置了处理消息的函数 step 而且设置了与其他状态完全不同的 tick 函数:

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L708-728
func (r *raft) becomeLeader() {
	r.step = stepLeader
	r.reset(r.Term)
	r.tick = r.tickHeartbeat
	r.lead = r.id
	r.state = StateLeader

	r.pendingConfIndex = r.raftLog.lastIndex()
	r.appendEntry(pb.Entry{Data: nil})
}

这里的 tick 函数 tickHeartbeat 每隔一段时间会通过 Step 方法向集群中的其他节点发送 MsgBeat 消息:

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L646-669
func (r *raft) tickHeartbeat() {
	r.heartbeatElapsed++
	r.electionElapsed++

	if r.electionElapsed >= r.electionTimeout {
		r.electionElapsed = 0
		if r.checkQuorum {
			r.Step(pb.Message{From: r.id, Type: pb.MsgCheckQuorum})
		}
	} 

	if r.heartbeatElapsed >= r.heartbeatTimeout {
		r.heartbeatElapsed = 0
		r.Step(pb.Message{From: r.id, Type: pb.MsgBeat})
	}
}

上述代码中的 MsgBeat 消息会在 Step 中被转换成 MsgHeartbeat 最终发送给其他的节点,Leader 节点超时之后的选举流程我们在前两节中也已经介绍过了,在这里就不再重复了。

存储

etcd 目前支持 V2 和 V3 两个大版本,这两个版本在实现上有比较大的不同,一方面是对外提供接口的方式,另一方面就是底层的存储引擎,V2 版本的实例是一个纯内存的实现,所有的数据都没有存储在磁盘上,而 V3 版本的实例就支持了数据的持久化。

etcd-storage

在这一节中,我们会介绍 V3 版本的 etcd 究竟是通过什么样的方式存储用户数据的。

后端

在 V3 版本的设计中,etcd 通过 backend 后端这一设计,很好地封装了存储引擎的实现细节,为上层提供一个更一致的接口,对于 etcd 的其他模块来说,它们可以将更多注意力放在接口中的约定上,不过在这里,我们更关注的是 etcd 对 Backend 接口的实现。

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/backend/backend.go#L51-69
type Backend interface {
	ReadTx() ReadTx
	BatchTx() BatchTx

	Snapshot() Snapshot
	Hash(ignores map[IgnoreKey]struct{}) (uint32, error)
	Size() int64
	SizeInUse() int64
	Defrag() error
	ForceCommit()
	Close() error
}

etcd 底层默认使用的是开源的嵌入式键值存储数据库 bolt,但是这个项目目前的状态已经是归档不再维护了,如果想要使用这个项目可以使用 CoreOS 的 bbolt 版本。

boltdb-logo

这一小节中,我们会简单介绍 etcd 是如何使用 BoltDB 作为底层存储的,首先可以先来看一下 pacakge 内部的 backend 结构体,这是一个实现了 Backend 接口的结构:

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/backend/backend.go#L80-104
type backend struct {
	size int64
	sizeInUse int64
	commits int64

	mu sync.RWMutex
	db *bolt.DB

	batchInterval time.Duration
	batchLimit    int
	batchTx       *batchTxBuffered

	readTx *readTx

	stopc chan struct{}
	donec chan struct{}

	lg *zap.Logger
}

从结构体的成员 db 我们就可以看出,它使用了 BoltDB 作为底层存储,另外的两个 readTx 和 batchTx 分别实现了 ReadTx 和 BatchTx 接口:

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/backend/read_tx.go#L30-36
type ReadTx interface {
	Lock()
	Unlock()

	UnsafeRange(bucketName []byte, key, endKey []byte, limit int64) (keys [][]byte, vals [][]byte)
	UnsafeForEach(bucketName []byte, visitor func(k, v []byte) error) error
}

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/backend/batch_tx.go#L28-38
type BatchTx interface {
	ReadTx
	UnsafeCreateBucket(name []byte)
	UnsafePut(bucketName []byte, key []byte, value []byte)
	UnsafeSeqPut(bucketName []byte, key []byte, value []byte)
	UnsafeDelete(bucketName []byte, key []byte)
	Commit()
	CommitAndStop()
}

从这两个接口的定义,我们不难发现它们能够对外提供数据库的读写操作,而 Backend 就能对这两者提供的方法进行封装,为上层屏蔽存储的具体实现:

etcd-backends

每当我们使用 newBackend 创建一个新的 backend 结构时,都会创建一个 readTx 和 batchTx 结构体,这两者一个负责处理只读请求,一个负责处理读写请求:

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/backend/backend.go#L137-176
func newBackend(bcfg BackendConfig) *backend {
	bopts := &bolt.Options{}
	bopts.InitialMmapSize = bcfg.mmapSize()
	db, _ := bolt.Open(bcfg.Path, 0600, bopts)

	b := &backend{
		db: db,
		batchInterval: bcfg.BatchInterval,
		batchLimit:    bcfg.BatchLimit,
		readTx: &readTx{
			buf: txReadBuffer{
				txBuffer: txBuffer{make(map[string]*bucketBuffer)},
			},
			buckets: make(map[string]*bolt.Bucket),
		},
		stopc: make(chan struct{}),
		donec: make(chan struct{}),
	}
	b.batchTx = newBatchTxBuffered(b)
	go b.run()
	return b
}

当我们在 newBackend 中进行了初始化 BoltDB、事务等工作后,就会开一个 goroutine 异步的对所有批量读写事务进行定时提交:

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/backend/backend.go#L289-305
func (b *backend) run() {
	defer close(b.donec)
	t := time.NewTimer(b.batchInterval)
	defer t.Stop()
	for {
		select {
		case <-t.C:
		case <-b.stopc:
			b.batchTx.CommitAndStop()
			return
		}
		if b.batchTx.safePending() != 0 {
			b.batchTx.Commit()
		}
		t.Reset(b.batchInterval)
	}
}

对于上层来说,backend 其实只是对底层存储的一个抽象,很多时候并不会直接跟它打交道,往往都是使用它持有的 ReadTx 和 BatchTx 与数据库进行交互。

只读事务

目前大多数的数据库对于只读类型的事务并没有那么多的限制,尤其是在使用了 MVCC 之后,所有的只读请求几乎不会被写请求锁住,这大大提升了读的效率,由于在 BoltDB 的同一个 goroutine 中开启两个相互依赖的只读事务和读写事务会发生死锁,为了避免这种情况我们还是引入了 sync.RWLock 保证死锁不会出现:

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/backend/read_tx.go#L38-47
type readTx struct {
	mu  sync.RWMutex
	buf txReadBuffer

	txmu    sync.RWMutex
	tx      *bolt.Tx
	buckets map[string]*bolt.Bucket
}

你可以看到在整个结构体中,除了用于保护 tx 的 txmu 读写锁之外,还存在另外一个 mu 读写锁,它的作用是保证 buf 中的数据不会出现问题,buf 和结构体中的 buckets 都是用于加速读效率的缓存。

etcd-backend-tx

对于一个只读事务来说,它对上层提供了两个获取存储引擎中数据的接口,分别是 UnsafeRange 和 UnsafeForEach,在这里会重点介绍前面方法的实现细节:

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/backend/read_tx.go#L52-90
func (rt *readTx) UnsafeRange(bucketName, key, endKey []byte, limit int64) ([][]byte, [][]byte) {
	if endKey == nil {
		limit = 1
	}
	keys, vals := rt.buf.Range(bucketName, key, endKey, limit)
	if int64(len(keys)) == limit {
		return keys, vals
	}

	bn := string(bucketName)
	bucket, ok := rt.buckets[bn]
	if !ok {
		bucket = rt.tx.Bucket(bucketName)
		rt.buckets[bn] = bucket
	}

	if bucket == nil {
		return keys, vals
	}
	c := bucket.Cursor()

	k2, v2 := unsafeRange(c, key, endKey, limit-int64(len(keys)))
	return append(k2, keys...), append(v2, vals...)
}

上述代码中省略了加锁保护读缓存以及 Bucket 中存储数据的合法性,也省去了一些参数的检查,不过方法的整体接口还是没有太多变化,UnsafeRange 会先从自己持有的缓存 txReadBuffer 中读取数据,如果数据不能够满足调用者的需求,就会从 buckets 缓存中查找对应的 BoltDB bucket 并从 BoltDB 数据库中读取。

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/backend/batch_tx.go#L121-141
func unsafeRange(c *bolt.Cursor, key, endKey []byte, limit int64) (keys [][]byte, vs [][]byte) {
	var isMatch func(b []byte) bool
	if len(endKey) > 0 {
		isMatch = func(b []byte) bool { return bytes.Compare(b, endKey) < 0 }
	} else {
		isMatch = func(b []byte) bool { return bytes.Equal(b, key) }
		limit = 1
	}

	for ck, cv := c.Seek(key); ck != nil && isMatch(ck); ck, cv = c.Next() {
		vs = append(vs, cv)
		keys = append(keys, ck)
		if limit == int64(len(keys)) {
			break
		}
	}
	return keys, vs
}

这个包内部的函数 unsafeRange 实际上通过 BoltDB 中的游标来遍历满足查询条件的键值对。

到这里为止,整个只读事务提供的接口就基本介绍完了,在 etcd 中无论我们想要后去单个 Key 还是一个范围内的 Key 最终都是通过 Range 来实现的,这其实也是只读事务的最主要功能。

读写事务

只读事务只提供了读数据的能力,包括 UnsafeRange 和 UnsafeForeach,而读写事务 BatchTx 提供的就是读和写数据的能力了:

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/backend/batch_tx.go#L40-46
type batchTx struct {
	sync.Mutex
	tx      *bolt.Tx
	backend *backend

	pending int
}

读写事务同时提供了不带缓存的 batchTx 实现以及带缓存的 batchTxBuffered 实现,后者其实『继承了』前者的结构体,并额外加入了缓存 txWriteBuffer 加速读请求:

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/backend/batch_tx.go#L243-246
type batchTxBuffered struct {
	batchTx
	buf txWriteBuffer
}

后者在实现接口规定的方法时,会直接调用 batchTx 的同名方法,并将操作造成的副作用的写入的缓存中,在这里我们并不会展开介绍这一版本的实现,还是以分析 batchTx 的方法为主。

当我们向 etcd 中写入数据时,最终都会调用 batchTx 的 unsafePut 方法将数据写入到 BoltDB 中:

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/backend/batch_tx.go#L65-67
func (t *batchTx) UnsafePut(bucketName []byte, key []byte, value []byte) {
	t.unsafePut(bucketName, key, value, false)
}

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/backend/batch_tx.go#L74-103
func (t *batchTx) unsafePut(bucketName []byte, key []byte, value []byte, seq bool) {
	bucket := t.tx.Bucket(bucketName)
	if err := bucket.Put(key, value); err != nil {
		plog.Fatalf("cannot put key into bucket (%v)", err)
	}
	t.pending++
}

这两个方法的实现非常清晰,作者觉得他们都并不值得展开详细介绍,只是调用了 BoltDB 提供的 API 操作一下 bucket 中的数据,而另一个删除方法的实现与这个也差不多:

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/backend/batch_tx.go#L144-169
func (t *batchTx) UnsafeDelete(bucketName []byte, key []byte) {
	bucket := t.tx.Bucket(bucketName)
	err := bucket.Delete(key)
	if err != nil {
		plog.Fatalf("cannot delete key from bucket (%v)", err)
	}
	t.pending++
}

它们都是通过 Bolt.Tx 找到对应的 Bucket,然后做出相应的增删操作,但是这写请求在这两个方法执行后其实并没有提交,我们还需要手动或者等待 etcd 自动将请求提交:

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/backend/batch_tx.go#L184-188
func (t *batchTx) Commit() {
	t.Lock()
	t.commit(false)
	t.Unlock()
}

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/backend/batch_tx.go#L210-241
func (t *batchTx) commit(stop bool) {
	if t.tx != nil {
		if t.pending == 0 && !stop {
			return
		}

		start := time.Now()

		err := t.tx.Commit()

		rebalanceSec.Observe(t.tx.Stats().RebalanceTime.Seconds())
		spillSec.Observe(t.tx.Stats().SpillTime.Seconds())
		writeSec.Observe(t.tx.Stats().WriteTime.Seconds())
		commitSec.Observe(time.Since(start).Seconds())
		atomic.AddInt64(&t.backend.commits, 1)

		t.pending = 0
	}
	if !stop {
		t.tx = t.backend.begin(true)
	}
}

在每次调用 Commit 对读写事务进行提交时,都会先检查是否有等待中的事务,然后会将数据上报至 Prometheus 中,其他的服务就可以将 Prometheus 作为数据源对 etcd 的执行状况进行监控了。

索引

经常使用 etcd 的开发者可能会了解到,它本身对于每一个键值对都有一个 revision 的概念,键值对的每一次变化都会被 BoltDB 单独记录下来,所以想要在存储引擎中获取某一个 Key 对应的值,要先获取 revision,再通过它才能找到对应的值,在里我们想要介绍的其实是 etcd 如何管理和存储一个 Key 的多个 revision 记录。

B-Tree

在 etcd 服务中有一个用于存储所有的键值对 revision 信息的 btree,我们可以通过 index 的 Get 接口获取一个 Key 对应 Revision 的值:

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/index.go#L68-76
func (ti *treeIndex) Get(key []byte, atRev int64) (modified, created revision, ver int64, err error) {
	keyi := &keyIndex{key: key}
	if keyi = ti.keyIndex(keyi); keyi == nil {
		return revision{}, revision{}, 0, ErrRevisionNotFound
	}
	return keyi.get(ti.lg, atRev)
}

上述方法通过 keyIndex 方法查找 Key 对应的 keyIndex 结构体,这里使用的内存结构体 btree 是 Google 实现的一个版本:

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/index.go#L84-89
func (ti *treeIndex) keyIndex(keyi *keyIndex) *keyIndex {
	if item := ti.tree.Get(keyi); item != nil {
		return item.(*keyIndex)
	}
	return nil
}

可以看到这里的实现非常简单,只是从 treeIndex 持有的成员 btree 中查找 keyIndex,将结果强制转换成 keyIndex 类型后返回;获取 Key 对应 revision 的方式也非常简单:

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/key_index.go#L149-171
func (ki *keyIndex) get(lg *zap.Logger, atRev int64) (modified, created revision, ver int64, err error) {
	g := ki.findGeneration(atRev)
	if g.isEmpty() {
		return revision{}, revision{}, 0, ErrRevisionNotFound
	}

	n := g.walk(func(rev revision) bool { return rev.main > atRev })
	if n != -1 {
		return g.revs[n], g.created, g.ver - int64(len(g.revs)-n-1), nil
	}

	return revision{}, revision{}, 0, ErrRevisionNotFound
}

KeyIndex

在我们具体介绍方法实现的细节之前,首先我们需要理解 keyIndex 包含的字段以及管理同一个 Key 不同版本的方式:

etcd-keyindex

每一个 keyIndex 结构体中都包含当前键的值以及最后一次修改对应的 revision 信息,其中还保存了一个 Key 的多个 generation,每一个 generation 都会记录当前 Key『从生到死』的全部过程,每当一个 Key 被删除时都会调用 timestone 方法向当前的 generation 中追加一个新的墓碑版本:

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/key_index.go#L127-145
func (ki *keyIndex) tombstone(lg *zap.Logger, main int64, sub int64) error {
	if ki.generations[len(ki.generations)-1].isEmpty() {
		return ErrRevisionNotFound
	}
	ki.put(lg, main, sub)
	ki.generations = append(ki.generations, generation{})
	return nil
}

这个 tombstone 版本标识这当前的 Key 已经被删除了,但是在每次删除一个 Key 之后,就会在当前的 keyIndex 中创建一个新的 generation 结构用于存储新的版本信息,其中 ver 记录当前 generation 包含的修改次数,created 记录创建 generation 时的 revision 版本,最后的 revs 用于存储所有的版本信息。

读操作

etcd 中所有的查询请求,无论是查询一个还是多个、是数量还是键值对,最终都会调用 rangeKeys 方法:

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/kvstore_txn.go#L112-165
func (tr *storeTxnRead) rangeKeys(key, end []byte, curRev int64, ro RangeOptions) (*RangeResult, error) {
	rev := ro.Rev

	revpairs := tr.s.kvindex.Revisions(key, end, rev)
	if len(revpairs) == 0 {
		return &RangeResult{KVs: nil, Count: 0, Rev: curRev}, nil
	}

	kvs := make([]mvccpb.KeyValue, int(ro.Limit))
	revBytes := newRevBytes()
	for i, revpair := range revpairs[:len(kvs)] {
		revToBytes(revpair, revBytes)
		_, vs := tr.tx.UnsafeRange(keyBucketName, revBytes, nil, 0)
		kvs[i].Unmarshal(vs[0])
	}
	return &RangeResult{KVs: kvs, Count: len(revpairs), Rev: curRev}, nil
}

为了获取一个范围内的所有键值对,我们首先需要通过 Revisions 函数从 btree 中获取范围内所有的 keyIndex

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/index.go#L106-120
func (ti *treeIndex) Revisions(key, end []byte, atRev int64) (revs []revision) {
	if end == nil {
		rev, _, _, err := ti.Get(key, atRev)
		if err != nil {
			return nil
		}
		return []revision{rev}
	}
	ti.visit(key, end, func(ki *keyIndex) {
		if rev, _, _, err := ki.get(ti.lg, atRev); err == nil {
			revs = append(revs, rev)
		}
	})
	return revs
}

如果只需要获取一个 Key 对应的版本,就是直接使用 treeIndex 的方法,但是当上述方法会从 btree 索引中获取一个连续多个 revision 值时,就会调用 keyIndex.get 来遍历整颗树并选取合适的版本:

func (ki *keyIndex) get(lg *zap.Logger, atRev int64) (modified, created revision, ver int64, err error) {
	g := ki.findGeneration(atRev)
	if g.isEmpty() {
		return revision{}, revision{}, 0, ErrRevisionNotFound
	}

	n := g.walk(func(rev revision) bool { return rev.main > atRev })
	if n != -1 {
		return g.revs[n], g.created, g.ver - int64(len(g.revs)-n-1), nil
	}

	return revision{}, revision{}, 0, ErrRevisionNotFound
}

因为每一个 Key 的 keyIndex 中其实都存储着多个 generation,我们需要根据传入的参数返回合适的 generation 并从其中返回主版本大于 atRev 的 revision 结构。

对于上层的键值存储来说,它会利用这里返回的 revision 从真正存储数据的 BoltDB 中查询当前 Key 对应 revision 的结果。

写操作

当我们向 etcd 中插入数据时,会使用传入的 key 构建一个 keyIndex 结构体并从树中获取相关版本等信息:

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/index.go#L53-66
func (ti *treeIndex) Put(key []byte, rev revision) {
	keyi := &keyIndex{key: key}

	item := ti.tree.Get(keyi)
	if item == nil {
		keyi.put(ti.lg, rev.main, rev.sub)
		ti.tree.ReplaceOrInsert(keyi)
		return
	}
	okeyi := item.(*keyIndex)
	okeyi.put(ti.lg, rev.main, rev.sub)
}

treeIndex.Put 在获取内存中的 keyIndex 结构之后会通过 keyIndex.put 其中加入新的 revision

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/key_index.go#L77-104
func (ki *keyIndex) put(lg *zap.Logger, main int64, sub int64) {
	rev := revision{main: main, sub: sub}

	if len(ki.generations) == 0 {
		ki.generations = append(ki.generations, generation{})
	}
	g := &ki.generations[len(ki.generations)-1]
	if len(g.revs) == 0 {
		g.created = rev
	}
	g.revs = append(g.revs, rev)
	g.ver++
	ki.modified = rev
}

每一个新 revision 结构体写入 keyIndex 时,都会改变当前 generation 的 created 和 ver 等参数,从这个方法中我们就可以了解到 generation 中的各个成员都是如何被写入的。

写入的操作除了增加之外,删除某一个 Key 的函数也会经常被调用:

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/kvstore_txn.go#L252-309
func (tw *storeTxnWrite) delete(key []byte) {
	ibytes := newRevBytes()
	idxRev := revision{main: tw.beginRev + 1, sub: int64(len(tw.changes))}
	revToBytes(idxRev, ibytes)

	ibytes = appendMarkTombstone(tw.storeTxnRead.s.lg, ibytes)

	kv := mvccpb.KeyValue{Key: key}

	d, _ := kv.Marshal()

	tw.tx.UnsafeSeqPut(keyBucketName, ibytes, d)
	tw.s.kvindex.Tombstone(key, idxRev)
	tw.changes = append(tw.changes, kv)
}

正如我们在文章前面所介绍的,删除操作会向结构体中的 generation 追加一个新的 tombstone 标记,用于标识当前的 Key 已经被删除;除此之外,上述方法还会将每一个更新操作的 revision 存到单独的 keyBucketName 中。

索引的恢复

因为在 etcd 中,所有的 keyIndex 都是在内存的 btree 中存储的,所以在启动服务时需要从 BoltDB 中将所有的数据都加载到内存中,在这里就会初始化一个新的 btree 索引,然后调用 restore 方法开始恢复索引:

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/kvstore.go#L321-433
func (s *store) restore() error {
	min, max := newRevBytes(), newRevBytes()
	revToBytes(revision{main: 1}, min)
	revToBytes(revision{main: math.MaxInt64, sub: math.MaxInt64}, max)

	tx := s.b.BatchTx()

	rkvc, revc := restoreIntoIndex(s.lg, s.kvindex)
	for {
		keys, vals := tx.UnsafeRange(keyBucketName, min, max, int64(restoreChunkKeys))
		if len(keys) == 0 {
			break
		}
		restoreChunk(s.lg, rkvc, keys, vals, keyToLease)
		newMin := bytesToRev(keys[len(keys)-1][:revBytesLen])
		newMin.sub++
		revToBytes(newMin, min)
	}
	close(rkvc)
	s.currentRev = <-revc

	return nil
}

在恢复索引的过程中,有一个用于遍历不同键值的『生产者』循环,其中由 UnsafeRange 和 restoreChunk 两个方法构成,这两个方法会从 BoltDB 中遍历数据,然后将键值对传到 rkvc 中,交给 restoreIntoIndex 方法中创建的 goroutine 处理:

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/kvstore.go#L486-506
func restoreChunk(lg *zap.Logger, kvc chan<- revKeyValue, keys, vals [][]byte, keyToLease map[string]lease.LeaseID) {
	for i, key := range keys {
		rkv := r evKeyValue{key: key}
		_ := rkv.kv.Unmarshal(vals[i])
		rkv.kstr = string(rkv.kv.Key)
		if isTombstone(key) {
			delete(keyToLease, rkv.kstr)
		} else if lid := lease.LeaseID(rkv.kv.Lease); lid != lease.NoLease {
			keyToLease[rkv.kstr] = lid
		} else {
			delete(keyToLease, rkv.kstr)
		}
		kvc <- rkv
	}
}

先被调用的 restoreIntoIndex 方法会创建一个用于接受键值对的 Channel,在这之后会在一个 goroutine 中处理从 Channel 接收到的数据,并将这些数据恢复到内存里的 btree 中:

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/kvstore.go#L441-484
func restoreIntoIndex(lg *zap.Logger, idx index) (chan<- revKeyValue, <-chan int64) {
	rkvc, revc := make(chan revKeyValue, restoreChunkKeys), make(chan int64, 1)
	go func() {
		currentRev := int64(1)
		defer func() { revc <- currentRev }()
		for rkv := range rkvc {
			ki = &keyIndex{key: rkv.kv.Key}
			ki := idx.KeyIndex(ki)

			rev := bytesToRev(rkv.key)
			currentRev = rev.main
			if ok {
				if isTombstone(rkv.key) {
					ki.tombstone(lg, rev.main, rev.sub)
					continue
				}
				ki.put(lg, rev.main, rev.sub)
			} else if !isTombstone(rkv.key) {
				ki.restore(lg, revision{rkv.kv.CreateRevision, 0}, rev, rkv.kv.Version)
				idx.Insert(ki)
			}
		}
	}()
	return rkvc, revc
}

恢复内存索引的相关代码在实现上非常值得学习,两个不同的函数通过 Channel 进行通信并使用 goroutine 处理任务,能够很好地将消息的『生产者』和『消费者』进行分离。

etcd-restore-index

Channel 作为整个恢复索引逻辑的一个消息中心,它将遍历 BoltDB 中的数据和恢复索引两部分代码进行了分离。

存储

etcd 的 mvcc 模块对外直接提供了两种不同的访问方式,一种是键值存储 kvstore,另一种是 watchableStore 它们都实现了包内公开的 KV 接口:

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/kv.go#L100-125
type KV interface {
	ReadView
	WriteView

	Read() TxnRead
	Write() TxnWrite

	Hash() (hash uint32, revision int64, err error)
	HashByRev(rev int64) (hash uint32, revision int64, compactRev int64, err error)

	Compact(rev int64) (<-chan struct{}, error)
	Commit()
	Restore(b backend.Backend) error
	Close() error
}

kvstore

对于 kvstore 来说,其实没有太多值得展开介绍的地方,它利用底层的 BoltDB 等基础设施为上层提供最常见的增伤改查,它组合了下层的 readTxbatchTx 等结构体,将一些线程不安全的操作变成线程安全的。

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/kvstore_txn.go#L32-40
func (s *store) Read() TxnRead {
	s.mu.RLock()
	tx := s.b.ReadTx()
	s.revMu.RLock()
	tx.Lock()
	firstRev, rev := s.compactMainRev, s.currentRev
	s.revMu.RUnlock()
	return newMetricsTxnRead(&storeTxnRead{s, tx, firstRev, rev})
}

它也负责对内存中 btree 索引的维护以及压缩一些无用或者不常用的数据,几个对外的接口 ReadWrite 就是对 readTxbatchTx 等结构体的组合并将它们的接口暴露给其他的模块。

watchableStore

另外一个比较有意思的存储就是 watchableStore 了,它是 mvcc 模块为外界提供 Watch 功能的接口,它负责了注册、管理以及触发 Watcher 的功能,我们先来看一下这个结构体的各个字段:

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/watchable_store.go#L45-65
type watchableStore struct {
	*store

	mu sync.RWMutex

	unsynced watcherGroup
	synced watcherGroup

	stopc chan struct{}
	wg    sync.WaitGroup
}

每一个 watchableStore 其实都组合了来自 store 结构体的字段和方法,除此之外,还有两个 watcherGroup类型的字段,其中 unsynced 用于存储未同步完成的实例,synced 用于存储已经同步完成的实例。

在初始化一个新的 watchableStore 时,我们会创建一个用于同步watcherGroup 的 Goroutine,在 syncWatchersLoop 这个循环中会每隔 100ms 调用一次 syncWatchers 方法,将所有未通知的事件通知给所有的监听者,这可以说是整个模块的核心:

func (s *watchableStore) syncWatchers() int {
	curRev := s.store.currentRev
	compactionRev := s.store.compactMainRev

	wg, minRev := s.unsynced.choose(maxWatchersPerSync, curRev, compactionRev)
	minBytes, maxBytes := newRevBytes(), newRevBytes()
	revToBytes(revision{main: minRev}, minBytes)
	revToBytes(revision{main: curRev + 1}, maxBytes)

	tx := s.store.b.ReadTx()
	revs, vs := tx.UnsafeRange(keyBucketName, minBytes, maxBytes, 0)
	evs := kvsToEvents(nil, wg, revs, vs)

	wb := newWatcherBatch(wg, evs)
	for w := range wg.watchers {
		w.minRev = curRev + 1

		eb, ok := wb[w]
		if !ok {
			s.synced.add(w)
			s.unsynced.delete(w)
			continue
		}

		w.send(WatchResponse{WatchID: w.id, Events: eb.evs, Revision: curRev})

		s.synced.add(w)
		s.unsynced.delete(w)
	}

	return s.unsynced.size()
}

简化后的 syncWatchers 方法中总共做了三件事情,首先是根据当前的版本从未同步的 watcherGroup 中选出一些待处理的任务,然后从 BoltDB 中后去当前版本范围内的数据变更并将它们转换成事件,事件和 watcherGroup 在打包之后会通过 send 方法发送到每一个 watcher 对应的 Channel 中。

etcd-mvcc-watch-module

上述图片中展示了 mvcc 模块对于向外界提供的监听某个 Key 和范围的接口,外部的其他模块会通过 watchStream.watch 函数与模块内部进行交互,每一次调用 watch 方法最终都会向 watchableStore 持有的 watcherGroup 中添加新的 watcher 结构。

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/watcher.go#L108-135
func (ws *watchStream) Watch(id WatchID, key, end []byte, startRev int64, fcs ...FilterFunc) (WatchID, error) {
	if id == AutoWatchID {
		for ws.watchers[ws.nextID] != nil {
			ws.nextID++
		}
		id = ws.nextID
		ws.nextID++
	}

	w, c := ws.watchable.watch(key, end, startRev, id, ws.ch, fcs...)

	ws.cancels[id] = c
	ws.watchers[id] = w
	return id, nil
}

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/mvcc/watchable_store.go#L111-142
func (s *watchableStore) watch(key, end []byte, startRev int64, id WatchID, ch chan<- WatchResponse, fcs ...FilterFunc) (*watcher, cancelFunc) {
	wa := &watcher{
		key:    key,
		end:    end,
		minRev: startRev,
		id:     id,
		ch:     ch,
		fcs:    fcs,
	}

	synced := startRev > s.store.currentRev || startRev == 0
	if synced {
		s.synced.add(wa)
	} else {
		s.unsynced.add(wa)
	}

	return wa, func() { s.cancelWatcher(wa) }
}

当 etcd 服务启动时,会在服务端运行一个用于处理监听事件的 watchServer gRPC 服务,客户端的 Watch 请求最终都会被转发到这个服务的 Watch 函数中:

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/etcdserver/api/v3rpc/watch.go#L136-206
func (ws *watchServer) Watch(stream pb.Watch_WatchServer) (err error) {
	sws := serverWatchStream{
		// ...
		gRPCStream:  stream,
		watchStream: ws.watchable.NewWatchStream(),
		ctrlStream: make(chan *pb.WatchResponse, ctrlStreamBufLen),
	}

	sws.wg.Add(1)
	go func() {
		sws.sendLoop()
		sws.wg.Done()
	}()

	go func() {
		sws.recvLoop()
	}()

	sws.wg.Wait()
	return err
}

当客户端想要通过 Watch 结果监听某一个 Key 或者一个范围的变动,在每一次客户端调用服务端上述方式都会创建两个 Goroutine,这两个协程一个会负责向监听者发送数据变动的事件,另一个协程会负责处理客户端发来的事件。

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/etcdserver/api/v3rpc/watch.go#L220-334 
func (sws *serverWatchStream) recvLoop() error {
	for {
		req, err := sws.gRPCStream.Recv()
		if err == io.EOF {
			return nil
		}
		if err != nil {
			return err
		}

		switch uv := req.RequestUnion.(type) {
		case *pb.WatchRequest_CreateRequest:
			creq := uv.CreateRequest

			filters := FiltersFromRequest(creq)
			wsrev := sws.watchStream.Rev()
			rev := creq.StartRevision
			id, _ := sws.watchStream.Watch(mvcc.WatchID(creq.WatchId), creq.Key, creq.RangeEnd, rev, filters...)
			wr := &pb.WatchResponse{
				Header:   sws.newResponseHeader(wsrev),
				WatchId:  int64(id),
				Created:  true,
				Canceled: err != nil,
			}
			select {
			case sws.ctrlStream <- wr:
			case <-sws.closec:
				return nil
			}

		case *pb.WatchRequest_CancelRequest: // ...
		case *pb.WatchRequest_ProgressRequest: // ...
		default:
			continue
		}
	}
}

在用于处理客户端的 recvLoop 方法中调用了 mvcc 模块暴露出的 watchStream.Watch 方法,该方法会返回一个可以用于取消监听事件的 watchID;当 gRPC 流已经结束后者出现错误时,当前的循环就会返回,两个 Goroutine 也都会结束。

如果出现了更新或者删除事件,就会被发送到 watchStream 持有的 Channel 中,而 sendLoop 会通过 select来监听多个 Channel 中的数据并将接收到的数据封装成 pb.WatchResponse 结构并通过 gRPC 流发送给客户端:

func (sws *serverWatchStream) sendLoop() {
	for {
		select {
		case wresp, ok := <-sws.watchStream.Chan():
			evs := wresp.Events
			events := make([]*mvccpb.Event, len(evs))
			for i := range evs {
				events[i] = &evs[i]			}

			canceled := wresp.CompactRevision != 0
			wr := &pb.WatchResponse{
				Header:          sws.newResponseHeader(wresp.Revision),
				WatchId:         int64(wresp.WatchID),
				Events:          events,
				CompactRevision: wresp.CompactRevision,
				Canceled:        canceled,
			}

			sws.gRPCStream.Send(wr)

		case c, ok := <-sws.ctrlStream: // ...
		case <-progressTicker.C: // ...
		case <-sws.closec:
			return
		}
	}
}

对于每一个 Watch 请求来说,watchServer 会根据请求创建两个用于处理当前请求的 Goroutine,这两个协程会与更底层的 mvcc 模块协作提供监听和回调功能:

etcd-watch-server

到这里,我们对于 Watch 功能的介绍就差不多结束了,从对外提供的接口到底层的使用的数据结构以及具体实现,其他与 Watch 功能相关的话题可以直接阅读 etcd 的源代码了解更加细节的实现。

应用

在上面已经介绍了核心的 Raft 共识算法以及使用的底层存储之后,这一节更想谈一谈 etcd 的一些应用场景,与之前谈到的 分布式协调服务 Zookeeper 一样,etcd 在大多数的集群中还是处于比较关键的位置,工程师往往都会使用 etcd 存储集群中的重要数据和元数据,多个节点之间的强一致性以及集群部署的方式赋予了 etcd 集群高可用性。

我们依然可以使用 etcd 实现微服务架构中的服务发现、发布订阅、分布式锁以及分布式协调等功能,因为虽然它被定义成了一个可靠的分布式键值存储,但是它起到的依然是一个分布式协调服务的作用,这也使我们在需要不同的协调服务中进行权衡和选择。

为什么要在分布式协调服务中选择 etcd 其实是一个比较关键的问题,很多工程师选择 etcd 主要是因为它使用 Go 语言开发、部署简单、社区也比较活跃,但是缺点就在于它相比 Zookeeper 还是一个比较年轻的项目,需要一些时间来成长和稳定。

总结

etcd 的实现原理非常有趣,我们能够在它的源代码中学习很多 Go 编程的最佳实践和设计,这也值得我们去研究它的源代码。

目前很多项目和公司都在生产环境中大规模使用 etcd,这对于社区来说是意见非常有利的事情,如果微服务的大部分技术栈是 Go,作者也更加推荐各位读者在选择分布式协调服务时选择 etcd 作为系统的基础设施。

相关文章

Reference

关于图片和转载

知识共享许可协议

本作品采用知识共享署名 4.0 国际许可协议进行许可。 转载时请注明原文链接,图片在使用时请保留图片中的全部内容,可适当缩放并在引用处附上图片所在的文章链接,图片使用 Sketch 进行绘制。

Go正走在成为下一个企业级编程语言的轨道上

本文copy from tony bai 原文链接https://tonybai.com/2019/05/03/go-is-on-a-trajectory-to-become-the-next-enterprise-programming-language/

发展演化了十年的Go语言已经被证明了是云计算时代的首选编程语言,但Go的用武之地显然不局限于此。Kevin Goslar近期在Hacker Noon发表了一篇名为:《Go is on a Trajectory to Become the Next Enterprise Programming Language》的文章,阐述了Go可能成为下一个企业编程语言的理由,这里是那篇文章的中文译文,分享给大家。

img{512x368}

摘要

Go是一种专门为大规模软件开发而设计的编程语言。它提供了强大的开发体验并避免了现有编程语言存在的许多问题。这些因素使其成为最有可能在未来替代Java主导企业软件平台的候选者之一。对于那些寻求在未来几十年内构建大规模云基础架构的安全和前瞻性技术的公司和开源计划而言,我建议它们将Go视为其主要的编程语言。Go的优势如下:

请继续阅读有关上述每个优势点的更多详细信息。然而,在进入Go之前,你应该注意:

简介

Go是Google开发的一种编程语言,在过去几年中取得了很大的成功。大部分现代云计算,网络和DevOps平台都是Go语言编写的,例如:DockerKubernetesTerraformETCDistio等。许多公司也将它用于通用软件开发。Go所具备的功能让这些项目吸引了大量用户,而Go的易用性也使得这些项目有了很多的贡献者。

Go的优势来自于简单和经过验证的想法的结合,同时避免了其他语言中出现的许多问题。这篇博客文章概述了Go背后的一些设计原则和工程智慧,并展示它们是如何结合在一起的 – 它们使Go成为下一代大型软件开发平台的优秀候选者。许多编程语言在个别领域都比较强大,但是在将所有领域都结合起来时,没有其他语言能够如此一致地“得分”,特别是在大型软件工程方面。

基于现实世界的经验

Go是由经验丰富的软件行业资深人士创建的,他们长期以来一直感受到现有语言的缺点带来的痛苦。几十年前,Rob PikeKen Thompson在Unix,C和Unicode的发明中发挥了重要作用。在实现了用于JavaScript和Java的V8和HotSpot虚拟机之后,Robert Griesemer在编译器和垃圾收集方面拥有着数十年的经验。在太多次的不得不等待他们的谷歌规模的C++/Java代码库的编译过程的推动下,他们开始着手创建一门新的编程语言,这门语言中凝聚了他们通过编写半个世纪代码过程中所学到的一切。

专注于大型工程

几乎任何编程语言都可以成功构建小型工程项目。当成千上万的开发人员在数十年的持续时间压力下在包含数千万行代码的大量代码库上进行协作时,真正痛苦的问题就会发生。这会导致以下问题:

Go专注于减轻这些大规模的工程难题,有时是以使小型工程变得更加繁琐为代价,例如在这里和那里需要一些额外的代码。

专注于可维护性

Go强调尽可能多地将工作转交到自动代码维护工具中。Go工具链提供了最常用的功能,如格式化代码和自动package导入、查找符号的定义和用法、简单的重构以及代码味道的识别。由于标准化的代码格式化和单一的惯用方式,机器生成的代码更改看起来非常接近Go中人为生成的更改。并而使用类似的模式,使得人和机器的协作更加无缝。

保持简单直接

初级程序员为简单问题创建简单的解决方案。高级程序员为复杂问题创建复杂的解决方案。伟大的程序员找到复杂问题的简单解决方案。-  查尔斯康奈尔

很多人都对Go不包含他们喜欢的其他语言概念感到惊讶。Go确实是一种非常小而简单的语言,只包含最少的正交和经过验证的概念。这鼓励开发人员以最少的认知开销编写最简单的代码,以便许多其他人可以理解并使用它。

使事情显式而明显

良好的代码是显而易见的,避免聪明,模糊的语言功能,扭曲的控制流和间接性。

许多语言都致力于使编写代码变得高效。然而,在其生命周期中,人们将花费大约(100倍)的时间阅读代码,而不是首先编写所需的代码。例如,审查,理解,调试,更改,重构或重用它。在查看代码时,通常只能看到并理解它的一小部分,通常没有对整个代码库的完整理解。为了解释这一点,Go将一切都显式化了。

一个例子是错误处理。让异常在各个点中断代码并使沿着调用链处理可能会更容易。Go需要手动处理或返回每个错误。这使得它可以准确地显示代码可以被中断的位置以及如何处理或包装错误。总的来说,这使得错误处理更容易编写,但更容易理解。

简单易学

Go非常小而且简单,可以在短短几天内研究整个语言及其基本概念。根据我们的经验,经过不超过一周的培训(与其他语言的以月为单位相比),初学者可以理解Go专家编写的代码,并为此做出贡献。为了方便大量人群,Go网站提供了所需的所有教程和深入的文章。这些教程在浏览器中运行,允许人们在将Go安装到本地计算机上之前学习和使用Go。

一种做事方式

Go语言通过个人自我表达赋予团队合作能力。

在Go(和Python)中,所有语言特征都是正交的并且彼此互补,通常做某事只有一种方法。如果您要求10位Python或Go程序员解决问题,您将获得10个相对类似的解决方案。不同的程序员在彼此的代码库中感觉更有家的感觉。在查看其他人的代码时,每分钟的WTF更少,而且人们的工作更好地融合在一起,从而形成一个人人都为之骄傲并且喜欢工作的一致性。这避免了大规模的工程问题,例如:

img{512x368}


来源:https://www.osnews.com/story/19266/wtfsm

简单,内置并发

Go专为现代多核硬件而设计。

目前使用的大多数编程语言(Java,JavaScript,Python,Ruby,C,C ++)都是在20世纪80年代到2000年代设计的,当时大多数CPU只有一个计算核心。这就是为什么它们本质上是单线程的,并将并行化视为事后增加的边缘情况,通过诸如线程和同步点之类的附加组件实现,这些附加组件既麻烦又难以正确使用。第三方库提供了更简单的并发形式,如Actor模型,但总有多个选项可用,导致语言生态系统碎片化。今天的硬件拥有越来越多的计算内核,软件必须并行化才能在其上高效运行。Go是在多核CPU时代编写的,并且在语言中内置了简单,高级的CSP风格的并发特性。

面向计算的语言原语

在基础层面上,计算机系统接收数据,处理它(通常经过几个步骤),并输出结果数据。例如,Web服务器从客户端接收HTTP请求,并将其转换为一系列数据库或后端调用。一旦这些调用返回,它就会将接收到的数据转换为HTML或JSON并将其输出给调用者。Go的内置语言原语直接支持这种范例:

由于所有计算原语都是由语言以直接的形式提供的,因此Go源代码可以更直接地表达服务器执行的操作。

OO – 好的部分

img{512x368}


在基类中改变某些东西的副作用

面向对象非常有用。这几十年OO的应用是富有成效的,并且让我们了解它的哪些部分比其他部分可以更好地扩展。基于这些认知,Go采用面向对象的新方法。它保留了封装和消息传递等优点。Go避免了继承,因为它现在被认为是有害的,Go为组合提供头等的支持

现代标准库

许多当前使用的编程语言(Java,JavaScript,Python,Ruby)是在互联网成为当今无处不在的计算平台之前设计的。因此,这些语言的标准库仅为未针对现代互联网优化的网络提供相对通用的支持。Go是十年前创建的,当时互联网已经全面展开。Go的标准库允许在没有第三方库的情况下创建更复杂的网络服务。这可以防止使用第三方库的常见问题:

Russ Cox的更多背景信息。

标准化格式

Gofmt的风格是没有人喜欢的,但gofmt是每个人的最爱。 – Rob Pike

Gofmt是一种以标准化方式格式化Go代码的程序。它不是最漂亮的格式化方式,而是最简单,最不讨厌的方式。标准化的源代码格式化具有惊人的积极影响:

许多其他语言社区现在正在开发gofmt等价物。当构建为第三方解决方案时,通常会有几种竞争格式标准。例如,JavaScript世界提供PrettierStandardJS。可以一起使用其中之一或两者。许多JS项目都没有采用它们,因为这是一个额外的决定。Go的格式化程序内置于该语言的标准工具链中,因此只有一个标准,每个人都在使用它。

快速编译

img{512x368}


来源:https://xkcd.com/303

大型代码库的长编译时间是引发Go语言起源的一个微小的原因。Google主要使用C++和Java,与Haskell,Scala或Rust等更复杂的语言相比,它可以相对快速地编译。尽管如此,当编译大型代码库时,即使是少量的慢速也会把人激怒,编译工作流中断导致编译延迟。Go是从头开始设计的,以使编译更有效,因此编译器速度非常快,几乎没有编译延迟。这为Go开发人员提供了类似于脚本语言的即时反馈,并具有静态类型检查的额外好处。

交叉编译

由于语言运行时非常简单,因此它已被移植到许多平台,如macOS,Linux,Windows,BSD,ARM等。Go可以开箱即用于编译所有这些平台的二进制文件。这使得我们可以轻松地从一台机器来进行部署。

快速执行

Go有着接近C的速度。与JITed(即时编译)语言(Java,JavaScript,Python等)不同,Go二进制文件不需要启动或预热时间,因为它们作为已编译和完全优化的本机代码提供。Go垃圾收集器仅以微秒的指令引入可忽略的暂停。在其快速的单核性能上面,Go使得利用所有的CPU内核更容易

小内存占用

像JVM,Python或Node这样的运行时不仅仅在运行时加载程序代码。它们还会加载大型且高度复杂的基础架构,以便在每次运行时编译和优化程序。这使得它们的启动时间变慢并导致它们使用大量(数百MB)的RAM。Go进程的开销较小,因为它们已经完全编译和优化,只需要运行。Go还以非常节省内存的方式存储数据。这在内存有限且昂贵的云环境中以及在开发期间非常重要,在开发期间我们希望在单个机器上快速启动整个堆栈,同时为其他软件留下内存。

小部署规模

Go二进制文件的大小非常简洁。Go应用程序的Docker镜像通常比用Java或Node编写的等效文件小10倍,因为它不需要包含编译器,JIT,并且需要更少的运行时基础结构。这在部署大型应用程序时很重要。想象一下,将一个简单的应用程序部署到100个生产服务器上 使用Node / JVM时,我们的docker仓库必须提供100个docker镜像@ 200 MB = 20 GB(总共)。这需要镜像仓库耗费一些时间来服务。想象一下,我们希望每天部署100次。使用Go服务时,Docker镜像仓库只需提供100个Docker镜像@ 20 MB = 2 GB。可以更快,更频繁地部署大型Go应用程序,从而允许重要更新更快地实现生产。

自包含部署

Go应用程序部署为包含所有依赖项的单个可执行文件。不需要安装特定版本的JVM,Node或Python运行时。不必将库下载到生产服务器上。不需要对运行Go二进制文件的机器进行任何更改。甚至不需要将Go二进制文件包装到Docker中来共享它们。您只需将Go二进制文件拖放到服务器上,无论该服务器上运行的是什么,它都会在那里运行。上述描述的唯一例外是使用net和os/user包时的动态链接glibc库时。

vendor依赖关系

Go故意避免使用第三方库的中央存储库。Go应用程序直接链接到相应的Git存储库,并将所有相关代码下载(vendor保存)到他们自己的代码库中。这有很多好处:

兼容性保证

Go团队承诺,现有的程序将继续适用于新版本语言。这使得即使是大型项目也可以轻松升级到更新编译器的版本,并从新版本带来的许多性能和安全性改进中受益。同时,由于Go二进制文件包含了他们需要的所有依赖项,因此可以在同一服务器计算机上并行运行使用不同版本的Go编译器编译的二进制文件,而无需进行复杂的设置多个版本的运行时或虚拟化。

文档

在大型工程中,文档对于使软件易于访问和维护非常重要。与其他功能类似,Go中的文档简单实用:

商业支持的开源

当商业实体在公开场合发展时,一些最流行和最全面设计的软件就会发生。这种设置结合了商业软件开发的优势 – 一致性和优化,使系统健壮,可靠,高效 – 具有开放式开发的优势,如来自许多行业的广泛支持,来自多个大型实体和许多用户的支持,以及长期支持,即使商业支持停止。Go就是这样开发的。

缺点

当然,Go并不完美,每种技术选择总是有利有弊。在进入Go之前,这里有一小部分需要考虑的方面。

未成熟

虽然Go的标准库在支持HTTP/2服务器推送等许多新概念方面处于行业领先地位,但与JVM生态系统中存在的相比,用于外部API的第三方Go库可能还不那么成熟。

即将到来的变化

Go团队知道几乎不可能改变现有的语言元素,因此只有在完全开发后才会添加新功能。在经历了10年稳定的故意阶段后,Go团队正在考虑对语言进行一系列更大的改进,作为Go 2.0之旅的一部分。

没有硬实时

虽然Go的垃圾收集器只引入了非常短的中断,但支持硬实时需要没有垃圾收集的技术,例如Rust。

结论

这篇博客文章给出了一些明智的背景知识,但往往没有那么明显的选择进入Go的设计,以及当他们的代码库和团队成数量级增长时,他们将如何从许多痛苦中拯救大型工程项目。总的来说,他们将Go定位为寻求Java之外的现代编程语言的大型开发项目的绝佳选择。