Hyperf 通过 elasticsearch-php v8.x 客户端集成 Elasticsearch，利用 bulk API 实现协程安全的批量更新、删除等操作；需合理分批（如每500条）、检查 errors 字段处理失败、结合生成器优化内存。
Hyperf 框架本身不内置 Elasticsearch 客户端，但可通过官方推荐的 elasticsearch-php SDK（v8.x）与 Hyperf 的协程兼容性良好地集成，实现高效、批量的数据修改（如批量更新、删除、upsert）。关键在于：使用 bulk API、确保协程安全、合理控制批次大小与错误处理。
1. 安装并配置 Elasticsearch 客户端
推荐使用官方维护的 elasticsearch/elasticsearch v8.x（支持 PHP 8.1+，原生协程友好）：
- 执行：
composer require elasticsearch/elasticsearch:^8.0 - 在
config/autoload/services.php中注册客户端实例（利用 Hyperf DI 容器管理单例）：
'elasticsearch' => [
'class' => \Elasticsearch\ClientBuilder::class,
'shared' => true,
'properties' => [
'hosts' => ['http://localhost:9200'],
'retryOnConflict' => 3,
// 可选：启用 curl 连接池（需配合 hyperf/curl）
'handler' => \Elasticsearch\Transport\Handlers\CurlHandler::class,
],
],
然后通过 Container::get(ClientBuilder::class)->build() 获取客户端；建议封装为 ElasticsearchService 类统一调用。
2. 批量修改的核心：使用 bulk API 构造操作数组
Elasticsearch 的 _bulk 接口支持混合操作（index/update/delete），每条操作需配对元数据行 + 数据行。Hyperf 中应避免同步阻塞调用，改用协程安全的异步请求（client->bulk() 默认是同步，但可在协程内安全调用）：
- 构造符合格式的
body数组（注意：必须是严格顺序的“动作行 + 文档行”交替） - 示例：批量更新商品价格（使用
update操作 +script）
$body = [];
foreach ($products as $product) {
$body[] = [
'update' => [
'_index' => 'products',
'_id' => $product['id'],
],
];
$body[] = [
'script' => [
'source' => 'ctx._source.price = params.price',
'params' => ['price' => $product['new_price']],
],
];
}
$response = $client->bulk(['body' => $body]);
⚠️ 注意：bulk 不保证原子性，需检查 $response['errors'] === true 并解析 $response['items'] 中各操作的 status 和 error 字段做失败重试或日志记录。
3. 控制批次大小与内存优化
单次 bulk 请求不宜过大（官方建议 5–15 MB，约 1000–5000 条），否则易触发超时或 OOM：
- 按固定数量切分（如每 500 条提交一次）
- Hyperf 中可结合
Swoole\Coroutine\WaitGroup实现并发多批次提交（慎用，需评估集群负载） - 避免在循环中拼接大数组；推荐使用生成器（
yield）流式构建 body，降低内存占用
4. 错误处理与幂等保障
批量操作失败常见于文档不存在、mapping 冲突、脚本语法错误等。建议：
- 对
bulk响应逐项检查，提取失败 ID 与原因，记录到日志或落库待人工干预 - 业务上为更新操作添加版本号（
_version）或时间戳字段，防止脏写 - 若需强一致性，可先查后更（不推荐高频场景），或结合 Redis 分布式锁控制同一文档的并发修改
不复杂但容易忽略。
