Drupal 11：内置于 Drupal 的批量操作

这是关于 Drupal 批量 API 系列文章的第六篇。批量 API 是 Drupal 里的一个系统，它允许将数据分小块处理，从而防止出现超时错误或内存问题。在 Drupal 开发和 Drupal 模块开发中，批量 API 都发挥着重要作用，特别是在 Drupal 11 版本中，其功能和性能都得到了进一步优化。

到目前为止，在这个系列中，我们已经学习了如何使用表单创建批量处理流程，接着如何创建批量类以便通过 Drush 运行批量任务，如何使用完成状态来控制批量处理，如何通过批量处理流程处理 CSV 文件，以及如何向正在运行的批量处理流程中添加任务。这些文章为你使用 Drupal 批量 API 打下了良好的基础。

在本文中，我们将探讨 Drupal 内部是如何使用批量 API 的。Drupal 中的批量 API 要么用于执行某个任务，我把它称作“直接”使用；要么将批量操作传递给一个钩子，我把这叫做“间接”使用。要知道，这些并非官方术语，我只是用它们来区分 Drupal 使用批量 API 的不同方式。我发现用这些术语来描述批量任务的运行位置很有用。

让我们先看看直接使用的情况。

一、直接使用

直接使用意味着 Drupal 中的某个方法创建一个 BatchBuilder 对象，然后用该对象来设置并触发批量任务的运行（通过 batch_set() 函数）。这种方式在 Drupal 的各种场景中都有应用，包括：

• 安装 Drupal。
• 安装模块。
• 导入翻译。
• 导入配置。
• 删除用户。
• 批量更新内容。
• 还有很多其他场景！

举个例子，我们来看看 Drupal 重建节点访问授权系统时的批量操作。这个系统本质上是一个表格，Drupal 用它来确定用户是否可以在特定页面上执行某个操作。有时候，为了包含节点访问系统的更改，这个表格需要重建。如果你使用过 hook_node_grants() 和 hook_node_access_records() 钩子，那么就需要重建这个系统来更新系统内容访问规则。

为了重建这个表格，我们需要查看网站中的每个节点，以确定需要什么样的访问矩阵。显然，这是一项繁重的工作，因此运行这个过程的 node_access_rebuild() 函数具备将该系统作为批量任务运行的能力。

以下是 core/modules/node/node.module 文件中 node_access_rebuild() 函数的批量任务设置代码。

function node_access_rebuild($batch_mode = FALSE) {
  // ... 为清晰起见，此处省略部分代码 ...
      $batch_builder = (new BatchBuilder())
        ->setTitle(t('Rebuilding content access permissions'))
        ->addOperation('_node_access_rebuild_batch_operation', [])
        ->setFinishCallback('_node_access_rebuild_batch_finished');
      batch_set($batch_builder->toArray());
  // ... 为清晰起见，此处省略部分代码 ...
}

_node_access_rebuild_batch_operation 函数被定义为这个批量任务要调用的单个操作，它的结构与我们在本系列其他文章中看到的批量处理代码非常相似。我就不在这里贴出完整的源代码了（你可以在同一个 node.module 文件中查看完整源码），但以下是重要的部分。

批量操作回调函数不接受任何自定义属性（我们仍然可以从批量任务运行中获取默认的 $context），所以首先要做的是确定批量任务运行的“最大值”。这可以通过实体查询系统来完成，该系统会对所有可用节点进行计数查询。数组中设置的另外两个参数用于跟踪进度。

