Golang服务的请求调度的实现

// SheddingHandler returns a middleware that does load shedding.
func SheddingHandler(shedder load.Shedder, metrics *stat.Metrics) func(http.Handler) http.Handler {
	if shedder == nil {
		return func(next http.Handler) http.Handler {
			return next
		}
	}
	ensureSheddingStat() // 负责每分钟打印shedding相关的数据
	return func(next http.Handler) http.Handler {
		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			sheddingStat.IncrementTotal()
			promise, err := shedder.Allow() // 判断是否允许此请求进入下一步
			if err != nil {
				metrics.AddDrop() // drop掉请求，在中间件层面就拒绝了请求
				sheddingStat.IncrementDrop()
				logx.Errorf("[http] dropped, %s - %s - %s",
					r.RequestURI, httpx.GetRemoteAddr(r), r.UserAgent())
				w.WriteHeader(http.StatusServiceUnavailable)// 返回503，提示服务不可用
				return
			}
			cw := response.NewWithCodeResponseWriter(w)
			defer func() {
				if cw.Code == http.StatusServiceUnavailable {
					promise.Fail() // 相关指标记录
				} else {
					sheddingStat.IncrementPass()
					promise.Pass() // 相关指标记录
				}
			}()
			next.ServeHTTP(cw, r)
		})
	}
}

// Allow implements Shedder.Allow.
func (as *adaptiveShedder) Allow() (Promise, error) {
	if as.shouldDrop() {// 判断是否应该丢弃
		as.droppedRecently.Set(true)
		return nil, ErrServiceOverloaded// 丢弃
	}
	as.addFlying(1) // 通过校验
	return &promise{
		start:   timex.Now(),
		shedder: as,
	}, nil
}

func (as *adaptiveShedder) shouldDrop() bool {
	if as.systemOverloaded() || as.stillHot() {// 如果任一满足，这个请求都会被过载
		if as.highThru() {
			flying := atomic.LoadInt64(&as.flying)
			as.avgFlyingLock.Lock()
			avgFlying := as.avgFlying
			as.avgFlyingLock.Unlock()
			msg := fmt.Sprintf(
				"dropreq, cpu: %d, maxPass: %d, minRt: %.2f, hot: %t, flying: %d, avgFlying: %.2f",
				stat.CpuUsage(), as.maxPass(), as.minRt(), as.stillHot(), flying, avgFlying)
			logx.Error(msg)
			stat.Report(msg)
			return true
		}
	}
	return false
}
func (as *adaptiveShedder) systemOverloaded() bool {
	if !systemOverloadChecker(as.cpuThreshold) { // 校验CPU的负载是否超出设定值
		return false
	}
	as.overloadTime.Set(timex.Now())// 超出设定值，记录当前的时间（这主要是为了后续流量减小，系统的恢复用）
	return true
}
func (as *adaptiveShedder) stillHot() bool {
	if !as.droppedRecently.True() {// 如果这个请求之前有请求被drop这里值为true，反之为false
		return false// 之前的请求没有被drop表示系统可能没有遇到过载的问题，返回false
	}
	overloadTime := as.overloadTime.Load()// 如果之前有请求被drop，表示存在过载
	if overloadTime == 0 {// 看看是否有记录过载的时间
		return false
	}
	if timex.Since(overloadTime) < coolOffDuration {// 如果小于冷却时间，表示系统依然是过载状态
		return true
	}
	as.droppedRecently.Set(false)// 表示CPU过载，上一次过载过了冷却器，这个请求可以继续执行，设置为false
	return false
}

const (
	// 250ms and 0.95 as beta will count the average cpu load for past 5 seconds
	cpuRefreshInterval = time.Millisecond * 250
	allRefreshInterval = time.Minute
	// moving average beta hyperparameter
	beta = 0.95
)
var cpuUsage int64
func init() {
	go func() {
		cpuTicker := time.NewTicker(cpuRefreshInterval)
		defer cpuTicker.Stop()
		allTicker := time.NewTicker(allRefreshInterval)
		defer allTicker.Stop()
		for {
			select {
			case <-cpuTicker.C:
				threading.RunSafe(func() {
					curUsage := internal.RefreshCpu() // 刷新CPU使用率数据
					prevUsage := atomic.LoadInt64(&cpuUsage)
					// cpu = cpuᵗ⁻¹ * beta + cpuᵗ * (1 - beta)
					usage := int64(float64(prevUsage)*beta + float64(curUsage)*(1-beta))
					atomic.StoreInt64(&cpuUsage, usage)
				})
			case <-allTicker.C:
				if logEnabled.True() {
					printUsage()
				}
			}
		}
	}()
}

func (as *adaptiveShedder) addFlying(delta int64) {
	flying := atomic.AddInt64(&as.flying, delta)// 请求通过检验进入后会加1，请求被服务处理完后会减1
	if delta < 0 {
		as.avgFlyingLock.Lock()
		// 平均请求数计算为当前平均请求数*0.9 + 当前运行请求数*0.1
		as.avgFlying = as.avgFlying*flyingBeta + float64(flying)*(1-flyingBeta)
		as.avgFlyingLock.Unlock()
	}
}
func (as *adaptiveShedder) highThru() bool {
	as.avgFlyingLock.Lock()
	avgFlying := as.avgFlying // 运行中的平均请求数
	as.avgFlyingLock.Unlock()
	maxFlight := as.maxFlight()// 运行的最大的请求数
	// 如果运行的平均请求数>最大的请求数且当前运行的请求数>最大的请求数，表示依旧高负载
	return int64(avgFlying) > maxFlight && atomic.LoadInt64(&as.flying) > maxFlight
}
func (as *adaptiveShedder) maxFlight() int64 {
	// windows = buckets per second
	// maxQPS = maxPASS * windows
	// minRT = min average response time in milliseconds
	// maxQPS * minRT / milliseconds_per_second
	// 最大的运行数的计算为最大请求数*窗口的长度*最小的处理时间
	return int64(math.Max(1, float64(as.maxPass()*as.windows)*(as.minRt()/1e3)))
}

func (as *adaptiveShedder) maxPass() int64 {
	var result float64 = 1
	as.passCounter.Reduce(func(b *collection.Bucket) {
		if b.Sum > result {
			result = b.Sum
		}
	})
	return int64(result)
}
func (as *adaptiveShedder) minRt() float64 {
	result := defaultMinRt
	as.rtCounter.Reduce(func(b *collection.Bucket) {
		if b.Count <= 0 {
			return
		}
		avg := math.Round(b.Sum / float64(b.Count))
		if avg < result {
			result = avg
		}
	})
	return result
}