源码阅读 | NOSAE

nsq阅读（1）——概述

Note 基于nsq v1.3.0 简介 NSQ是类似kafka、rabbitmq那样的消息队列系统，关于他怎么高性能，怎么好上手这些都不必多说，都是吹逼。这篇主要介绍一下nsq的整个大致架构，建立一个概念，方便后续的源码分析有迹可循。架构 NSQ由三个守护进程组成： ...

nsq阅读（2）——diskqueue

开篇 DiskQueue 是 NSQ 中的持久化组件，承担着将消息存储到磁盘的任务，它以高效、可靠的方式在内存与磁盘之间进行切换。在开始分析 nsq 那几个大件之前，先深入分析下 DiskQueue 的核心设计和实现，解密其背后的关键技术点，包括写入机制、读取逻辑以及在异常场景下的恢复策略。设计 nsq将diskqueue抽取出来作为一个库来维护：https://github.com/nsqio/go-diskqueue 里面只有一个源码文件diskqueue.go，里面核心结构体是基于磁盘的消息队列diskQueue，实现了nsqd中的BackendQueue接口，比较简单： type BackendQueue interface { Put([]byte) error ReadChan() <-chan []byte Close() error Delete() error Depth() int64 Empty() error } 消息以字节数组的方式传递，接收端需要对字节数组进行解码，存储端的实现就能相对简单。 diskqueue在创建时会运行一个叫做ioLoop的协程用于处理io相关的事务，并与主协程通过以下各个chan进行通信： type diskQueue struct { ... // 队列大小（即depth在nsq中表示的就是可读消息数） depthChan chan int64 // 写磁盘的数据 writeChan chan []byte // 写磁盘的结果 writeResponseChan chan error // 清空队列 emptyChan chan int // 清空队列的结果 emptyResponseChan chan error // 退出 exitChan chan int // 主协程等待ioLoop退出 exitSyncChan chan int } logrotation与metadata diskqueue还用了rotation的方式来限制单个文件的大小（拓展：在linux中有一个实现logrotation命令叫logrotate），这样便于在消息读出之后，通过直接删除文件的方式来删除不再需要的数据。为了实现logrotation，还会维护一些元数据：队列大小（消息数）正在读第几个文件中的第几个字节正在写第几个文件中的第几个字节代码中通过persistMetaData方法来原子性地持久化这些元数据。元数据非常重要，如果不是原子性地写入，比如写一半就宕机了，那么所有数据将会不可用。原子性是通过写临时文件、将临时文件重命名为最终文件名来保证的写数据使用接口方法Put写入： func (d *diskQueue) Put(data []byte) error { d.RLock() defer d.RUnlock() if d.exitFlag == 1 { return errors.New("exiting") } d.writeChan <- data return <-d.writeResponseChan } 由ioLoop协程处理数据的写入，调用writeOne进行处理： func (d *diskQueue) writeOne(data []byte) error { var err error dataLen := int32(len(data)) totalBytes := int64(4 + dataLen) if d.writePos > 0 && d.writePos+totalBytes > d.maxBytesPerFile { // 写入的大小将超出单个文件大小限制，触发logrotation，即写入下一个新文件 if d.readFileNum == d.writeFileNum { d.maxBytesPerFileRead = d.writePos } d.writeFileNum++ d.writePos = 0 // 每次logrotation前，都sync将最后一个文件以及元数据落盘 err = d.sync() if err != nil { d.logf(ERROR, "DISKQUEUE(%s) failed to sync - %s", d.name, err) } // 关闭序号为i的数据文件 if d.writeFile != nil { d.writeFile.Close() d.writeFile = nil } } // 创建序号为i+1的数据文件 if d.writeFile == nil { curFileName := d.fileName(d.writeFileNum) d.writeFile, err = os.OpenFile(curFileName, os.O_RDWR|os.O_CREATE, 0600) if err != nil { return err } d.logf(INFO, "DISKQUEUE(%s): writeOne() opened %s", d.name, curFileName) if d.writePos > 0 { _, err = d.writeFile.Seek(d.writePos, 0) if err != nil { d.writeFile.Close() d.writeFile = nil return err } } } // 数据大小写入缓存 d.writeBuf.Reset() err = binary.Write(&d.writeBuf, binary.BigEndian, dataLen) if err != nil { return err } // 数据写入缓存 _, err = d.writeBuf.Write(data) if err != nil { return err } // 缓存写入文件 _, err = d.writeFile.Write(d.writeBuf.Bytes()) if err != nil { d.writeFile.Close() d.writeFile = nil return err } d.writePos += totalBytes d.depth += 1 return err } 这里要注意在整个rotation的过程中并不保证原子性，而是分为三步： ...