function _node_access_rebuild_batch_operation(&$context) {
  $node_storage = \Drupal::entityTypeManager()->getStorage('node');
  if (empty($context['sandbox'])) {
    // 启动多步骤处理。
    $context['sandbox']['progress'] = 0;
    $context['sandbox']['current_node'] = 0;
    $context['sandbox']['max'] = \Drupal::entityQuery('node')->accessCheck(FALSE)->count()->execute();
  }

然后，我们以 20 个节点为一批，遍历系统中的所有节点。使用带有条件和范围的实体查询来获取我们需要的节点 ID。在批量处理过程中，我们会更新“progress”属性和“current_node”属性以跟踪处理进度。

  // 处理接下来的 20 个节点。
  $limit = 20;
  $nids = \Drupal::entityQuery('node')
    ->condition('nid', $context['sandbox']['current_node'], '>')
    ->sort('nid', 'ASC')
    ->accessCheck(FALSE)
    ->range(0, $limit)
    ->execute();
  $nodes = Node::loadMultiple($nids);
  foreach ($nids as $nid) {
    // ... 为清晰起见，此处省略部分代码 ...
    $context['sandbox']['progress']++;
    $context['sandbox']['current_node'] = $nid;
  }

最后，我们使用当前进度除以可用项目的最大数量来设置“finished”属性。如果这两个值相同，则不设置“finished”属性，这意味着使用默认值“1”，批量操作结束。

  // 多步骤处理：报告进度。
  if ($context['sandbox']['progress'] != $context['sandbox']['max']) {
    $context['finished'] = $context['sandbox']['progress'] / $context['sandbox']['max'];
  }
}

我在本系列的其他文章中已经展示过这些概念，但要是遇到困难，在 Drupal 中搜索 BatchBuilder，看看项目中是如何将这些内容组合在一起的，会很有帮助。

接下来，让我们看看 Drupal 中的间接批量任务。

二、间接使用

批量 API 的间接使用意味着批量操作在更上层被定义，然后被传递到可能使用它们的地方。一个典型的例子就是普通的更新钩子。这包括 hook_update_N() 和 hook_post_update_NAME()，它们都用于处理更新。在 Drupal 升级过程中，这些钩子的合理运用对于确保系统稳定至关重要。

这些钩子都会收到一个可选参数 $sandbox，其中包含一个活跃的批量处理流程。大多数情况下，不需要使用这个变量，实际上，如果你愿意，甚至可以在函数声明中省略它。

然而，如果你打算在更新钩子中完成一些工作，那么可以使用 $sandbox 参数来接入批量系统。Drupal 在调用这些钩子之前会创建一个批量任务，并会监听你在 $sandbox 参数中设置的属性，以确定是否需要再次调用更新钩子。这会将更新钩子变成一个批量处理函数，你可以反复调用它，直到工作完成。

你可以使用以下属性在更新钩子中控制批量操作。

• #finished - 这与普通批量 API 处理回调中的“finished”属性作用类似。将其设置为小于 1 意味着更新批量任务尚未完成，因此会再次调用该钩子。将其设置为大于或等于 1 意味着更新批量任务已完成。默认情况下，假设该值为 1，因此更新方法将只被调用一次。
• #abort - 添加这个属性，无论其值是什么，都会导致更新钩子报告失败。这反过来会导致整个更新过程因错误条件而停止。

注意这些属性前面的“#”符号！这对于更新系统识别它们很重要。

让我们看看一个示例更新钩子的典型设置。由于 hook_update_N() 钩子不应用于更新内容，它最适合用于修改多个配置项、直接操作表中的数据。因此，我不会在这里做任何实际有用的操作，因为那样会使这个示例变得更加复杂。

在我们的模块 batch_update_example 中，我们需要添加一个更新钩子，所以我们创建一个名为 batch_update_example.install 的文件，并添加以下函数声明。

function batch_update_example_update_10001(&$sandbox = NULL) {
}

在更新钩子中要做的第一件事是设置一些属性。我们知道需要设置 #finished 属性，但我们设置的任何其他属性在函数调用之间都会被保留。因此，我们设置一个进度属性和一个最大值属性来跟踪批量任务的进度。然后，完成状态将是进度属性除以最大值属性的结果。

function batch_update_example_update_10001(&$sandbox = NULL) {
   if (!isset($sandbox['progress'])) {
    $sandbox['progress'] = 0;
    $sandbox['max'] = 1000;
  }

  // 批量操作代码放在这里...

  $sandbox['#finished'] = $sandbox['progress'] / $sandbox['max'];
}

在批量任务运行期间，我们可以使用标准的 Drupal 消息服务向用户输出任何消息。这与普通的批量 API 略有不同，在普通批量 API 中我们会创建一个消息属性；在这里设置该属性不会有相同的效果。

