后端电商前端【免费下载链接】CRMEBCRMEB开源商城免费开源多语言商城系统Tp6框架商城系统可商用包含小程序商城、H5商城、公众号商城、PC商城、App支持分销、拼团、砍价、秒杀、优惠券、积分、会员等级、小程序直播、页面DIY前后端分离方便二开更有详细使用文档、接口文档、数据字典、二开文档/视频教程欢迎大家提出宝贵意见和建议项目地址https://gitcode.com/gh_mirrors/cr/CRMEB点击查看免费下载本文基于当前仓库内 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.phpprotected 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 的并行本质上是依托CurlMulticurl 多句柄一次性地把多个请求加入队列再由底层一次性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 三个核心角色原文档描述了一套清晰的三层协作模型角色接口/类职责队列容器Guzzle\Batch\Batch用add()把待发送项加入队列flush()时统一输送本质是一个SplQueue分组器DivisorGuzzle\Batch\BatchDivisorInterface把队列拆分成若干个可管理的小批次createBatches()传输器TransferGuzzle\Batch\BatchTransferInterface真正执行一个批次的传输transfer($batch)在 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如下先调用createBatches()把队列中的剩余项交给 divisor 划分成若干批次数组依次遍历每个已划分批次调用$this-transferStrategy-transfer($batch)完成传输把成功传输的项合并进返回值若某个批次抛出异常则包装为Guzzle\Batch\Exception\BatchTransferException抛出其中携带失败的批次getBatch()、此前已成功传输的项getTransferredItems()以及传输/分组策略对象见 BatchTransferException.php。需要注意SplQueue被设置为IT_MODE_DELETE迭代模式即createBatches()遍历队列的过程会消耗队列中的元素所以一次flush()之后已传输的项不再残留于队列。2.2 最小可运行的 Batch 示例原文档给出的最小示例是BatchBatchRequestTransfer的组合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校验队列中的每一项都必须是RequestInterface否则抛出InvalidArgumentException用SplObjectStorage按$item-getClient()把请求分组同一 client 共享同一 curl_multi 句柄对每组再按$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 分组、batchSize2 切块以及跨 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.phpcreateBatchesWith(BatchDivisorInterface $divisorStrategy)指定自定义 divisortransferWith(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.phpflush()先调用内部 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.phpadd()时先把项写入内部$history数组再转发给被装饰 batch。传输完成后可用getHistory()取回全部已添加并传输的项数组。原文档特别提醒务必定期用clearHistory()清理历史否则内存占用会随传输量线性增长。测试见 HistoryBatchTest.php。5.4ExceptionBufferingBatch异常缓冲失败后继续使用ExceptionBufferingBatch装饰器可以在传输过程中缓冲异常让你尽可能多地传输项然后再统一处理出错批次。传输结束后用getExceptions()取回一组Guzzle\Batch\Exception\BatchTransferException对象并可用它们重试失败的批次处理完毕务必用clearExceptions()清空缓冲。对应 builder 方法bufferExceptions()。实现细节见 ExceptionBufferingBatch.phpflush()进入while (!$this-decoratedBatch-isEmpty())循环反复冲刷内部 batch捕获BatchTransferException时把异常存入$exceptions并把$e-getTransferredItems()失败前已成功传输的项合并进返回值——因此即使某些批次失败其他批次仍会继续传输。测试 ExceptionBufferingBatchTest.php 给出了一个经典场景三个项foo/baz/barBatchSizeDivisor(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()透明代理到底层对象上因此可以直接调用。八、常见问题与排查要点现象原因与处置build()抛RuntimeException: No transfer strategy...未调用transferRequests()/transferCommands()/transferWith()之一transfer 与 divisor 均为空见 BatchBuilder.phpInvalidArgumentException: All items must implement...向BatchRequestTransfer/BatchCommandTransfer的队列中混入了非RequestInterface/CommandInterface对象见 BatchRequestTransfer.php、BatchCommandTransfer.phpInconsistentClientTransferException同一批次内混用了不同 client 的命令正常经BatchCommandTransfer分组后不会发生多出现在自定义 divisor 与transferCommands混用时见 BatchCommandTransfer.phpNotifyingBatch构造失败传入的notify()参数不是 callable见 NotifyingBatch.php内存持续增长keepHistory()开启后未调用clearHistory()清理历史或bufferExceptions()未调用clearExceptions()清空异常缓冲队列冲刷不彻底FlushingBatch只在add()计数达到阈值时自动冲刷不满阈值时队列会保留到最后一次手动flush()收尾需自行调用结语Guzzle 3 的批处理框架Guzzle\Batch是一套结构精巧、扩展性极强的并行传输方案Batch负责队列与冲刷调度BatchDivisorInterface与BatchTransferInterface分别负责分与传而FlushingBatch、NotifyingBatch、HistoryBatch、ExceptionBufferingBatch四个装饰器则为批处理叠加了自动冲刷、回调通知、历史记录与异常缓冲等实用行为。结合仓库内的 batching.rst 官方文档、src/Guzzle/Batch 的完整源码与 tests/Guzzle/Tests/Batch 的单元测试你可以把这些机制彻底吃透并迁移到任何需要批量 并行 容错的业务场景中。赞分享后端电商前端【免费下载链接】CRMEBCRMEB开源商城免费开源多语言商城系统Tp6框架商城系统可商用包含小程序商城、H5商城、公众号商城、PC商城、App支持分销、拼团、砍价、秒杀、优惠券、积分、会员等级、小程序直播、页面DIY前后端分离方便二开更有详细使用文档、接口文档、数据字典、二开文档/视频教程欢迎大家提出宝贵意见和建议项目地址https://gitcode.com/gh_mirrors/cr/CRMEB点击查看免费下载相关推荐Guzzle Pool批量请求1000并发任务高效处理方案Guzzle Pool批量请求1000并发任务高效处理方案 你是否还在为PHP项目中大量API请求导致的性能瓶颈发愁当需要处理成百上千个并发HTTP请求时后端tRPC 客户端 Links 与请求批处理Request Batching完全指南tRPC 客户端 Links 与请求批处理Request Batching完全指南 本指南以 tRPC v9 官方文档 Links Request Ba后端RPC框架前端Hermes WebUI 多模型支持从配置第一个提供商到会话里换模型Hermes WebUI 多模型支持从配置第一个提供商到会话里换模型 Hermes WebUI 的多模型支持让你把 OpenAI、Anthropic、Oll人工智能AI 应用AI Agent交互助手MCP 服务前端创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考