nsq阅读（3）——nsqd

Note 基于nsq v1.3.0 执行流程在第一篇中，主要对nsq中涉及到的主要组件，以及数据流在这些组件中的流动进行了简单说明。在本篇中，我们跟随上一篇最后给出的demo，对应到nsqd的代码中，看一下执行流程是怎么样的。上篇最后我们向4151端口发起http请求来发送消息： curl -d "test message" "http://127.0.0.1:4151/pub?topic=test_topic" 这个调用的是http接口/pub，对应nsqd源码http.go中的doPUB函数： ...

gRPC阅读（4）——负载均衡

负载均衡算法常见的负载均衡算法如下： RoundRobin（轮询） Weight-RoundRobin（加权轮询）不同的后端服务器可能机器的配置和当前系统的负载并不相同，因此它们的抗压能力也不相同。给配置高、负载低的机器配置更高的权重，而配置低、负载高的机器，给其分配较低的权重，降低其系统负载，加权轮询能很好地处理这一问题，并将请求顺序且按照权重分配到后端。 Random（随机） Weight-Random（加权随机）通过系统的随机算法，根据后端服务器的列表随机选取其中的一台服务器进行访问源地址哈希法源地址哈希的思想是根据获取客户端的 IP 地址，通过哈希函数计算得到的一个数值，用该数值对服务器列表的大小进行取模运算，得到的结果便是客服端要访问服务器的序号。采用源地址哈希法进行负载均衡，同一 IP 地址的客户端，当后端服务器列表不变时，它每次都会映射到同一台后端服务器进行访问最小连接数法最小连接数算法比较灵活和智能，由于后端服务器的配置不尽相同，对于请求的处理有快有慢，它是根据后端服务器当前的连接情况，动态地选取其中当前积压连接数最少的一台服务器来处理当前的请求，尽可能地提高后端服务的利用效率，将负责合理地分流到每一台服务器 Consistent-Hash（一致性哈希算法）常见的是 Ketama 算法（虚拟节点），该算法是用来解决 cache 失效导致的缓存穿透的问题的，当然也可以适用于 gRPC 长连接的场景自适应算法（P2C：多选二，二选一）：即从可用节点列表中随机选择两个节点，计算它们的负载率，选择负载率较低的进行请求基于最小负载策略：该策略是 linkerd 的默认负载均衡器。当确定发送请求的位置时，linkerd 随机从负载均衡器池中选择两个副本，并选择两者中最少负载的副本。负载由每个副本的未完成请求数决定。该算法为单个副本提供了可管理的负载上限，与具有相似性能的其他算法相比，开销较少峰值 EWMA（预测）策略：该算法是上述的变体，同样在发送请求时仍然在两个副本之间进行选择。不一样的是，该算法需要保持观察到的延迟的动态平均值，并且使用它来对每个副本的未完成请求的数量进行加权。这种方法对延迟波动更敏感，并通过向较慢的后端发送更少的请求来允许他们恢复时间（可以通过调参来改变对请求延时的敏感度） gRPC负载均衡接着上篇，通过DNS进行服务发现获取服务器IP list之后，调用ccResolverWrapper.UpdateState更新状态： // UpdateState is called by resolver implementations to report new state to gRPC // which includes addresses and service config. func (ccr *ccResolverWrapper) UpdateState(s resolver.State) error { ... return ccr.cc.updateResolverStateAndUnlock(s, nil) } func (cc *ClientConn) updateResolverStateAndUnlock(s resolver.State, err error) error { ... var ret error // 应用最新的从dns发现的ServiceConfig if cc.dopts.disableServiceConfig { channelz.Infof(logger, cc.channelz, "ignoring service config from resolver (%v) and applying the default because service config is disabled", s.ServiceConfig) cc.maybeApplyDefaultServiceConfig() } else if s.ServiceConfig == nil { cc.maybeApplyDefaultServiceConfig() } else { if sc, ok := s.ServiceConfig.Config.(*ServiceConfig); s.ServiceConfig.Err == nil && ok { configSelector := iresolver.GetConfigSelector(s) if configSelector != nil { if len(s.ServiceConfig.Config.(*ServiceConfig).Methods) != 0 { channelz.Infof(logger, cc.channelz, "method configs in service config will be ignored due to presence of config selector") } } else { configSelector = &defaultConfigSelector{sc} } cc.applyServiceConfigAndBalancer(sc, configSelector) } else { ... } } // ServiceConfig的负载均衡配置 var balCfg serviceconfig.LoadBalancingConfig if cc.sc != nil && cc.sc.lbConfig != nil { balCfg = cc.sc.lbConfig } // 负载均衡器 bw := cc.balancerWrapper cc.mu.Unlock() // 应用服务发现结果 uccsErr := bw.updateClientConnState(&balancer.ClientConnState{ResolverState: s, BalancerConfig: balCfg}) if ret == nil { ret = uccsErr // prefer ErrBadResolver state since any other error is // currently meaningless to the caller. } return ret } // ccBalancerWrapper解耦ClientConn和Balancer，在updateClientConnState调用时才会创建Balancer // 并且ccBalancerWrapper使用gracefulswitch.Balancer支持Balancer的优雅切换 func (ccb *ccBalancerWrapper) updateClientConnState(ccs *balancer.ClientConnState) error { errCh := make(chan error) uccs := func(ctx context.Context) { defer close(errCh) if ctx.Err() != nil || ccb.balancer == nil { return } name := gracefulswitch.ChildName(ccs.BalancerConfig) if ccb.curBalancerName != name { ccb.curBalancerName = name channelz.Infof(logger, ccb.cc.channelz, "Channel switches to new LB policy %q", name) } // 通知gracefulBalancer状态更新 err := ccb.balancer.UpdateClientConnState(*ccs) if logger.V(2) && err != nil { logger.Infof("error from balancer.UpdateClientConnState: %v", err) } errCh <- err } onFailure := func() { close(errCh) } ccb.serializer.ScheduleOr(uccs, onFailure) return <-errCh } // gsb更新状态 func (gsb *Balancer) UpdateClientConnState(state balancer.ClientConnState) error { // 获取最新的balancer balToUpdate := gsb.latestBalancer() gsbCfg, ok := state.BalancerConfig.(*lbConfig) if ok { // Switch to the child in the config unless it is already active. if balToUpdate == nil || gsbCfg.childBuilder.Name() != balToUpdate.builder.Name() { var err error // 切换到新的balancer balToUpdate, err = gsb.switchTo(gsbCfg.childBuilder) if err != nil { return fmt.Errorf("could not switch to new child balancer: %w", err) } } // Unwrap the child balancer's config. state.BalancerConfig = gsbCfg.childConfig } if balToUpdate == nil { return errBalancerClosed } // 通知真正的Balancer状态更新 return balToUpdate.UpdateClientConnState(state) } // gsb优雅切换balancer func (gsb *Balancer) switchTo(builder balancer.Builder) (*balancerWrapper, error) { gsb.mu.Lock() if gsb.closed { gsb.mu.Unlock() return nil, errBalancerClosed } bw := &balancerWrapper{ builder: builder, gsb: gsb, lastState: balancer.State{ ConnectivityState: connectivity.Connecting, Picker: base.NewErrPicker(balancer.ErrNoSubConnAvailable), }, subconns: make(map[balancer.SubConn]bool), } balToClose := gsb.balancerPending // nil if there is no pending balancer if gsb.balancerCurrent == nil { gsb.balancerCurrent = bw } else { gsb.balancerPending = bw } gsb.mu.Unlock() // 关闭旧的balancer balToClose.Close() // 创建新的balancer newBalancer := builder.Build(bw, gsb.bOpts) if newBalancer == nil { gsb.mu.Lock() if gsb.balancerPending != nil { gsb.balancerPending = nil } else { gsb.balancerCurrent = nil } gsb.mu.Unlock() return nil, balancer.ErrBadResolverState } bw.Balancer = newBalancer return bw, nil } // pick_first balancer状态更新 func (b *pickfirstBalancer) UpdateClientConnState(state balancer.ClientConnState) error { ... cfg, ok := state.BalancerConfig.(pfConfig) ... // 展开endpoints/addrs var addrs []resolver.Address ... // balancer之前维护了子连接 // 如果连接的地址在addrs中，则保持连接 // 否则断开并用addrs重连 if b.subConn != nil { b.cc.UpdateAddresses(b.subConn, addrs) return nil } // balancer之前没有维护子连接 // 创建新的子连接 var subConn balancer.SubConn subConn, err := b.cc.NewSubConn(addrs, balancer.NewSubConnOptions{ StateListener: func(state balancer.SubConnState) { // 子连接状态更新，回调通知balancer b.updateSubConnState(subConn, state) }, }) ... b.subConn = subConn b.state = connectivity.Idle // 通知cc现在是Connecting状态 b.cc.UpdateState(balancer.State{ ConnectivityState: connectivity.Connecting, Picker: &picker{err: balancer.ErrNoSubConnAvailable}, }) b.subConn.Connect() return nil } 可以看到在pick_first策略下，会将所有地址都试一遍，连上其中一个后，后续picker返回的都是那个连接，相当于不做负载均衡： ...

