背景
Consul 作为HashiCorp 出品的分布式注册中心和配置中心,是cp模型的,即强调一致性,通过raft协议实现
一致性
consul 一致性支持三种模式,即要强一致还是,最终一致, 可以交个用户选择,这才是一个优秀的分布式系统应该具备的,要了解一致性读,需要先了解consul的三种一致性模式,如下:
- Default
用consul 没有做任何改动的话,大部分都是这个模式工作的,default模式consul考虑了读的一致性还是很高的,读写都是通过leader来处理的,只是一种情况出现脑裂时,可能存在2个leader,在服务,但是老的leader肯定是不能写的,但是有可能服务读,读到过期的数据,但也不是一直会这样,leader有个租约,租约到期这个leader就下线了。
Consistent
consistent 即强一致读,如果是这个模式,consul 每次读请求都要向集群的超过半数的server检查他是不是leader,就比defalut 模式多一次rtt开销,因为即使你是leader,还要请求server确认是否存在其他的leader,这样肯定不会读到过时的数据。Stale
stale是吞吐量最高的模式,但也是一致性最差的模式,所以一致性和吞吐量是矛盾的,因为stale 模式下,consul 集群的任何节点都能服务读请求,意外着即使集群没有leader,还是可以对外提供读请求。
我们了解了consul支持三种一致性模式,你是不是很好奇,consul是怎么实现的呢,我们平时部署一个consul集群也没有让我指定是那一种啥,consul既然是交给用户来选择,所以consul通过api的参数来确定,需要用那种读一致性。
在哪里指定一致性级别
有聪明的同学就会问,说了这么多,我到底在哪里指定这个一致性级别,别急,下面就开始说
consul 通过http 接口提供服务,就在http的api里可以指定,客户端sdk就不说了,有很多版本,这里只说consul agent端,因为线上一般都是直接请求localhost:8500 访问本地的consul agent的。下面是所有consul agent http接口都要执行的一个逻辑parseConsistency
,就是解析一致性
func (s *HTTPServer) parseConsistency(resp http.ResponseWriter, req *http.Request, b structs.QueryOptionsCompat) bool {
query := req.URL.Query()
//这里默认就认为是default模式。
defaults := true
//解析http请求如果带了stale参数,则是允许读过期的数据,那就server不用转发给leader
if _, ok := query["stale"]; ok {
b.SetAllowStale(true)
defaults = false
}
//解析http请求如果带了consistent参数,代表要读最新的数据。
if _, ok := query["consistent"]; ok {
b.SetRequireConsistent(true)
defaults = false
}
//解析http请求如果带了consistent参数,代表要从leader读。
if _, ok := query["leader"]; ok {
defaults = false
}
//解析http请求如果带了cached参数,代表可以从agent读,不需要请求server
if _, ok := query["cached"]; ok {
b.SetUseCache(true)
defaults = false
}
if maxStale := query.Get("max_stale"); maxStale != "" {
dur, err := time.ParseDuration(maxStale)
if err != nil {
resp.WriteHeader(http.StatusBadRequest)
fmt.Fprintf(resp, "Invalid max_stale value %q", maxStale)
return true
}
b.SetMaxStaleDuration(dur)
if dur.Nanoseconds() > 0 {
b.SetAllowStale(true)
defaults = false
}
}
...
上面解析了客户端的读模式,下面看怎么用的,随便看一个consul读的代码,比如查看健康的service node 的一段代码:
//如果可以用cache的数据,则直接从当前agent响应。
if args.QueryOptions.UseCache {
raw, m, err := s.agent.cache.Get(cachetype.HealthServicesName, &args)
if err != nil {
return nil, err
}
defer setCacheMeta(resp, &m)
reply, ok := raw.(*structs.IndexedCheckServiceNodes)
if !ok {
// This should never happen, but we want to protect against panics
return nil, fmt.Errorf("internal error: response type not correct")
}
out = *reply
} else {
//否则需要通过rpc请求server节点。
RETRY_ONCE:
if err := s.agent.RPC("Health.ServiceNodes", &args, &out); err != nil {
return nil, err
}
if args.QueryOptions.AllowStale && args.MaxStaleDuration > 0 && args.MaxStaleDuration < out.LastContact {
args.AllowStale = false
args.MaxStaleDuration = 0
goto RETRY_ONCE
}
}
我们只有指定了cache参数,consul 才会从agent 本地直接响应数据,这里也可以看出,agent 是会缓存数据的,否则就需要请求server节点,这个时候问题又来了,server节点一般我们是一个集群,最少3个节点,那请求那一个呢,有负载均衡吗,带着这个问题,我们看下代码,怎么选server的, 代码如下:
// FindServer takes out an internal "read lock" and searches through the list
// of servers to find a "healthy" server. If the server is actually
// unhealthy, we rely on Serf to detect this and remove the node from the
// server list. If the server at the front of the list has failed or fails
// during an RPC call, it is rotated to the end of the list. If there are no
// servers available, return nil.
func (m *Manager) FindServer() *metadata.Server {
l := m.getServerList()
numServers := len(l.servers)
if numServers == 0 {
m.logger.Warn("No servers available")
return nil
}
// Return whatever is at the front of the list because it is
// assumed to be the oldest in the server list (unless -
// hypothetically - the server list was rotated right after a
// server was added).
return l.servers[0]
}
consul 这里是不是处理的很简单,每次都是取第一个,人家注释也说了,如果这个出现失败了,会移到最后。
Consul Server的逻辑
consul agent 发现不用本地cache的数据,那就要rpc请求server节点,server节点接受到任何请求,都会执行forward方法,来检查是否要转发请求还是就自己响应数据。
func (s *Server) forward(method string, info structs.RPCInfo, args interface{}, reply interface{}) (bool, error) {
var firstCheck time.Time
// Handle DC forwarding
// 检查dc是否一致,不一致就要转发到正确的dc
dc := info.RequestDatacenter()
if dc != s.config.Datacenter {
// Local tokens only work within the current datacenter. Check to see
// if we are attempting to forward one to a remote datacenter and strip
// it, falling back on the anonymous token on the other end.
if token := info.TokenSecret(); token != "" {
done, ident, err := s.ResolveIdentityFromToken(token)
if done {
if err != nil && !acl.IsErrNotFound(err) {
return false, err
}
if ident != nil && ident.IsLocal() {
// Strip it from the request.
info.SetTokenSecret("")
defer info.SetTokenSecret(token)
}
}
}
err := s.forwardDC(method, dc, args, reply)
return true, err
}
// Check if we can allow a stale read, ensure our local DB is initialized
// 这里server开始检查读一致性,如果允许读过期的数据,则直接用当前server的数据。
// 不需要后面的检查是否为leader了。
if info.IsRead() && info.AllowStaleRead() && !s.raft.LastContact().IsZero() {
return false, nil
}
CHECK_LEADER:
// Fail fast if we are in the process of leaving
select {
case <-s.leaveCh:
return true, structs.ErrNoLeader
default:
}
// Find the leader
// 到这里就是要default读或者consistent读,都需要从leader读数据。
isLeader, leader := s.getLeader()
// Handle the case we are the leader
// 如果当前是leader,则不需要再转发到leader了。
if isLeader {
return false, nil
}
// Handle the case of a known leader
// 不是leader,则需要再转发到leader节点,多一次网络请求。
rpcErr := structs.ErrNoLeader
if leader != nil {
rpcErr = s.connPool.RPC(s.config.Datacenter, leader.ShortName, leader.Addr,
leader.Version, method, leader.UseTLS, args, reply)
if rpcErr != nil && canRetry(info, rpcErr) {
goto RETRY
}
return true, rpcErr
}
有同学看到这里,不是还有consistent 模式没有讲吗,这个就不在分析了, 不然文章太长了,没有人看
总结
一写就这么多,总算把consul的一致性读的特性,怎么用的,和背后的原理给说明了,我们默认情况都是default模式,即请求都是需要通过访问agent,agent再请求server,如果server不是leader,还要转发到leader节点。要1次http,2次rpc才能获取到数据,所以如果有consul server压力大的,可以通过cache来缓解server特别是leader的压力。