\Drupal::messenger()->addMessage($sandbox['progress'] . ' items processed.');

以下是一个示例更新钩子，它会处理 1000 个项目，每个项目暂停约 4000 微秒。这样会产生足够的延迟，以便可以看到更新钩子的处理进度，否则整个过程几乎会瞬间完成。

function batch_update_example_update_10001(&$sandbox) {
  if (!isset($sandbox['progress'])) {
    $sandbox['progress'] = 0;
    $sandbox['max'] = 1000;
  }

  $batchSize = 100;
  $batchUpperRange = $sandbox['progress'] + $batchSize;

  for ($i = $sandbox['progress']; $i < $batchUpperRange; $i++) {
    // 跟踪进度。
    $sandbox['progress']++;

    // 在这里处理更新，例如，我们可能对多个不同的表或配置实体执行一些操作。
    // 在这里对实体执行操作是不安全的（请参阅 hook_post_update_NAME()）。
    // 为了模拟正在进行的工作，我们将暂停 4 秒，再加上批量任务的 ID 编号。
    usleep(4000 + $i);
  }
  \Drupal::messenger()->addMessage($sandbox['progress'] . ' items processed.');

  // 返回完成属性，但注意完成属性前面的 "#" 符号。
  // 这是更新钩子所必需的。
  $sandbox['#finished'] = $sandbox['progress'] / $sandbox['max'];
}

当我们运行这个更新钩子时，会看到以下输出。

$ drush updatedb --yes
 ---------------------- --------------- --------------- ------------------------------------------------------------------------
  Module                 Update ID       Type            Description
 ---------------------- --------------- --------------- ------------------------------------------------------------------------
  batch_update_example   10001           hook_update_n   10001 - A demonstration of the hook_update_N hook using the Batch API.
 ---------------------- --------------- --------------- ------------------------------------------------------------------------

// 你是否希望运行指定的待更新任务？：是的。

>  [notice] Update started: batch_update_example_update_10001
>  [notice] Update completed: batch_update_example_update_10001
>  [notice] Message: 100 items processed.
>
>  [notice] Message: 200 items processed.
>
>  [notice] Message: 300 items processed.
>
>  [notice] Message: 400 items processed.
>
>  [notice] Message: 500 items processed.
>
>  [notice] Message: 600 items processed.
>
>  [notice] Message: 700 items processed.
>
>  [notice] Message: 800 items processed.
>
>  [notice] Message: 900 items processed.
>
>  [notice] Message: 1000 items processed.
>
 [success] Finished performing updates.

你可以将此作为自己更新钩子的基础。

请记住，由于并非所有钩子都会被触发，因此 hook_update_N() 钩子不应用于操作你网站上的内容。相反，应使用 hook_post_update_NAME() 钩子来对内容进行更改。这两个钩子都可以访问相同的批量系统，因此你可以以相同的方式为每个钩子运行批量任务。

还有其他一些情况会将 $sandbox 参数传递给函数，这意味着你可以将它们作为批量系统运行。然而，它们的主要用途还是在更新钩子中。

如果你想要这个示例的代码，可以在相关 Drupal 批量示例仓库中的 batch_update_example 模块里找到。这个模块有一个表单，你可以用它来重置 Drupal 中的模块版本标记，这样你就可以反复运行更新钩子。post hook_post_update_NAME() 钩子将使用批量系统更新网站中的每个节点。

三、结论

批量 API 不仅能在 Drupal 中使用，而且是系统多个不同部分的组成部分。这既可以是直接使用批量 API 来执行任务，也可以是设置一个批量任务并将其提供给上层钩子。

批量任务的间接设置意味着，如果你需要处理大量数据，可以在更新钩子中使用批量系统。更新钩子在不使用批量系统的情况下也能正常工作，我只在少数情况下使用过这个系统。然而，在 Drupal 的重要部分拥有这个系统是非常有用的。

本系列文章的所有代码都可以在相关 Drupal 批量示例仓库中找到。请查看本系列的其他文章，以获取更多关于如何以各种不同方式使用批量系统的信息。