一、Drupal 11：使用 Drush 进行批量处理

2024年9月1日 - 阅读时长17分钟

这是探讨 Drupal 批量 API 的系列文章的第二部分。在 Drupal 开发里，批量 API 是一个很实用的系统，它允许分小批次处理数据，能有效防止出现超时错误或内存问题。

在上一篇文章中，我们探讨了如何使用表单来设置批量处理流程，当时批量处理方法包含在表单类的结构中。当表单提交时，批量处理流程会处理1000个项目，最后输出结果。

虽说将所有内容放在表单类中运行批量 API 没什么毛病，但通常把批量处理代码抽象到一个单独的类中会更合适。

使用一个单独的批量类来包含处理和完成方法，是设置这些流程的更佳方式。因为这样可以将批量处理流程与启动它的操作分离开来，这意味着我们可以从任何地方启动批量处理，甚至能通过 Drush 命令来启动，这在 Drupal 模块开发中是很实用的技巧。

允许通过 Drush 运行批量处理流程，是模块具备的一个非常强大的特性。这意味着用户可以运行的任何大型流程都能通过 Drush 命令自动运行。

二、批量类

要创建一个批量类，我公司通常会在模块的“/src”目录下创建一个名为“Batch”的目录，在其中定义所需的任何批量类。这个类的内容是之前表单类中使用的两个批量方法，即 batchProcess() 和 batchFinished() 方法。

以下是这个类的基本结构。

<?php

namespace Drupal\batch_class_example\Batch;

/**
 * 定义批量处理的处理和完成方法。
 */
class BatchClass {

  /**
   * 处理批量操作。
   *
   * @param int $batchId
   *   批量处理 ID。
   * @param array $chunk
   *   要处理的数据块。
   * @param array $context
   *   批量处理上下文。
   */
  public static function batchProcess(int $batchId, array $chunk, array &$context): void {
    // 在这里处理批量操作...
  }

  /**
   * 处理批量处理完成后的操作。
   *
   * @param bool $success
   *   如果所有批量 API 任务都成功完成，则为 TRUE。
   * @param array $results
   *   批量处理操作的结果数组。
   * @param array $operations
   *   未完成的操作列表。
   * @param string $elapsed
   *   batch.inc 会贴心地提供已用的处理时间（秒）。
   */
  public static function batchFinished(bool $success, array $results, array $operations, string $elapsed): void {
    // 在这里完成批量处理。
  }

}

我公司完全没有更改这两个方法的内部代码，因此这里不再重复列出。如果想获取代码，可以查看上一篇关于批量 API 的文章。这个批量 API 示例以及本系列中其他批量 API 文章的所有源代码都能找到。

三、通过表单运行批量类

现在我们已经有了批量类，就可以重新编写表单提交处理程序以使用这个类。这只是将 setFinishCallback() 和 addOperation() 方法指向新类的问题。

这样做会将运行批量处理代码的表单提交处理程序修改如下。

public function submitForm(array &$form, FormStateInterface $form_state): void {
  $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,
    ];
    $batch->addOperation([BatchClass::class, 'batchProcess'], $args);
  }
  batch_set($batch->toArray());

  $form_state->setRedirectUrl(new Url($this->getFormId()));
}

当我们提交表单时，批量操作会正常运行，这里唯一的区别是批量操作的运行位置。

四、通过 Drush 运行批量处理

现在我们已经将批量处理运行代码分离到一个单独的类中，就可以通过 Drush 命令运行相同的代码。在进行 Drupal 升级或者开发新功能时，这种方式能提高效率。

要设置一个 Drush 命令，我们需要在模块的根目录下创建一个 drush.services.yml 文件。这个文件看起来非常像我们在 Drupal 8+ 中用于创建自定义服务的普通模块 *.services.yml 文件。

将以下代码添加到 drush.services.yml 文件中，这将设置一个 Drush 类，该类会将命令注入到 Drush 命名空间中。我们还注入了 logger.factory 服务，以便在命令中记录信息。

services:
  batch_class_example.commands:
    class: \Drupal\batch_class_example\Commands\BatchCommands
    tags:
      - { name: drush.command }
    arguments: ['@logger.factory']

Drush 类包含一个单一的方法（除了注入日志服务的样板代码之外），该方法定义了 Drush 命令。我们使用一些注解来告诉 Drush 我们定义了一个命令。

/**
 * 通过 Drush 命令运行批量操作。
 *
 * @command batch_class_example:run
 *
 * @validate-module-enabled batch_class_example
 *
 * @usage batch_class_example:run
 */
public function runBatchclassExample() {
}

有了这个设置，我们就可以运行 drush batch_class_example:run，这个方法就会被执行，所以让我们添加批量处理设置代码。

