Drupal 11：为正在运行的批量任务添加操作

这是关于 Drupal 批量处理 API 系列文章的第五篇。批量处理 API 是 Drupal 里的一个系统，它能把数据分成小块来处理，避免出现超时错误或内存问题，在 Drupal 开发中有着重要作用。

截至目前，在这个系列里，我们探讨了用表单创建批量处理流程、创建批量处理类以便通过 Drush 运行批量任务、利用完成状态控制批量处理，以及通过批量处理流程处理 CSV 文件。这些内容为如何使用 Drupal 批量处理 API 奠定了良好基础。

在本文中，我们会更深入探究批量处理系统如何通过在已运行的批量处理流程中创建新的批量运行来处理项目。这将展示批量处理系统的运行机制，以及当你尝试为正在运行的批量任务添加额外操作时会出现什么情况。

下面让我们来设置初始的批量操作。

一、设置批量任务

这个批量处理流程的设置和其他文章中的类似，它会启动一个批量处理流程，以每次 100 个项目的速度处理 1000 个项目。

  
    $batch = new BatchBuilder();
    $batch->setTitle('正在运行批量处理流程。')
      ->setFinishCallback([BatchClass::class, 'batchFinished'])
      ->setInitMessage('开始')
      ->setProgressMessage('正在处理...')
      ->setErrorMessage('处理过程中发生错误。');

    // 创建 10 组，每组 100 个项目。
    $chunks = array_chunk(range(1, 1000), 100);

    // 处理数组中的每个组。
    foreach ($chunks as $id => $chunk) {
      $args = [
        $id,
        $chunk,
        TRUE,
      ];
      $batch->addOperation([BatchClass::class, 'batchProcess'], $args);
    }
    batch_set($batch->toArray());
  

这里的关键区别在于，我们还会在参数中传入一个标志，该标志会让原始批量处理流程在运行时启动一个新的批量处理流程，在 Drupal 模块开发中，这种灵活的设置方式能满足更多复杂的业务需求。

有了这样的设置，我们就可以看看批量操作了。

二、添加到批量操作中

我们为 batchProcess() 方法设置的参数中有一个额外的 $addBatch 标志，对于所有初始批量操作，该标志都设为 TRUE。

在批量处理流程方法正常运行时，我们会检查这个标志，如果它为 TRUE，就使用 BatchBuilder 类启动一个新的批量处理流程。这里的主要区别是，我们在内部操作中将标志设为 FALSE，这样它们就不会启动额外的批量运行（否则会导致无限循环）。

其余的批量操作和之前的文章一样，我们只是掷骰子来决定操作的结果，以此模拟各种情况。

当我们调用 batch_set() 时，Drupal 会调用 batch_get() 查看是否有批量操作正在运行。如果有，就会调用 _batch_append_set()，它会把新的批量运行直接追加到正在运行的批量之后。这意味着我们能在当前流程运行期间设置并启动一个新的批量处理流程，而不会中断或干扰当前操作。新的操作将在当前流程结束时运行。

下面是 batchProcess() 函数中运行额外批量创建和 batch_set() 操作的部分。可以看到，它只是调用了我们通常用来启动批量处理流程的相同的 BatchBuilder 代码，只是这次是在处理方法内部被调用的。

  
    public static function batchProcess(int $batchId, array $chunk, bool $addBatch, array &$context): void {

      // -- 设置批量沙盒。

      if ($addBatch === TRUE) {
        // 对于每个 "原始 "批量运行，我们添加另一组数字进行处理。
        // 这模拟了在现有批量中生成额外的批量运行。
        $batch = new BatchBuilder();
        $batch->setTitle('正在运行批量处理流程。')
          ->setFinishCallback([BatchClass::class, 'batchFinished'])
          ->setInitMessage('开始')
          ->setProgressMessage('正在处理...')
          ->setErrorMessage('处理过程中发生错误。');

        // 添加一个新的组进行处理。这里将第三个参数设为 false
        // 意味着我们不会在这个内部启动另一个批量。
        $args = [
          $batchId + 1000,
          range(1, 100),
          FALSE,
        ];
        $batch->addOperation([BatchClass::class, 'batchProcess'], $args);

        batch_set($batch->toArray());
      }

      // --- 继续批量处理。
    }
  

这个批量操作完成后，我们将处理 2000 个项目，包括最初批量中的 1000 个项目和额外子批量处理运行中的 1000 个项目（原内容此处表述可能有误，推测是 10 组每组 100 个共 1000 个）。

额外批量运行在当前运行完成后执行，这意味着它们的完成方法也会单独运行。这会产生一个有趣的效果，即此操作完成后会打印出 11 条消息（初始批量运行 1 条，额外批量运行 10 条），让批量完成页面呈现出这样的状态。

你可以通过不打印最终总数消息或不在内部批量设置中传递完成回调来关闭此功能。我公司在示例中保留了这个问题，以便清晰展示实际情况。

这种批量添加功能是批量处理 API 的一个实用部分，这意味着任何批量操作都能被执行，即便 Drupal 当前正在运行一个批量任务。

