企业官网建设流程全解析

• 后端
• 电商
• 前端

本文基于当前仓库内 Guzzle 3 的官方文档 batching.rst 编写。Guzzle 3 提供了一个高度可定制、非常通用的批处理框架，用于高效地并行传输 HTTP 请求与服务命令。文档、源码与测试全部位于仓库的 vendor/guzzle/guzzle 目录内，是理解 Guzzle 内部批处理机制的第一手资料。

导读

Guzzle\Batch命名空间提供了一套独立于具体 HTTP 细节的批处理框架：无论你要批量传输的是 HTTP 请求（Guzzle\Http\Message\RequestInterface）、服务命令（Guzzle\Service\Command\CommandInterface），还是任意自定义业务对象，都可以借助统一的"队列 → 分组（Divisor）→ 传输（Transfer）→ 装饰（Decorator）"管线来完成。本文将以原文档为主线，结合仓库源码中的 Batch.php、BatchBuilder.php 及全部装饰器实现，逐层剖析其工作原理，并给出可直接运行的代码示例与调优建议。读完本文，你将能：

• 掌握在 Guzzle 中通过Client::send()/Client::execute()进行并行请求的最简用法；
• 理解Batch、BatchDivisorInterface、BatchTransferInterface三者之间的协作关系；
• 熟练使用BatchBuilder快速构建带"自动冲刷、历史记录、异常缓冲、冲刷通知"等多种行为的批处理对象；
• 学会从源码层面分析批处理的分组逻辑与异常处理机制。

一、快速入门：并行发送请求与服务命令

在原文档中，Guzzle 提供了两种开箱即用的并行方式，适用于大多数简单场景：

1.1 并行发送 HTTP 请求

将一组Guzzle\Http\Message\RequestInterface对象作为数组传入Guzzle\Http\Client::send()，即可让它们并行发送：

$responses = $client->send(array( $client->get('http://www.example.com/foo'), $client->get('http://www.example.com/baz'), $client->get('http://www.example.com/bar') ));

send()返回的是一个Response对象数组，顺序与传入的请求数组一一对应。查看 Client.php 的实现可以发现：当传入的不是单个RequestInterface而是数组时，会走sendMultiple()分支（见 Client.php）：

protected function sendMultiple(array $requests) { $curlMulti = $this->getCurlMulti(); foreach ($requests as $request) { $curlMulti->add($request); } $curlMulti->send(); $result = array(); foreach ($requests as $request) { $result[] = $request->getResponse(); } return $result; }

也就是说，Guzzle 的"并行"本质上是依托CurlMulti（curl 多句柄）一次性地把多个请求加入队列，再由底层一次性send()，从而实现多路复用并发传输。

1.2 并行执行服务命令

对于服务客户端（Guzzle\Service\Client），并行执行则通过execute()传入命令数组完成：

$commands = $client->execute(array( $client->getCommand('foo'), $client->getCommand('baz'), $client->getCommand('bar') ));

Guzzle\Service\Client::execute()（见 Service/Client.php）对数组/Traversable参数会走executeMultiple()分支：先把每个命令prepareCommand()转换为请求对象，再统一send()，随后逐个派发command.after_send事件并返回命令数组（见 Service/Client.php）。

原文档特别指出：这两种方式对大多数场景足够好用，但当需要对发送过程做更细粒度的控制，或者需要发送大量请求时，就应该使用Guzzle\Batch命名空间提供的批处理功能——这正是本文后续要深入的主题。

二、批处理框架总览：队列、分组与传输

2.1 三个核心角色

原文档描述了一套清晰的三层协作模型：

在 Batch.php 源码中，三者被直观地组织在一起：