要设置和运行批量处理，我们只需从表单类中复制批量处理设置代码。本质上，我们以相同的方式设置批量处理，并且批量处理将以类似的方式执行。在 Drush 中设置批量处理时的关键区别在于，我们还添加了对 drush_backend_batch_process() 函数的调用。这是一个 Drush 内部函数，它会逐步处理批量处理直到完成，而不是通过逐步加载页面来处理批量处理。

以下是完整的 BatchCommands 类。

<?php

namespace Drupal\batch_class_example\Commands;

use Drupal\batch_class_example\Batch\BatchClass;
use Drupal\Core\Batch\BatchBuilder;
use Drupal\Core\Logger\LoggerChannelFactoryInterface;
use Drupal\Core\StringTranslation\StringTranslationTrait;
use Drush\Commands\DrushCommands;

/**
 * batch_class_example 模块的 Drush 命令。
 */
class BatchCommands extends DrushCommands {

  use StringTranslationTrait;

  /**
   * 日志服务。
   *
   * @var \Drupal\Core\Logger\LoggerChannelFactoryInterface
   */
  private $loggerChannelFactory;

  /**
   * 构造一个新的 BatchCommands 对象。
   *
   * @param \Drupal\Core\Logger\LoggerChannelFactoryInterface $loggerChannelFactory
   *   日志服务。
   */
  public function __construct(LoggerChannelFactoryInterface $loggerChannelFactory) {
    $this->loggerChannelFactory = $loggerChannelFactory;
  }

  /**
   * 通过 Drush 命令运行批量操作。
   *
   * @command batch_class_example:run
   *
   * @validate-module-enabled batch_class_example
   *
   * @usage batch_class_example:run
   */
  public function runBatchclassExample() {
    $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,
      ];
      $batch->addOperation([BatchClass::class, 'batchProcess'], $args);
    }
    batch_set($batch->toArray());

    drush_backend_batch_process();

    // 完成。
    $this->logger()->notice("批量操作结束。");
    $this->loggerChannelFactory->get('batch_class_example')->info('批量操作结束。');
  }

}

批量处理方法的最后一步是向用户发送消息形式的反馈，并记录批量处理已完成。

现在我们可以在 Drush 中运行批量操作，它将输出以下消息。

$ drush batch_class_example:run
>  [通知] 正在处理批次 #0，批次大小为 100，共 1,000 个项目。
>  [通知] 正在处理批次 #1，批次大小为 100，共 1,000 个项目。
>  [通知] 正在处理批次 #2，批次大小为 100，共 1,000 个项目。
>  [通知] 正在处理批次 #3，批次大小为 100，共 1,000 个项目。
>  [通知] 正在处理批次 #4，批次大小为 100，共 1,000 个项目。
>  [通知] 正在处理批次 #5，批次大小为 100，共 1,000 个项目。
>  [通知] 正在处理批次 #6，批次大小为 100，共 1,000 个项目。
>  [通知] 正在处理批次 #7，批次大小为 100，共 1,000 个项目。
>  [通知] 正在处理批次 #8，批次大小为 100，共 1,000 个项目。
>  [通知] 正在处理批次 #9，批次大小为 100，共 1,000 个项目。
>  [通知] 消息：类批量处理完成，处理了 1000 个项目，跳过 272 个，更新 504 个，失败 224 个，用时
> 0 秒。
>
 [通知] 批量操作结束。

和上一篇文章一样，这个批量命令只是处理1000个项目，并随机决定运行结果，而不会对系统上的数据进行任何更改。这样做是有意的，以便可以多次运行这个批量示例，而不会在 Drupal 11 网站上添加或删除数据。

五、结论

Drupal 中的批量操作本身就很有用，但通过将它们的处理和完成方法与设置它们的位置分离开来，我们可以在任何需要的地方运行批量操作。这样做可以让系统用户在命令行上自动化任务，从而无需用户坐在那里等待批量处理页面完成。

这通常是我公司在使用批量 API 时的设置方式。

我们甚至可以更进一步，将批量操作的设置集中在一个地方，比如一个服务，然后从表单和 Drush 命令中调用这个服务。这意味着当我们想要启动批量处理时，不需要复制和粘贴任何代码。

在本系列的下一篇文章中，我们将探讨批量处理器方法的完成状态，这将使我们能够根据需要长时间运行批量处理。这与我目前展示的情况不同，在目前的情况中，我们设置批量运行的初始条件，然后让它执行到完成。

如果想在自己的项目中试验和使用这些代码，可以找到这个模块的所有源代码。源代码是一个模块的子模块，该模块包含本系列中所有关于 Drupal 批量 API 文章的所有源代码，在 Drupal 模块开发中可以很好地利用这些资源。