Drupal 11：在批量处理中使用完成状态
这是Drupal批量API系列文章中的第三篇。在Drupal开发里，批量API是一个非常实用的系统，它允许我们将数据分成小块进行处理，这样能有效防止出现超时错误或内存问题。
到目前为止，在这个系列中，我们已经知道了如何使用表单创建批量流程，还学会了创建批量类，从而可以通过Drush运行批量处理。这两个示例都是借助批量API，通过一定数量的处理函数回调来运行一定数量的项。在设置批量运行时，我们先创建一个想要处理的项列表，然后把这个列表分成小块，每个小块都会被发送到一个批量处理回调中。
其实，还有另一种设置批量API的方法，它能运行相同数量的操作，而且无需事先定义要运行的次数。这可以通过在批量上下文中使用“finished”设置来实现。
接下来，让我们一起创建一个可以使用完成设置来运行和控制的批量处理流程。
一、设置
首先，我们得创建一个批量处理流程，这个流程会接受我们想要处理的数组。这个数组和我们在前两篇文章中处理的数组是一样的，但这次我们通过BatchBuilder类的addOperation()方法，将整个数组传递给一个单独的回调。
$batch = new BatchBuilder();
$batch->setTitle('Running batch process.')
->setFinishCallback([BatchClass::class, 'batchFinished'])
->setInitMessage('Commencing')
->setProgressMessage('Processing...')
->setErrorMessage('An error occurred during processing.');
$array = range(1, 1000);
$batch->addOperation([BatchClass::class, 'batchProcess'], [$array]);
batch_set($batch->toArray());由于我们更改了传递给batchProcess()方法的参数，所以需要对之前示例中的函数签名稍微做些修改。在之前的示例中，第一个参数是我们正在运行的批量处理的ID，后面跟着我们需要处理的数组块。
因为我们只将一个数组作为批量处理的参数传递，所以我们只需从批量回调中接受数组和$context参数就行。
public static function batchProcess(array $array, array &$context): void {
}$context数组在正常操作期间用于报告处理进度很有帮助，但这次它对于正确控制批量处理流程起着至关重要的作用。
批量处理的其他方面保持不变。我们依旧使用在前两篇文章中定义的相同的完成方法。
二、运行批量处理
当我们首次启动批量操作时，$context数组包含以下项。要记住，这个数组是通过引用传递的，所以对它所做的任何更改对于调用它的代码都是可见的。
Array(
[sandbox] => Array()
[results] => Array()
[finished] => 1
[message] =>
)默认情况下，这里的完成操作设置为1，这就意味着当到达批量处理方法的末尾时，批量API将不会再次运行它。如果我们设置一个大于或等于1的值，那么该操作将从批量队列中移除，处理流程要么进入下一个操作，要么整个运行过程结束，然后我们调用完成方法。
这里的关键是设置一个小于1的值。要是我们这样做了，批量处理将不会从队列中移除该操作，并且批量处理流程会再次调用处理方法。如果选择这种操作方式，我们就需要确保跟踪批量操作的进度，这样才能判断是否已经完成了批量操作。
当首次调用批量处理时，$context数组的“sandbox”部分为空，我们可以利用这个数组来跟踪进度以及还剩下多少项需要处理。
public static function batchProcess(array $array, array &$context): void {
if (!isset($context['sandbox']['progress'])) {
$context['sandbox']['progress'] = 0;
$context['sandbox']['max'] = count($array);
}
}假设我们想一次处理100项，就像我们在前几篇批量处理文章中做的那样。在下面的代码中，我们把想要处理的项数设置在$batchSize变量中，然后使用这个变量在一个简单的for循环中遍历这些项；使用批量沙箱设置中的当前进度值。
$batchSize = 100;
for ($i = $context['sandbox']['progress']; $i < $context['sandbox']['progress'] + $batchSize; $i++) {
$context['results']['progress']++;
}一旦循环完成，我们就可以更新批量沙箱中的进度设置，以包含我们刚刚处理的项数。
// Keep track of progress.
$context['sandbox']['progress'] += $batchSize;我们在batchProcess()方法中要做的最后一件事是更新完成设置，这可以通过几种方式来实现。
我们可以简单地比较进度和最大值，如果还没有达到处理的末尾，就将完成值设置为0。
if ($context['sandbox']['progress'] <= $context['sandbox']['max']) {
$context['finished'] = 0;
}或者，我们可以直接将进度计数除以最大计数，如果进度小于最大计数，这会导致完成值小于1。
$context['finished'] = $context['sandbox']['progress'] / $context['sandbox']['max'];注意！使用此方法时要小心，因为如果最大值为零，可能会导致除零错误。
把所有这些放在一起，我们得到的batchProcess方法如下所示。
public static function batchProcess(array $array, array &$context): void {
if (!isset($context['sandbox']['progress'])) {
$context['sandbox']['progress'] = 0;
$context['sandbox']['max'] = count($array);
}
if (!isset($context['results']['updated'])) {
$context['results']['updated'] = 0;
$context['results']['skipped'] = 0;
$context['results']['failed'] = 0;
$context['results']['progress'] = 0;
$context['results']['process'] = 'Finish batch completed';
}
// Message above progress bar.
$context['message'] = t('Processing batch @progress of total @count items.', [
'@progress' => number_format($context['sandbox']['progress']),
'@count' => number_format($context['sandbox']['max']),
]);
$batchSize = 100;
for ($i = $context['sandbox']['progress']; $i < $context['sandbox']['progress'] + $batchSize; $i++) {
$context['results']['progress']++;
// Sleep for a bit to simulate work being done.
usleep(4000 + $array[$i]);
// Decide on the result of the batch.
$result = rand(1, 4);
switch ($result) {
case '1':
case '2':
$context['results']['updated']++;
break;
case '3':
$context['results']['skipped']++;
break;
case '4':
$context['results']['failed']++;
break;
}
}
// Keep track of progress.
$context['sandbox']['progress'] += $batchSize;
$context['finished'] = $context['sandbox']['progress'] / $context['sandbox']['max'];
}有了这些设置，我们将以与前两个示例完全相同的方式处理包含1000项的相同数组，每次批量处理运行处理100项。
为了更清楚地说明这里发生的情况，让我们逐步了解批量处理过程。
- 首次调用批量处理。
- 将最大值设置为1000，并处理数组中的100项。
- 将进度计数更新为100。
- 将完成值设置为100/1000（即0.1）。
- 由于这个数字小于1，处理操作将再次被调用。
- 第二次批量运行。
- 将进度计数更新为200。
- 将完成值设置为200/1000（即0.2）。
- 由于这个数字小于1，处理操作将再次被调用。
- 第三次批量运行。
- 将进度计数更新为300。
- 将完成值设置为300/1000（即0.3）。
- 由于这个数字小于1，处理操作将再次被调用。
- 跳过几次迭代...
- 第十次批量运行。
- 将进度计数更新为1000。
- 将完成值设置为1000/1000（即1）。
- 由于这个数字是1，处理操作被视为完成，并从批量处理中移除。
- 运行完成回调方法，将批量处理的结果传递给回调。
我公司在Drupal模块开发中所涉及的这里看到的所有代码，都可以在相关的Drupal批量处理示例仓库中找到，完整的示例作为可用的子模块之一存在。你可以自由地将此作为你自己的批量处理方法的基础。
三、实际示例
当然，处理数字相对简单，让我们用完成设置做一些更有趣的事情。
假设我们想要处理一些节点以对它们执行操作。这可能是以某种方式更新它们，甚至删除它们。我们不是在开始时加载所有想要处理的节点，而是在不加载任何内容或不向addOperation()方法传递任何参数的情况下启动批量处理。
$batch = new BatchBuilder();
$batch->setTitle('Running batch process.')
->setFinishCallback([BatchClass::class, 'batchFinished'])
->setInitMessage('Commencing')
->setProgressMessage('Processing...')
->setErrorMessage('An error occurred during processing.');
$batch->addOperation([BatchClass::class, 'batchProcess']);
batch_set($batch->toArray());这意味着我们的batchProcess()回调将仅接受$context数组作为单个参数。在处理回调中，我们要做的第一件事是通过执行计数查询来找出目标服务的大小，以统计系统中存在的节点数量。然后将得到的$count变量设置为我们的最大值。
public static function batchProcess(array &$context): void {
if (!isset($context['sandbox']['progress'])) {
$query = \Drupal::entityQuery('node');
$query->accessCheck(FALSE);
$count = $query->count()->execute();
$context['sandbox']['progress'] = 0;
$context['sandbox']['max'] = $count;
}然后，我们可以将实体分成每组10项的小批量进行处理。进度计数是我们指向最后一个处理项的指针，所以我们使用它来确定从数据库中加载的ID范围。使用这些ID，我们可以加载实际的节点，并以我们想要的任何方式处理它们。
$batchSize = 10;
$storage = \Drupal::entityTypeManager()->getStorage('node');
$query = $storage->getQuery();
$query->accessCheck(FALSE);
$query->range($context['sandbox']['progress'], $batchSize);
$ids = $query->execute();
foreach ($storage->loadMultiple($ids) as $entity) {
// Keep track of progress.
$context['sandbox']['progress']++;
$context['results']['progress']++;
// Process the entity here, for example, we might run $entity->delete() to
// delete the entity.
}
$context['finished'] = $context['sandbox']['progress'] / $context['sandbox']['max'];通过这种方式，我们不会在批量处理开始时花费很长时间加载想要处理的项。相反，节点每次最多加载10项。在Drupal开发场景中，这种方式能有效提升处理效率，应对大规模数据处理。
四、结论
在本文中，我们了解了如何在Drupal 11开发里使用提供的批量上下文中的完成设置来控制批量处理流程。此设置允许我们通过在批量处理中设置一个值来控制批量操作的流程。完成值可用于运行开放式批量操作，在这种操作中，我们不需要在处理开始时加载所有要处理的项。
这是我公司使用批量操作的首选方法，尤其是在涉及实体类型时。加载要处理的实体列表的批量操作在处理少量项时效果很好。问题是，当你向数据库中添加10万个甚至100万个记录时，批量处理往往在开始之前就会失败。在过去的几年里，我公司已经将一些批量操作转换为使用完成系统，所以在编写批量操作时，总是尽量从这个原则开始。在实际进行Drupal升级操作时，运用这种灵活的批量处理方式也能很好地解决数据处理难题。
我公司在Drupal模块开发中涉及的这里的所有代码，都可以在相关的Drupal批量处理示例仓库中找到。如果你觉得这些文章或代码有用，请告诉我们。
在下一篇文章中，我们将了解如何使用批量API和完成设置来处理任意长度的CSV文件。