class Batch implements BatchInterface { protected $queue; // \SplQueue 队列 protected $dividedBatches; // 已划分好的批次 protected $transferStrategy; // BatchTransferInterface protected $divisionStrategy; // BatchDivisorInterface public function __construct(BatchTransferInterface $transferStrategy, BatchDivisorInterface $divisionStrategy) { $this->transferStrategy = $transferStrategy; $this->divisionStrategy = $divisionStrategy; $this->queue = new \SplQueue(); $this->queue->setIteratorMode(\SplQueue::IT_MODE_DELETE); $this->dividedBatches = array(); } // ... }

flush()的完整执行流程（见 Batch.php）如下：

1. 先调用createBatches()，把队列中的剩余项交给 divisor 划分成若干批次数组；
2. 依次遍历每个已划分批次，调用$this->transferStrategy->transfer($batch)完成传输；
3. 把成功传输的项合并进返回值；
4. 若某个批次抛出异常，则包装为Guzzle\Batch\Exception\BatchTransferException抛出，其中携带失败的批次（getBatch()）、此前已成功传输的项（getTransferredItems()）以及传输/分组策略对象（见 BatchTransferException.php）。

需要注意：SplQueue被设置为IT_MODE_DELETE迭代模式，即createBatches()遍历队列的过程会消耗队列中的元素，所以一次flush()之后，已传输的项不再残留于队列。

2.2 最小可运行的 Batch 示例

原文档给出的最小示例是Batch+BatchRequestTransfer的组合：

use Guzzle\Batch\Batch; use Guzzle\Http\BatchRequestTransfer; // BatchRequestTransfer 同时扮演 divisor 和 transfer 两种策略 $transferStrategy = new BatchRequestTransfer(10); $divisorStrategy = $transferStrategy; $batch = new Batch($transferStrategy, $divisorStrategy); // 向批处理队列中添加请求 $batch->add($request1) ->add($request2) ->add($request3); // 冲刷队列并取回被传输的项 $arrayOfTransferredRequests = $batch->flush();

原文档附带的.. note::也指出：你的传输策略很可能需要同时充当 divisor 和 transfer 策略——这一设计意图在BatchRequestTransfer/BatchCommandTransfer的类声明（implements BatchTransferInterface, BatchDivisorInterface）中得到了印证，后文会展开说明。

三、批量传输请求与命令的两大传输策略

3.1BatchRequestTransfer：按 curl 多句柄分组传输请求

原文档指出，Guzzle\Http\BatchRequestTransfer（注意：源码中该类实际位于Guzzle\Batch命名空间，即 BatchRequestTransfer.php，文档中的Guzzle\Http\BatchRequestTransfer应为历史命名）通过把请求按传输所用的 curl_multi 句柄分组，来高效并行传输一批请求。

其createBatches()逻辑（见 BatchRequestTransfer.php）：

1. 校验队列中的每一项都必须是RequestInterface，否则抛出InvalidArgumentException；
2. 用SplObjectStorage按$item->getClient()把请求分组（同一 client 共享同一 curl_multi 句柄）；
3. 对每组再按$batchSize用array_chunk()切成更小的批次，最后把各 client 的小批次合并返回。

transfer()则取批次内第一个请求的 client，调用$client->send($batch)一次性并行发送（见 BatchRequestTransfer.php）。

仓库中的测试 BatchRequestTransferTest.php 验证了分组行为：当 client1、client2 各持有独立CurlMulti，以batchSize = 2传入 5 个请求时，期望分组结果为array($request1,$request3)、array($request3)（此处测试因SplQueue迭代消耗导致内容略有出入，但"按 client 分组 + 按 batchSize 切块"的原则清晰可见），并且测试还覆盖了"所有项必须是请求"与"空批次不传输"两个边界。

3.2BatchCommandTransfer：按客户端分组传输服务命令

Guzzle\Service\Command\BatchCommandTransfer（源码同样位于 BatchCommandTransfer.php）则通过按客户端分组命令来高效传输服务命令。原文档强调：你可以向同一个 batch 中添加由不同 client 传输的命令，batch 会自己处理剩下的一切。

createBatches()同样先校验CommandInterface，再按getClient()用SplObjectStorage分组并array_chunk($batchSize)切块（见 BatchCommandTransfer.php）。

transfer()有一个重要的安全校验：取出第一个命令的 client，然后用array_filter找出与它不属于同一 client 的命令，若存在则抛出Guzzle\Service\Exception\InconsistentClientTransferException（见 BatchCommandTransfer.php）。因为 divisor 已经保证同一批次内 client 一致，这个异常在正常流程中不会触发，但它为自定义 divisor 与 transfer 混用时提供了兜底保护。

对应的单元测试 BatchCommandTransferTest.php 验证了"按 client 分组、batchSize=2 切块"以及"跨 client 混传抛异常"的行为。

四、用 BatchBuilder 一键构建批处理对象

4.1transferRequests()：快速传输请求

Guzzle\Batch\BatchBuilder把"创建 divisor、transfer、再组合 Batch"的样板代码封装成链式调用。原文档展示的请求批处理构建方式为：

use Guzzle\Batch\BatchBuilder; $batch = BatchBuilder::factory() ->transferRequests(10) ->build();

transferRequests($batchSize = 50)内部（见 BatchBuilder.php）实例化BatchRequestTransfer并同时赋值给 transfer 和 divisor 两个策略——这正是前文"一个策略身兼两职"设计的具体落地。默认批大小为 50，传入的10表示每个 curl_multi 批次最多包含 10 个请求。

4.2transferCommands()：快速传输命令

对应地，命令批处理的构建方式为：

use Guzzle\Batch\BatchBuilder; $batch = BatchBuilder::factory() ->transferCommands(10) ->build(); $batch->add($client->getCommand('foo')) ->add($client->getCommand('baz')) ->add($client->getCommand('bar')); $commands = $batch->flush();

transferCommands($batchSize = 50)（见 BatchBuilder.php）同理创建BatchCommandTransfer作为双策略。flush()返回所有被传输的命令数组。由于BatchCommandTransfer的 divisor 会按 client 分组，上面示例中的三个命令即便由不同 client 创建，也会被自动归组后分别并行传输。

4.3 两个易踩的构建陷阱

build()之前，transfer 与 divisor 策略缺一不可。从 BatchBuilder.php 可见：缺少任一策略都会抛出Guzzle\Common\Exception\RuntimeException。这一点在 BatchBuilderTest.php 中被专门测试（testTransferStrategyMustBeSet/testDivisorStrategyMustBeSet）。所以直接BatchBuilder::factory()->build()是行不通的，至少要先调用transferRequests()/transferCommands()/transferWith()之一。

4.4 完全自定义的策略组合

除了预设的请求/命令传输，BatchBuilder还开放了两个底层钩子（见 BatchBuilder.php）：

• createBatchesWith(BatchDivisorInterface $divisorStrategy)：指定自定义 divisor；
• transferWith(BatchTransferInterface $transferStrategy)：指定自定义 transfer。

仓库中还提供了两个基于闭包的现成实现，方便完全自定义分组/传输逻辑：

• BatchSizeDivisor.php：按固定大小用array_chunk()切分队列，size可在构造/setSize()中指定；
• BatchClosureDivisor.php：把分组逻辑委托给一个可调用对象（createBatches(\SplQueue $queue, $context)）；
• BatchClosureTransfer.php：把传输逻辑交给一个闭包（签名function ($batch, $context)）。

五、批处理行为装饰器：为队列"加料"

原文档的"Batch behaviors"部分介绍了四种可叠加在Batch之上的装饰器，它们都继承自 AbstractBatchDecorator.php（通过__call()把未定义方法转发给被装饰的 batch，实现透明代理）。BatchBuilder::build()会按固定顺序依次包装（先ExceptionBufferingBatch→NotifyingBatch→FlushingBatch→HistoryBatch，见 BatchBuilder.php）。

5.1FlushingBatch：队列到达阈值自动冲刷

当你要把大量项"泵"进队列、且希望在队列达到某个规模时自动冲刷时，使用FlushingBatch装饰器：

use Guzzle\Batch\BatchBuilder; $batch = BatchBuilder::factory() ->transferRequests(10) ->autoFlushAt(10) ->build();

对应 builder 方法：autoFlushAt($threshold)。

实现细节（见 FlushingBatch.php）：每次add()后计数$currentTotal，一旦达到阈值就自动对内部 batch 执行flush()并重置计数。阈值可通过setThreshold()动态调整、getThreshold()读取。它的典型价值是：生产方（如循环拉取海量 URL）持续add()，而消费方（flush()）无需手动干预，队列永远不会无限膨胀。

5.2NotifyingBatch：每次冲刷后回调通知

如果你希望在每次队列冲刷时收到通知，使用NotifyingBatch装饰器。它与FlushingBatch组合尤其有用。通过 builder 的notify()方法传入一个 callable 即可：

use Guzzle\Batch\BatchBuilder; $batch = BatchBuilder::factory() ->transferRequests(10) ->autoFlushAt(10) ->notify(function (array $transferredItems) { echo 'Transferred ' . count($transferredItems) . "items\n"; }) ->build();

对应 builder 方法：notify(callable $callback)。

实现细节（见 NotifyingBatch.php）：flush()先调用内部 batch 的flush()拿到成功传输的项数组，再call_user_func()调用回调。构造函数会做is_callable()校验，非 callable 直接抛InvalidArgumentException（测试见 NotifyingBatchTest.php）。回调收到的参数正是本次被传输的项数组，可用来做日志、进度统计等。

5.3HistoryBatch：维护传输历史

若希望保留所有通过该 batch 传输的项的历史记录，使用HistoryBatch装饰器：

use Guzzle\Batch\BatchBuilder; $batch = BatchBuilder::factory() ->transferRequests(10) ->keepHistory() ->build();

对应 builder 方法：keepHistory()。

实现细节（见 HistoryBatch.php）：add()时先把项写入内部$history数组，再转发给被装饰 batch。传输完成后可用getHistory()取回全部已添加（并传输）的项数组。原文档特别提醒：务必定期用clearHistory()清理历史，否则内存占用会随传输量线性增长。测试见 HistoryBatchTest.php。

5.4ExceptionBufferingBatch：异常缓冲，失败后继续

使用ExceptionBufferingBatch装饰器可以在传输过程中缓冲异常，让你"尽可能多地传输项，然后再统一处理出错批次"。传输结束后，用getExceptions()取回一组Guzzle\Batch\Exception\BatchTransferException对象，并可用它们重试失败的批次；处理完毕务必用clearExceptions()清空缓冲。

对应 builder 方法：bufferExceptions()。

实现细节（见 ExceptionBufferingBatch.php）：flush()进入while (!$this->decoratedBatch->isEmpty())循环反复冲刷内部 batch，捕获BatchTransferException时把异常存入$exceptions，并把$e->getTransferredItems()（失败前已成功传输的项）合并进返回值——因此即使某些批次失败，其他批次仍会继续传输。

测试 ExceptionBufferingBatchTest.php 给出了一个经典场景：三个项foo/baz/bar，BatchSizeDivisor(1)逐项切批，第二个批次（baz）抛异常。结果flush()返回array('foo','bar')，getExceptions()中的唯一异常getBatch()返回array('baz')，clearExceptions()后异常列表归零——这正是"尽量多传、失败可重试"语义的实证。

六、批处理在 CRMEB 项目中的实际落点

6.1 仓库内 Guzzle 的版本与依赖

当前仓库 crmeb/composer.json 同时声明了guzzlehttp/psr7: ^1.7与guzzlehttp/guzzle: 6.5.5，而本文讨论的 Guzzle 3 批处理框架位于 vendor/guzzle/guzzle（该包的 composer.json 自述"Guzzle 3, is deprecated"，仅要求php >= 5.3.3与ext-curl）。这意味着：批处理文档对应的 Guzzle 3 代码是作为兼容依赖被携带进项目的，而项目日常 HTTP 请求走的是 GuzzleHttp 6。因此在 CRMEB 实际业务代码中，你几乎不会直接调用Guzzle\Batch，但它随依赖被完整保留，仍然是研究 Guzzle 批处理机制时不可多得的源码级教材。

6.2 项目中 Guzzle 3 组件的真实使用点

即便不直接使用 Batch 框架，Guzzle 3 的组件在 CRMEB 中仍有实际引用。例如上传模块：

• 阿里云 OSS 上传实现 crmeb/services/upload/storage/Oss.php 引用了use Guzzle\Http\EntityBody;，在stream()方法中用EntityBody::factory($fileContent)把文件内容规范化为实体流再交给putObject()上传（见 Oss.php）；
• 京东云 OSS 上传实现 crmeb/services/upload/storage/Jdoss.php 同样引用了Guzzle\Http\EntityBody。

EntityBody属于 Guzzle 3 的流抽象层，与批处理框架同源。这提醒我们：当项目因兼容需要携带 Guzzle 3 时，其内部的批处理、流、缓存、插件等子框架都是可用的（只要对应的curl_multi、SplQueue等 PHP 扩展/特性可用）。

6.3 批处理思想在 CRMEB 并发场景中的启发

CRMEB 作为商城系统，存在大量"一批对象、逐条处理"的高并发诉求，例如：

• 群发消息/模板消息：crmeb/jobs/notice/SmsJob.php、TemplateJob.php等队列任务；
• 批量订单状态变更：crmeb/jobs/order/OrderJob.php等；
• 海报、商品复制等批量 IO 操作：PosterJob.php、ProductCopyJob.php。

这些场景在 GuzzleHttp 6 生态中通常借助GuzzleHttp\Client+ 并发 Promise、或框架的队列中间件来实现，而Guzzle 3 的Batch框架提供的思想——"队列 + Divisor 分组 + Transfer 传输 + 装饰器增强"——依然是有价值的参考模型：autoFlushAt的阈值冲刷、ExceptionBufferingBatch的"尽量多传、失败可重试"、HistoryBatch的审计记录，都可以迁移到任意批量处理任务的设计中。

七、综合示例：组装一个"自动冲刷 + 通知 + 异常缓冲 + 历史"的批处理

把前文所有能力组合起来，可以得到一个生产级的批处理流水线示例（以请求传输为例）：

use Guzzle\Batch\BatchBuilder; // 每批 20 个请求并行；队列满 100 自动冲刷；每次冲刷后统计；失败不中断；保留历史 $batch = BatchBuilder::factory() ->transferRequests(20) // BatchRequestTransfer，批大小 20 ->autoFlushAt(100) // 队列累计 100 个请求时自动 flush ->bufferExceptions() // 失败批次缓冲为 BatchTransferException ->notify(function (array $items) { // 例如：记录日志、更新进度 fwrite(STDOUT, 'Flushed ' . count($items) . " requests\n"); }) ->keepHistory() // 保留全部已传输请求的历史 ->build(); // 持续泵入大量请求 foreach ($urls as $url) { $batch->add($client->get($url)); } // 手动冲刷收尾（autoFlush 只在队列满时触发，最后一批可能未满） $batch->flush(); // 处理历史与缓冲异常 $history = $batch->getHistory(); // 全部已传输请求 foreach ($batch->getExceptions() as $e) { // $e->getBatch() 本次失败的批次 // $e->getTransferredItems() 失败前已成功传输的项 // 可在此按需重试 $e->getBatch() } $batch->clearHistory(); $batch->clearExceptions();

注意：NotifyingBatch包装在内层、HistoryBatch包装在最外层（见 BatchBuilder.php 的组装顺序）。对上述组合而言，getHistory()、getExceptions()、clearHistory()、clearExceptions()均能通过装饰器的__call()透明代理到底层对象上，因此可以直接调用。

八、常见问题与排查要点

结语

Guzzle 3 的批处理框架（Guzzle\Batch）是一套结构精巧、扩展性极强的并行传输方案：Batch负责队列与冲刷调度，BatchDivisorInterface与BatchTransferInterface分别负责"分"与"传"，而FlushingBatch、NotifyingBatch、HistoryBatch、ExceptionBufferingBatch四个装饰器则为批处理叠加了自动冲刷、回调通知、历史记录与异常缓冲等实用行为。结合仓库内的 batching.rst 官方文档、src/Guzzle/Batch 的完整源码与 tests/Guzzle/Tests/Batch 的单元测试，你可以把这些机制彻底吃透，并迁移到任何需要"批量 + 并行 + 容错"的业务场景中。

• 后端
• 电商
• 前端