背景

這個(gè)問(wèn)題簡(jiǎn)單說(shuō)一下背景，如果不明白可以看上篇文章 ，不想看也沒關(guān)系，這是個(gè)通用的解法，后面我會(huì)總結(jié)抽象下。

在上篇文章的最后提到對(duì)每個(gè)摘除的地址做決策時(shí)，需要順序執(zhí)行，且每一個(gè)要摘除的地址都要實(shí)時(shí)獲取該集群的地址信息，以便做出是否需要兜底的決策。

當(dāng)被摘除的機(jī)器非常多時(shí)，獲取地址信息的請(qǐng)求量就會(huì)非常大，對(duì)注冊(cè)中心造成了不小的壓力。

請(qǐng)求數(shù)據(jù)源的接口如下所示（其中 cuuid 是集群的 id）

type Read interface {
 ListClusterEndpoints(ctx context.Context, cuuid string) ([]ptypes.Endpoint, error)
}

相信大家也能理解這個(gè)非常簡(jiǎn)單的背景并且能想到一些解法。每次決策需要按 cuuid 獲取集群，也就是單個(gè)單個(gè)地獲取實(shí)時(shí)集群地址信息，由于是實(shí)時(shí)信息，緩存首先排除，其次自然而然地能想到如果能將請(qǐng)求合并一下，是不是就能解決請(qǐng)求量大的問(wèn)題？

難點(diǎn)

如果只是改邏輯合并一下請(qǐng)求，吭哧吭哧改代碼就完了，也不值得寫這篇文章了，如何改最少的代碼來(lái)實(shí)現(xiàn)合并請(qǐng)求才是最難的。

解法

那天遇到這個(gè)問(wèn)題，晚上輾轉(zhuǎn)反側(cè)想到了這個(gè)解法，其實(shí)主要也是參考 Go http client 的實(shí)現(xiàn)，都說(shuō)看源碼沒用，這不就是用處么？

Read? 數(shù)據(jù)源接口定義保持不變，也就是上層的業(yè)務(wù)代碼完全不用改，只需要把 ListClusterEndpoints 的實(shí)現(xiàn)換掉。

我們可以用一個(gè)隊(duì)列把每個(gè)請(qǐng)求入隊(duì)，入隊(duì)列以后，調(diào)用方阻塞，然后起一些協(xié)程去隊(duì)列里取一批請(qǐng)求參數(shù)，發(fā)起批量請(qǐng)求，響應(yīng)之后喚醒阻塞的調(diào)用方。

為此，我們實(shí)現(xiàn)一個(gè)可以阻塞并被其他協(xié)程喚醒的工具：

type token struct {
 value interface{}
 err   error
}

type Token chan token

func NewToken() Token {
 return make(Token, 1)
}

func (t Token) Done(value interface{}, err error) {
 t <- token{value: value, err: err}
}

func (t Token) Wait(timeout time.Duration) (value interface{}, err error) {
 if timeout <= 0 {
  tk := <-t
  return tk.value, tk.err
 }

 select {
 case tk := <-t:
  return tk.value, tk.err
 case <-time.After(timeout):
  return nil, ErrTokenTimeout
 }
}

其次，定義隊(duì)列和其他參數(shù)：

type DataSource struct {
 paramCh chan param
 readTimeout time.Duration
 concurrency int
 step int
}

type param struct {
 cuuid string
 token Token
}

替換掉原來(lái) ListClusterEndpoints 的實(shí)現(xiàn)：

func (p *DataSource) ListClusterEndpoints(ctx context.Context, cuuid string) ([]ptypes.Endpoint, error) {
 req := param{
  cuuid: cuuid,
  token: NewToken(),
 }

 select {
 case p.paramCh <- req:
 default:
  return nil, fmt.Errorf("list cluster endpoints write channel failed")
 }

 value, err := req.token.Wait(p.readTimeout)
 if err != nil {
  return nil, err
 }
 eps, ok := value.([]ptypes.Endpoint)
 if !ok {
  return nil, fmt.Errorf("value is not endpoints")
 }
 return endpoints, nil
}

再起幾個(gè)協(xié)程來(lái)處理任務(wù)：

func (p *DataSource) startListClusterEndpointsLoop() {
 for i := 0; i < p.concurrency; i++ {
  go func() {
   for {
    reqs := p.getListClusterEndpointsReqFromChan()
    p.doBatchListClusterEndpoints(reqs)
   }
  }()
 }
}

最關(guān)鍵的是 getListClusterEndpointsReqFromChan 的實(shí)現(xiàn)，既不能讓協(xié)程空跑，這樣太消耗cpu，又要能及時(shí)地取到一批參數(shù)，我們采取的方法是先阻塞地獲取一個(gè)參數(shù)，如果沒數(shù)據(jù)則阻塞，如果有數(shù)據(jù)，繼續(xù)取，直到數(shù)量達(dá)到上限或者取不到數(shù)據(jù)為止，此時(shí)這一批數(shù)據(jù)就可以批量地進(jìn)行調(diào)用了。

func (p *DataSource) getListClusterEndpointsReqFromChan() []param {
 reqs := make([]param, 0)
 select {
 case req := <-p.paramCh:
  reqs = append(reqs, req)
  for i := 1; i < p.step; i++ {
   select {
   case reqNext := <-p.paramCh:
    reqs = append(reqs, reqNext)
   default:
    break
   }
  }
 }
 return reqs
}

最后

這個(gè)方法很簡(jiǎn)單，但是有一些要注意的地方，得做好監(jiān)控，比如調(diào)用方單個(gè)請(qǐng)求的QPS、RT，實(shí)際批量請(qǐng)求的QPS、RT，這樣才好計(jì)算出處理協(xié)程開多少個(gè)合適，還有隊(duì)列寫入失敗、隊(duì)列長(zhǎng)度等等監(jiān)控，當(dāng)容量不足時(shí)及時(shí)做出調(diào)整。