GraphQL Dataloader 在 Golang 當中的打開方法與原理解析（下）

1
2
3
4
5

// IconURL is the resolver for the icon_url field.
func (r *roomFeatureResolver) IconURL(ctx context.Context, obj *model.RoomFeature) (*string, error) {
  f, _ := storage.GetFeature(ctx, *obj.ID)
  return f.IconUrl, nil
}

1
2
3
4
5
6
7

func GetFeature(ctx context.Context, featureID int) (*model.RoomFeatureWithData, error) {
  //...
  thunk := loaders.FeatureLoader.Load(ctx, DataLoader.StringKey(fmt.Sprintf("%d", featureID)))
  result, err := thunk()
  //...
  return result.(*model.RoomFeatureWithData), nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

query {
  content(hotel_id: 1) {
    rooms {
      features {
        id
        icon_url
      }
    }
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

//github.com/graph-gophers/DataLoader@v5.0.0+incompatible/DataLoader.go
func (l *Loader) Load(originalContext context.Context, key Key) Thunk {
  ctx, finish := l.tracer.TraceLoad(originalContext, key)

  c := make(chan *Result, 1)
  var result struct {
    mu    sync.RWMutex
    value *Result
  }

  l.cacheLock.Lock()
  if v, ok := l.cache.Get(ctx, key); ok {
    defer finish(v)
    defer l.cacheLock.Unlock()
    return v
  }
  //...下面續
}

1

type Thunk func() (interface{}, error)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

func (l *Loader) Load(originalContext context.Context, key Key) Thunk {
  // ...接續上面
  thunk := func() (interface{}, error) {
    result.mu.RLock() //上讀鎖，避免在讀取資料時，同時有其他線程想更新該資料
    resultNotSet := result.value == nil
    result.mu.RUnlock() // 釋放讀鎖

    if resultNotSet {
        result.mu.Lock() //上獨占鎖，避免在更新資料時，同時有其他線程想更新該資料
        if v, ok := <-c; ok { // block 住直到 c 有 result 傳入
          result.value = v
        }
        result.mu.Unlock() //釋放獨佔鎖
    }
    result.mu.RLock()
    defer result.mu.RUnlock()
    return result.value.Data, result.value.Error
  }
  defer finish(thunk)

  l.cache.Set(ctx, key, thunk)
  l.cacheLock.Unlock()

  //...下面續
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

type batchRequest struct {
  key     Key
  channel chan *Result
}

type batcher struct {
  input    chan *batchRequest
  batchFn  BatchFunc
    //...
}

func (l *Loader) Load(originalContext context.Context, key Key) Thunk {
    //...接續上面

  // req 包含了要查詢的 key 和 等待接收結果的 channel 
  req := &batchRequest{key, c}

  l.batchLock.Lock()

  if l.curBatcher == nil {
    l.curBatcher = l.newBatcher(l.silent, l.tracer)

    go l.curBatcher.batch(originalContext)

    l.endSleeper = make(chan bool)
    go l.sleeper(l.curBatcher, l.endSleeper)
  }

  l.curBatcher.input <- req

  // 如果有額外設定 batchCap，可以限制一次最多可以接受的 req 數量
  if l.batchCap > 0 { // 但我們並沒有設定，所以 batchCap 為 0，不會進入這邊
    //...
  }
  l.batchLock.Unlock()

  return thunk
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

func (b *batcher) batch(originalContext context.Context) {
  var (
    keys     = make(Keys, 0)
    reqs     = make([]*batchRequest, 0)
    items    = make([]*Result, 0)
    panicErr interface{}
  )

    // keys 的順序 等同 reqs 的順序
  for item := range b.input {
    keys = append(keys, item.key)
    reqs = append(reqs, item)
  }

  ctx, finish := b.tracer.TraceBatch(originalContext, keys)
  defer finish(items)

  func() {
        // ...
    items = b.batchFn(ctx, keys) // 執行自定義的資料庫查詢方法
  }()

    // ...

    // 因為自定義方法當中也有依照 keys 的順序回傳 result，且 keys 的順序 等同 reqs 的順序
    // 所以這樣放入順序不會亂
  for i, req := range reqs { 
    req.channel <- items[i]
    close(req.channel)
  }
}

1
2
3
4
5
6
7
8

if l.batchCap > 0 { 
  //...
  if l.count == l.batchCap {
    l.curBatcher.end() // 直接終止接收 req，不再繼續等待
    close(l.endSleeper) // 關閉等待特定一段時間的 channel
    l.reset()
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

func (l *Loader) sleeper(b *batcher, close chan bool) {
  select {
  // 只有在 Load 方法中進入 `l.batchCap > 0` 的 if 區塊時，會對 close channel 傳送訊號
  // 這邊收到訊號表示已經提前關閉了 input channel，也不需繼續等待下去
  case <-close:
    return
  case <-time.After(l.wait):
  }

  l.batchLock.Lock()
  b.end()  // <-----here

    //...
}

func (b *batcher) end() {
  if !b.finished {
    close(b.input)
    b.finished = true
  }
}

1
2
3
4
5
6
7
8

func NewBatchedLoader(batchFn BatchFunc, opts ...Option) *Loader {
  loader := &Loader{
    batchFn:  batchFn,
    inputCap: 1000,
    wait:     16 * time.Millisecond, // <-----here
  }
  //...
}