gRPC阅读（3）—— 服务发现

服务发现概述平时用浏览器上过网都知道，输入一个网址比如google.com就能访问内容，背后是DNS帮我们将google.com解析成IP地址，最终浏览器才能基于TCP协议，从本地连接到这个服务提供商的IP地址。所以DNS属于服务发现的其中一种方式。所以服务发现提供的就是通过自动化的方式帮助服务在网络中找到彼此，无需手动配置。一个好的服务发现需要：服务地址动态变化：服务的 IP 或端口可能因为容器化或自动扩展而频繁改变。高可用：需要在服务实例宕机时快速感知并移除不健康的实例。负载均衡：服务发现需要为调用方提供负载均衡能力，选择最佳的服务实例。服务发现通常与负载均衡同时实现，分为两种方式：客户端服务发现（如eureka、consul）：在客户端做负载均衡，选择一个实例进行调用，优点是避免集中式LB可能存在的瓶颈，性能较好，但是每个客户端需要维护服务端列表，服务端这部分的负载可能变高。并且更新LB或其他相关组件的策略时需要所有客户端都一起更新，管理不方便。并且需要多语言支持代理服务发现（如k8s+coreDNS、nginx+consul）：客户端将请求发送到负载均衡器（如 API 网关），由负载均衡器查询服务注册中心并将请求转发给目标服务实例。独立LB进程：LB与消费者在同一个主机中，但分别作为不同的进程，避免了需要多语言支持，以及LB的更新不需要调用方改代码。服务发现的核心组件有：注册中心、服务提供者、客户端（服务消费者）服务发现的关键功能有：服务注册、服务查询、健康检查、动态更新 gRPC服务发现 gRPC使用客户端服务发现，gRPC中称为名称解析（Name Resolution），默认情况下使用DNS-resolver。通过服务发现解析出IP列表后就通过LB组件进行负载均衡并建立连接。下面基于target=localhost:50052这个服务端地址来进行分析，并且是默认的DNS作为resolver（不用官方例子的50051端口是因为被mac的launchd进程占用了）。首先gRPC在创建cc（ClientConn）的时候，使用initParsedTargetAndResolverBuilder创建resolver.Builder。这一步决定的是采用什么服务发现机制，默认是DNS。 func (cc *ClientConn) initParsedTargetAndResolverBuilder() error { logger.Infof("original dial target is: %q", cc.target) // 尝试直接解析target并获取相应的resolver.Builder var rb resolver.Builder parsedTarget, err := parseTarget(cc.target) if err == nil { rb = cc.getResolver(parsedTarget.URL.Scheme) if rb != nil { cc.parsedTarget = parsedTarget cc.resolverBuilder = rb return nil } } // target没有指定schema（比如我们的localhost:50052是没有指定schema的）或者无法匹配schema对应的resolver.Builder // 那么使用默认的schema，即dns defScheme := cc.dopts.defaultScheme if internal.UserSetDefaultScheme { defScheme = resolver.GetDefaultScheme() } // 此处canonicalTarget为dns:///localhost:50052 // "//"与第三个"/"之间的是authority canonicalTarget := defScheme + ":///" + cc.target // 再次尝试target并获取相应的resolver.Builder，此处会拿到dns.dnsBuilder parsedTarget, err = parseTarget(canonicalTarget) if err != nil { return err } rb = cc.getResolver(parsedTarget.URL.Scheme) if rb == nil { return fmt.Errorf("could not get resolver for default scheme: %q", parsedTarget.URL.Scheme) } // 保存parsedTarget和resolverBuilder cc.parsedTarget = parsedTarget cc.resolverBuilder = rb return nil } 那么resolverBuilder在什么时候会Build一个resolver出来呢？在ide的帮助下，可以直接定位到这个函数中： ...