Dataloader如何缓存和批量数据库请求？

Posted 2023-02-16

【中文标题】Dataloader如何缓存和批量数据库请求？

【英文标题】：How does Dataloader cache and batch database requests?

【发布时间】：2017-06-23 18:06:36

【问题描述】：

查看DataLoader library，它是如何缓存和批处理请求的？

说明以下列方式指定用法：

var DataLoader = require('dataloader')

var userLoader = new DataLoader(keys => myBatchGetUsers(keys));

userLoader.load(1)
    .then(user => userLoader.load(user.invitedByID))
    .then(invitedBy => console.log(`User 1 was invited by $invitedBy`));

// Elsewhere in your application
userLoader.load(2)
    .then(user => userLoader.load(user.lastInvitedID))
    .then(lastInvited => console.log(`User 2 last invited $lastInvited`));

但我不清楚load 函数是如何工作的，以及myBatchGetUsers 函数可能是什么样子。如果可能的话，请你给我一个例子！

【参考方案1】：

Facebook 的 DataLoader 实用程序通过将输入请求与您必须提供的批处理功能相结合来工作。它仅适用于使用 Identifiers 的请求。

分为三个阶段：

聚合阶段：Loader 对象上的任何请求都会延迟到 process.nextTick 批处理阶段：Loader 只需调用您提供的myBatchGetUsers 函数，并结合所有请求的键。 拆分阶段：然后将结果“拆分”，以便输入请求获得响应的所需部分。

这就是为什么在您提供的示例中您应该只有两个请求：

用户 1 和 2 一个 然后是相关用户的一个 (invitedByID)

以 mongodb 为例，您只需定义 myBatchGetUsers 函数以适当地使用 find 方法：

function myBatchGetUsers(keys) 
    // usersCollection is a promisified mongodb collection
    return usersCollection.find(
       
          _id:  $in: keys 
       
    )

【参考方案2】：

我发现重新创建我使用的dataloader 的部分很有帮助，以了解一种可能的实现方式。 （就我而言，我只使用.load() 函数）

因此，创建DataLoader 构造函数的新实例会为您带来两件事：

标识符列表（开头为空） 使用此标识符列表查询数据库的函数（由您提供）。

构造函数可能看起来像这样：

function DataLoader (_batchLoadingFn) 
  this._keys = []
  this._batchLoadingFn = _batchLoadingFn

DataLoader 构造函数的实例可以访问.load() 函数，该函数需要能够访问_keys 属性。所以它是在DataLoad.prototype 对象上定义的：

DataLoader.prototype.load = function(key) 
   // this._keys references the array defined in the constructor function

当通过 DataLoader 构造函数 (new DataLoader(fn)) 创建一个新对象时，您传递的 fn 需要从某个地方获取数据，将一组键作为参数，并返回一个解析为对应于初始键数组的值。

例如，这里有一个虚拟函数，它接受一个键数组，并将相同的数组传回，但值加倍：

const batchLoadingFn = keys => new Promise( resolve => resolve(keys.map(k => k * 2)) )

keys: [1,2,3]
vals: [2,4,6]

keys[0] corresponds to vals[0]
keys[1] corresponds to vals[1]
keys[2] corresponds to vals[2]

然后每次调用.load(indentifier) 函数时，都会向_keys 数组添加一个键，然后在某个时候调用batchLoadingFn，并将_keys 数组作为参数传递。

诀窍是... 如何多次调用.load(id) 但batchLoadingFn 只执行一次？这很酷，也是我探索这个库如何工作的原因。

我发现可以通过指定batchLoadingFn在超时后执行来做到这一点，但是如果在超时间隔之前再次调用.load()，那么超时被取消，一个新的键被添加并且一个致电batchLoadingFn 已重新安排。在代码中实现这一点如下所示：

DataLoader.prototype.load = function(key) 
  clearTimeout(this._timer)
  this._timer = setTimeout(() => this.batchLoadingFn(), 0)

基本上调用.load() 会删除对batchLoadingFn 的挂起调用，然后在事件循环的后面安排对batchLoadingFn 的新调用。这保证了在很短的时间内如果.load() 被多次调用，batchLoadingFn 只会被调用一次。这实际上与去抖动非常相似。或者，至少在构建网站并且您想在mousemove 事件上做某事时它很有用，但是您得到的事件比您想要处理的要多得多。我认为这称为去抖动。

但是调用.load(key) 还需要将一个键推送到_keys 数组，我们可以在.load 函数的主体中通过将key 参数推送到_keys（只需this._keys.push(key)） .但是，.load 函数的约定是它返回与 key 参数解析的内容有关的单个值。在某些时候，batchLoadingFn 将被调用并得到一个结果（它必须返回一个与_keys 对应的结果）。此外，batchLoadingFn 要求实际返回该值的承诺。

我认为接下来的这一点特别聪明（非常值得花时间查看源代码）！

dataloader 库，而不是在 _keys 中保留键列表，实际上保留了与 resolve 函数的引用相关联的键列表，当调用该函数时，值被解析为.load() 的结果。 .load() 返回一个 Promise，当它的 resolve 函数被调用时，一个 Promise 被解析。

所以_keys 数组实际上保留了[key, resolve] 元组的列表。当你的batchLoadingFn 返回时，resolve 函数会调用一个值（希望通过索引号对应于_keys 数组中的项目）。

所以.load 函数看起来像这样（就将[key, resolve] 元组推送到_keys 数组而言）：

DataLoader.prototype.load = function(key) 
  const promisedValue = new Promise ( resolve => this._keys.push(key, resolve) )
  ...
  return promisedValue

剩下的就是使用_keys 键作为参数执行batchLoadingFn，并在其返回时调用正确的resolve 函数

this._batchLoadingFn(this._keys.map(k => k.key))
  .then(values => 
    this._keys.forEach((resolve, i) => 
      resolve(values[i])
    )
    this._keys = [] // Reset for the next batch
  )

结合起来，实现上面的所有代码都在这里：

function DataLoader (_batchLoadingFn) 
  this._keys = []
  this._batchLoadingFn = _batchLoadingFn


DataLoader.prototype.load = function(key) 
  clearTimeout(this._timer)
  const promisedValue = new Promise ( resolve => this._keys.push(key, resolve) )

  this._timer = setTimeout(() => 
    console.log('You should only see me printed once!')
    this._batchLoadingFn(this._keys.map(k => k.key))
      .then(values => 
        this._keys.forEach((resolve, i) => 
          resolve(values[i])
        )
        this._keys = []
      )
  , 0)

  return promisedValue


// Define a batch loading function
const batchLoadingFunction = keys => new Promise( resolve => resolve(keys.map(k => k * 2)) )

// Create a new DataLoader
const loader = new DataLoader(batchLoadingFunction)

// call .load() twice in quick succession
loader.load(1).then(result => console.log('Result with key = 1', result))
loader.load(2).then(result => console.log('Result with key = 2', result))

如果我没记错的话，我认为dataloader 库不使用setTimeout，而是使用process.nextTick。但我无法让它发挥作用。