三、这有什么用呢？

读到这里，我想你们有些人可能会想：“为什么要这样做呢？” 其实，这种技术看似有些深奥，但它确实有实际用途。如果你不清楚 “完成” 状态会是怎样，或者正在执行递归操作，那么添加额外的批量操作就非常有用。这意味着你不能用已知元素预加载批量任务，而且使用批量处理 API 的完成参数也不合适。

任何需要递归处理的批量任务都能采用这种方式设置，初始的起始数据只是操作期间将执行的已知进程的一部分。操作的递归性质决定了在批量处理开始时，你无法确定需要处理多少项，直到处理完这些项为止。

例如，我公司最近有个任务是从站点地图索引文件中导入数千个链接。sitemap.xml 文件是一个 XML 文档，它列出了网站上可用的链接，而且它还可能是一个指向其他 sitemap.xml 文件的索引文件，而这些文件反过来也可能是索引文件。

用户输入 sitemap.xml 索引文件的位置后，会触发一个批量处理流程，该流程会为索引中找到的每个额外的 sitemap.xml 索引文件触发额外的批量处理流程。其中一些 sitemap.xml 文件也是索引文件，所以我们需要启动额外的处理运行来处理这些文件。之后，sitemap.xml 文件中的所有链接都会保存到数据库表中，以便日后分析。

以下是解析 sitemap.xml 文件并创建额外批量操作或将链接保存到数据库的批量处理方法的关键部分。这里使用的库是我公司在执行此任务之前编写的 Sitemap Checker PHP 库。使用这个库大大加快了此任务的开发速度，因为它内置了大部分站点地图解析代码。这里的 “Link” 实体是一个简单的自定义实体，用于存储解析后的 URL 信息。

  
    $client = \Drupal::service('http_client');

    $sitemapSource = new SitemapXmlSource($client);
    $sitemapData = $sitemapSource->fetch($sitemap);

    if ($sitemapSource->isSitemapIndex()) {
      $sitemapIndexXmlParse = new SitemapIndexXmlParser();
      $sitemapList = $sitemapIndexXmlParse->parse($sitemapData);

      $batch = new BatchBuilder();
      $batch->setTitle('正在导入站点地图的处理过程')
        ->setFinishCallback([BatchLinkImport::class, 'batchFinished'])
        ->setInitMessage('开始')
        ->setProgressMessage('已完成 @current')
        ->setErrorMessage('处理过程中发生错误。');

      // 启动新的批量处理压缩站点地图中的项目。
      foreach ($sitemapList as $key => $sitemapUrl) {
        $args = [
          $key,
          $sitemapUrl->getRawUrl(),
        ];
        $batch->addOperation('\Drupal\link_audit\Batch\BatchLinkImport::processSitemapsBatch', $args);
      }
      batch_set($batch->toArray());
    }
    else {
      $sitemapParser = new SitemapXmlParser();
      $list = $sitemapParser->parse($sitemapData);

      foreach ($list as $url) {
        $context['results']['progress']++;

        $link = Link::create(
          [
            'label' => $url->getPath(),
            'url' => $url->getRawUrl(),
            'path' => $url->getPath(),
            'location' => $sitemap,
          ]
        );
        $link->save();
      }
    }
  

通过添加额外的将条目写入数据库的操作，而非在此处执行写入操作，这个流程可以进一步优化，提升 Drupal 开发的效率和性能。

运用这种技术，批量运行能够处理 40000 多个链接，且无需事先知晓有多少链接可供处理，处理过程中也不会出现超时或内存不足的情况。批量处理流程会持续处理 sitemap.xml 文件，直至没有更多的 sitemap.xml 文件或链接需要处理。

四、结论

在本文中，我们探讨了在批量处理流程已经启动的情况下如何设置额外的批量处理运行。尽管不建议在每个批量处理流程中都运用这种技术，但在某些特定使用场景下，它是批量处理工具包中非常有用的一部分。

如果你需要执行递归操作，采用这种额外的批量运行技术是个不错的选择。我公司发现这通常涉及某种 API 或网络调用，在操作开始时无法确定最终结果（除了没有更多的项目需要处理）。

这里需要重点理解的是，你创建的任何额外批量操作都会追加到当前批量运行的末尾。这意味着一旦当前批量运行启动，就无法对其进行更改。需要注意的是，如果你抛出异常，批量处理将完全停止。从底层的数据结构角度考虑，就会明白无法更改批量操作是合理的。Drupal 批量处理 API 是基于 Drupal 队列 API 构建的，这意味着一旦队列就位，就很难对其进行修改（至少在不直接进行数据库查询的情况下不容易修改，我公司不建议这样做）。

这个模块的所有代码都可以在相关示例模块中找到。你还可以查看 Drupal 批量处理示例仓库中本系列所有其他模块的源代码，在进行 Drupal 升级或开发 Drupal 11 项目时，这些资源都能提供很大的帮助。

在本系列的下一篇文章中，我们将探讨 Drupal 本身如何使用批量处理 API 来执行可能复杂的任务。