PHP小码农

Appearance

PHP 客户端操作 ​

elasticsearch-php 是 Elasticsearch 官方提供的 PHP 客户端库，封装了所有 RESTful API 调用，支持连接池、重试机制、异步操作等高级特性。本节将深入讲解如何在 PHP 项目中高效地与 Elasticsearch 交互，涵盖 CRUD 操作、批量处理、复杂查询 DSL 的构建以及生产级封装实践。

前置知识

阅读本节前，建议先了解：Elasticsearch 基础 和 Composer 包管理

安装与配置 ​

安装 elasticsearch-php ​

bash
# 安装官方 PHP 客户端（需要 PHP 8.0+）
composer require elasticsearch/elasticsearch

# 安装 PSR-18 HTTP 客户端（elasticsearch-php 需要一个 HTTPlug 实现）
# Guzzle 7 是最常用的选择
composer require guzzlehttp/guzzle

# 如果项目需要日志记录
composer require monolog/monolog

版本兼容性

elasticsearch-php 的主版本号需要与 Elasticsearch 服务器版本对应：

elasticsearch-phpElasticsearch 服务器PHP 版本
8.x8.x>= 8.0
7.x7.x>= 7.1
6.x6.x>= 7.0

基础连接配置 ​

php
<?php

declare(strict_types=1);

namespace App\Service;

use Elasticsearch\Client;
use Elasticsearch\ClientBuilder;
use Monolog\Logger;
use Monolog\Handler\StreamHandler;

class ElasticsearchService
{
    private Client $client;

    public function __construct()
    {
        $this->client = ClientBuilder::create()
            ->setHosts($this->getHosts())
            ->setRetries(3)
            ->setLogger($this->createLogger())
            ->build();
    }

    /**
     * 获取 ES 集群节点列表
     */
    private function getHosts(): array
    {
        return [
            [
                'host'   => '192.168.1.101',
                'port'   => 9200,
                'scheme' => 'http',
                'user'   => 'elastic',
                'pass'   => 'your_password',
            ],
            [
                'host'   => '192.168.1.102',
                'port'   => 9200,
                'scheme' => 'http',
                'user'   => 'elastic',
                'pass'   => 'your_password',
            ],
            [
                'host'   => '192.168.1.103',
                'port'   => 9200,
                'scheme' => 'http',
                'user'   => 'elastic',
                'pass'   => 'your_password',
            ],
        ];
    }

    /**
     * 创建日志记录器
     */
    private function createLogger(): Logger
    {
        $logger = new Logger('elasticsearch');
        $logger->pushHandler(
            new StreamHandler('/var/log/elasticsearch.log', Logger::WARNING)
        );
        return $logger;
    }

    public function getClient(): Client
    {
        return $this->client;
    }
}

连接池配置 ​

php
<?php

declare(strict_types=1);

namespace App\Service;

use Elasticsearch\ClientBuilder;
use Elasticsearch\Serializers\SmartSerializer;
use GuzzleHttp\Ring\Client\CurlHandler;
use GuzzleHttp\Ring\Client\MockHandler;

// 自定义连接池配置
class ElasticsearchPoolConfig
{
    public static function createClient(): \Elasticsearch\Client
    {
        $builder = ClientBuilder::create();

        // 设置主机列表
        $builder->setHosts([
            'http://elastic:password@es-node1:9200',
            'http://elastic:password@es-node2:9200',
            'http://elastic:password@es-node3:9200',
        ]);

        // 连接池选择器（默认是 SimpleSelector，轮询方式）
        // - SimpleSelector：轮询（Round-robin）
        // - StaticNoPingSelector：不检测健康状态，性能更好
        // - StickySelector：粘性选择，同一个请求发到同一个节点
        $builder->setSelector('\Elasticsearch\ConnectionPool\StaticNoPingSelector');

        // 重试策略
        $builder->setRetries(3);           // 请求失败重试次数
        $builder->setConnectionParams([
            'timeout'       => 5,           // 连接超时（秒）
            'read_timeout'  => 30,          // 读取超时（秒）
        ]);

        // 使用自定义序列化器
        $builder->setSerializer(new SmartSerializer());

        // 设置追踪日志（调试用，生产环境关闭）
        // $builder->setTracer(...);

        return $builder->build();
    }
}

文档 CRUD 操作 ​

索引文档（创建/更新） ​

php
<?php

declare(strict_types=1);

namespace App\Repository;

use Elasticsearch\Client;

class ProductRepository
{
    public function __construct(
        private readonly Client $esClient
    ) {}

    /**
     * 创建文档（指定文档 ID）
     */
    public function create(string $index, string $id, array $document): array
    {
        $params = [
            'index' => $index,
            'id'    => $id,
            'body'  => $document,
        ];

        $response = $this->esClient->index($params);

        return [
            'result'  => $response['result'],       // 'created' 或 'updated'
            '_id'     => $response['_id'],
            '_index'  => $response['_index'],
            '_version'=> $response['_version'],
        ];
    }

    /**
     * 创建文档（自动生成 ID）
     */
    public function createAutoId(string $index, array $document): array
    {
        $params = [
            'index' => $index,
            'body'  => $document,
        ];

        $response = $this->esClient->index($params);

        return [
            'result'   => $response['result'],
            '_id'      => $response['_id'],
            '_index'   => $response['_index'],
            '_version' => $response['_version'],
        ];
    }

    /**
     * 添加商品到索引
     */
    public function indexProduct(array $productData): array
    {
        return $this->create('products', $productData['id'], [
            'product_id'    => $productData['id'],
            'title'         => $productData['title'],
            'description'   => $productData['description'],
            'price'         => $productData['price'],
            'original_price'=> $productData['original_price'] ?? 0,
            'category'      => $productData['category'],
            'brand'         => $productData['brand'] ?? '',
            'tags'          => $productData['tags'] ?? [],
            'sales_count'   => $productData['sales_count'] ?? 0,
            'rating'        => $productData['rating'] ?? 0,
            'is_on_sale'    => $productData['is_on_sale'] ?? true,
            'created_at'    => date('Y-m-d H:i:s'),
            'updated_at'    => date('Y-m-d H:i:s'),
        ]);
    }
}

读取文档 ​

php
<?php

declare(strict_types=1);

namespace App\Repository;

use Elasticsearch\Client;
use Elasticsearch\Common\Exceptions\Missing404Exception;

class ProductRepository
{
    public function __construct(
        private readonly Client $esClient
    ) {}

    /**
     * 通过 ID 获取文档
     */
    public function getById(string $index, string $id): ?array
    {
        try {
            $params = [
                'index' => $index,
                'id'    => $id,
            ];

            $response = $this->esClient->get($params);

            return $response['_source'];
        } catch (Missing404Exception $e) {
            return null;
        }
    }

    /**
     * 批量获取文档（MGET）
     */
    public function getByIds(string $index, array $ids): array
    {
        $params = [
            'index' => $index,
            'body'  => ['ids' => $ids],
        ];

        $response = $this->esClient->mget($params);

        $results = [];
        foreach ($response['docs'] as $doc) {
            if ($doc['found'] ?? false) {
                $results[$doc['_id']] = $doc['_source'];
            }
        }

        return $results;
    }

    /**
     * 判断文档是否存在
     */
    public function exists(string $index, string $id): bool
    {
        $params = [
            'index' => $index,
            'id'    => $id,
        ];

        return $this->esClient->exists($params);
    }

    /**
     * 获取文档并包含元数据
     */
    public function getWithMeta(string $index, string $id): ?array
    {
        try {
            $params = [
                'index' => $index,
                'id'    => $id,
            ];

            $response = $this->esClient->get($params);

            return [
                '_id'       => $response['_id'],
                '_index'    => $response['_index'],
                '_version'  => $response['_version'],
                '_seq_no'   => $response['_seq_no'],
                '_primary_term' => $response['_primary_term'],
                'source'    => $response['_source'],
            ];
        } catch (Missing404Exception $e) {
            return null;
        }
    }
}

更新文档 ​

php
<?php

declare(strict_types=1);

namespace App\Repository;

use Elasticsearch\Client;
use Elasticsearch\Common\Exceptions\Missing404Exception;

class ProductRepository
{
    public function __construct(
        private readonly Client $esClient
    ) {}

    /**
     * 部分更新文档
     */
    public function update(string $index, string $id, array $partialDoc): array
    {
        $params = [
            'index' => $index,
            'id'    => $id,
            'body'  => [
                'doc' => array_merge($partialDoc, [
                    'updated_at' => date('Y-m-d H:i:s'),
                ]),
            ],
        ];

        try {
            return $this->esClient->update($params);
        } catch (Missing404Exception $e) {
            return ['error' => 'document_not_found', 'id' => $id];
        }
    }

    /**
     * 使用脚本更新文档（原子操作）
     */
    public function incrementSalesCount(string $index, string $id, int $count = 1): array
    {
        $params = [
            'index' => $index,
            'id'    => $id,
            'body'  => [
                'script' => [
                    'source' => 'ctx._source.sales_count += params.count; ctx._source.updated_at = params.now',
                    'params' => [
                        'count' => $count,
                        'now'   => date('Y-m-d H:i:s'),
                    ],
                ],
            ],
        ];

        try {
            return $this->esClient->update($params);
        } catch (Missing404Exception $e) {
            return ['error' => 'document_not_found', 'id' => $id];
        }
    }

    /**
     * Upsert 操作（文档存在则更新，不存在则创建）
     */
    public function upsert(string $index, string $id, array $document, array $updateFields = []): array
    {
        $params = [
            'index' => $index,
            'id'    => $id,
            'body'  => [
                'doc_as_upsert' => true,
                'doc'           => array_merge($updateFields, [
                    'updated_at' => date('Y-m-d H:i:s'),
                ]),
            ],
        ];

        return $this->esClient->update($params);
    }

    /**
     * 条件更新（Update by Query）
     */
    public function updateByCondition(string $index, array $query, array $script): array
    {
        $params = [
            'index' => $index,
            'body'  => [
                'query'  => $query,
                'script' => $script,
            ],
        ];

        return $this->esClient->updateByQuery($params);
    }
}

删除文档 ​

php
<?php

declare(strict_types=1);

namespace App\Repository;

use Elasticsearch\Client;
use Elasticsearch\Common\Exceptions\Missing404Exception;

class ProductRepository
{
    public function __construct(
        private readonly Client $esClient
    ) {}

    /**
     * 通过 ID 删除文档
     */
    public function deleteById(string $index, string $id): bool
    {
        try {
            $params = [
                'index' => $index,
                'id'    => $id,
            ];

            $response = $this->esClient->delete($params);
            return $response['result'] === 'deleted';
        } catch (Missing404Exception $e) {
            return false;
        }
    }

    /**
     * 条件删除（Delete by Query）
     */
    public function deleteByQuery(string $index, array $query): array
    {
        $params = [
            'index' => $index,
            'body'  => [
                'query' => $query,
            ],
            'client' => [
                'timeout' => '30s',
            ],
        ];

        return $this->esClient->deleteByQuery($params);
    }

    /**
     * 删除指定索引中所有文档
     */
    public function deleteAll(string $index): array
    {
        return $this->deleteByQuery($index, [
            'match_all' => (object)[],
        ]);
    }
}

批量操作（Bulk） ​

批量操作是 Elasticsearch 高性能写入的关键。通过减少网络往返次数，bulk API 可以显著提升数据导入效率。

基本批量操作 ​

php
<?php

declare(strict_types=1);

namespace App\Repository;

use Elasticsearch\Client;

class BulkRepository
{
    public function __construct(
        private readonly Client $esClient
    ) {}

    /**
     * 批量索引文档
     */
    public function bulkIndex(string $index, array $documents): array
    {
        $params = [
            'index' => $index,
            'body'  => [],
        ];

        foreach ($documents as $doc) {
            $id = $doc['id'] ?? null;
            unset($doc['id']);

            if ($id !== null) {
                $params['body'][] = ['index' => ['_id' => $id]];
            } else {
                $params['body'][] = ['index' => []];
            }
            $params['body'][] = $doc;
        }

        return $this->esClient->bulk($params);
    }

    /**
     * 批量删除文档
     */
    public function bulkDelete(string $index, array $ids): array
    {
        $params = [
            'index' => $index,
            'body'  => [],
        ];

        foreach ($ids as $id) {
            $params['body'][] = ['delete' => ['_id' => $id]];
        }

        return $this->esClient->bulk($params);
    }

    /**
     * 混合批量操作（同时包含 index、update、delete）
     */
    public function bulkMixed(string $index, array $actions): array
    {
        $params = [
            'index' => $index,
            'body'  => [],
            'refresh' => false,   // 批量写入时不要实时刷新
        ];

        foreach ($actions as $action) {
            match ($action['type']) {
                'index' => [
                    $params['body'][] = ['index' => ['_id' => $action['id']]],
                    $params['body'][] = $action['body'],
                ],
                'update' => [
                    $params['body'][] = ['update' => ['_id' => $action['id']]],
                    $params['body'][] = ['doc' => $action['body']],
                ],
                'delete' => [
                    $params['body'][] = ['delete' => ['_id' => $action['id']]],
                ],
                default => null,
            };
        }

        return $this->esClient->bulk($params);
    }
}

大数据量批量导入 ​

php
<?php

declare(strict_types=1);

namespace App\Service;

use Elasticsearch\Client;

class DataImportService
{
    public function __construct(
        private readonly Client $esClient
    ) {}

    /**
     * 大数据量批量导入（内存优化版）
     *
     * @param string $index 索引名称
     * @param iterable $dataGenerator 数据生成器（避免一次性加载全部数据到内存）
     * @param int $batchSize 每批文档数
     * @param callable|null $onProgress 进度回调
     */
    public function bulkImport(
        string $index,
        iterable $dataGenerator,
        int $batchSize = 2000,
        ?callable $onProgress = null
    ): array {
        $stats = [
            'total'      => 0,
            'indexed'    => 0,
            'errors'     => 0,
            'batches'    => 0,
            'error_items' => [],
        ];

        $batch = [];
        $count = 0;

        foreach ($dataGenerator as $doc) {
            $id = $doc['id'] ?? null;
            unset($doc['id']);

            if ($id !== null) {
                $batch[] = ['index' => ['_index' => $index, '_id' => $id]];
            } else {
                $batch[] = ['index' => ['_index' => $index]];
            }
            $batch[] = $doc;
            $count++;
            $stats['total']++;

            // 达到批次大小时执行批量写入
            if ($count >= $batchSize) {
                $result = $this->executeBulk($batch);
                $stats['indexed'] += $result['success_count'];
                $stats['errors'] += $result['error_count'];
                $stats['error_items'] = array_merge(
                    $stats['error_items'],
                    $result['error_items']
                );
                $stats['batches']++;
                $batch = [];
                $count = 0;

                if ($onProgress !== null) {
                    $onProgress($stats);
                }
            }
        }

        // 处理最后一批
        if (!empty($batch)) {
            $result = $this->executeBulk($batch);
            $stats['indexed'] += $result['success_count'];
            $stats['errors'] += $result['error_count'];
            $stats['error_items'] = array_merge(
                $stats['error_items'],
                $result['error_items']
            );
            $stats['batches']++;
        }

        return $stats;
    }

    /**
     * 执行批量请求
     */
    private function executeBulk(array $body): array
    {
        try {
            $response = $this->esClient->bulk([
                'body'    => $body,
                'timeout' => '60s',
            ]);

            $successCount = 0;
            $errorCount = 0;
            $errorItems = [];

            foreach ($response['items'] as $item) {
                $action = array_key_first($item);
                $status = $item[$action]['status'] ?? 0;

                if ($status >= 200 && $status < 300) {
                    $successCount++;
                } else {
                    $errorCount++;
                    $errorItems[] = [
                        'action' => $action,
                        'id'     => $item[$action]['_id'] ?? null,
                        'status' => $status,
                        'error'  => $item[$action]['error'] ?? null,
                    ];
                }
            }

            return [
                'success_count' => $successCount,
                'error_count'   => $errorCount,
                'error_items'   => $errorItems,
            ];
        } catch (\Throwable $e) {
            return [
                'success_count' => 0,
                'error_count'   => count($body) / 2,
                'error_items'   => [['error' => $e->getMessage()]],
            ];
        }
    }

    /**
     * 从数据库批量同步到 ES（使用生成器节省内存）
     */
    public function syncFromDatabase(
        \PDO $pdo,
        string $index,
        string $sql,
        int $batchSize = 2000
    ): array {
        $stmt = $pdo->query($sql);
        $stmt->setFetchMode(\PDO::FETCH_ASSOC);

        $generator = (function () use ($stmt) {
            while ($row = $stmt->fetch()) {
                yield $row;
            }
        })();

        return $this->bulkImport($index, $generator, $batchSize);
    }
}

搜索 DSL（Domain Specific Language） ​

全文搜索查询 ​

php
<?php

declare(strict_types=1);

namespace App\Repository;

use Elasticsearch\Client;

class ProductSearchRepository
{
    public function __construct(
        private readonly Client $esClient
    ) {}

    /**
     * 全文搜索（match query）
     */
    public function search(string $index, string $query, int $page = 1, int $size = 20): array
    {
        $params = [
            'index' => $index,
            'body'  => [
                'query' => [
                    'match' => [
                        'title' => [
                            'query'    => $query,
                            'operator' => 'and',
                            'minimum_should_match' => '75%',
                        ],
                    ],
                ],
                'from' => ($page - 1) * $size,
                'size' => $size,
            ],
        ];

        $response = $this->esClient->search($params);

        return $this->parseSearchResponse($response);
    }

    /**
     * 多字段搜索（multi_match）
     */
    public function multiFieldSearch(string $index, string $keyword, int $page = 1, int $size = 20): array
    {
        $params = [
            'index' => $index,
            'body'  => [
                'query' => [
                    'multi_match' => [
                        'query'  => $keyword,
                        'fields'  => ['title^3', 'description^1', 'brand^2', 'tags^2'],
                        'type'    => 'best_fields',
                        'fuzziness' => 'AUTO',
                    ],
                ],
                'from' => ($page - 1) * $size,
                'size' => $size,
            ],
        ];

        $response = $this->esClient->search($params);

        return $this->parseSearchResponse($response);
    }

    /**
     * 短语搜索（match_phrase）
     */
    public function phraseSearch(string $index, string $phrase, int $page = 1, int $size = 20): array
    {
        $params = [
            'index' => $index,
            'body'  => [
                'query' => [
                    'match_phrase' => [
                        'description' => [
                            'query' => $phrase,
                            'slop'  => 2,  // 允许词间隔
                        ],
                    ],
                ],
                'from' => ($page - 1) * $size,
                'size' => $size,
            ],
        ];

        $response = $this->esClient->search($params);

        return $this->parseSearchResponse($response);
    }

    /**
     * 前缀搜索
     */
    public function prefixSearch(string $index, string $prefix): array
    {
        $params = [
            'index' => $index,
            'body'  => [
                'query' => [
                    'prefix' => [
                        'title.keyword' => [
                            'value' => $prefix,
                        ],
                    ],
                ],
                'size' => 10,
            ],
        ];

        $response = $this->esClient->search($params);

        return $this->parseSearchResponse($response);
    }
}

精确匹配与过滤 ​

php
<?php

declare(strict_types=1);

namespace App\Repository;

use Elasticsearch\Client;

class ProductFilterRepository
{
    public function __construct(
        private readonly Client $esClient
    ) {}

    /**
     * term 精确匹配
     */
    public function filterByCategory(string $index, string $category): array
    {
        $params = [
            'index' => $index,
            'body'  => [
                'query' => [
                    'term' => [
                        'category' => $category,
                    ],
                ],
            ],
        ];

        $response = $this->esClient->search($params);
        return $this->parseSearchResponse($response);
    }

    /**
     * terms 多值匹配（类似 SQL IN）
     */
    public function filterByCategories(string $index, array $categories): array
    {
        $params = [
            'index' => $index,
            'body'  => [
                'query' => [
                    'terms' => [
                        'category' => $categories,
                    ],
                ],
            ],
        ];

        $response = $this->esClient->search($params);
        return $this->parseSearchResponse($response);
    }

    /**
     * range 范围查询
     */
    public function filterByPriceRange(
        string $index,
        ?float $min = null,
        ?float $max = null
    ): array {
        $range = [];
        if ($min !== null) {
            $range['gte'] = $min;
        }
        if ($max !== null) {
            $range['lte'] = $max;
        }

        $params = [
            'index' => $index,
            'body'  => [
                'query' => [
                    'range' => [
                        'price' => $range,
                    ],
                ],
            ],
        ];

        $response = $this->esClient->search($params);
        return $this->parseSearchResponse($response);
    }

    /**
     * exists 查询（字段不为空）
     */
    public function filterByExists(string $index, string $field): array
    {
        $params = [
            'index' => $index,
            'body'  => [
                'query' => [
                    'exists' => [
                        'field' => $field,
                    ],
                ],
            ],
        ];

        $response = $this->esClient->search($params);
        return $this->parseSearchResponse($response);
    }
}

复合查询（Bool Query） ​

php
<?php

declare(strict_types=1);

namespace App\Repository;

use Elasticsearch\Client;

class ProductAdvancedRepository
{
    public function __construct(
        private readonly Client $esClient
    ) {}

    /**
     * 综合搜索（组合多种查询条件）
     */
    public function advancedSearch(array $criteria): array
    {
        $query = ['bool' => []];

        // must：必须匹配（影响评分）
        if (!empty($criteria['keyword'])) {
            $query['bool']['must'][] = [
                'multi_match' => [
                    'query'  => $criteria['keyword'],
                    'fields' => ['title^3', 'description^1', 'tags^2'],
                    'type'   => 'cross_fields',
                ],
            ];
        }

        // filter：必须匹配（不影响评分，可缓存）
        if (!empty($criteria['category'])) {
            $query['bool']['filter'][] = [
                'term' => ['category' => $criteria['category']],
            ];
        }

        if (isset($criteria['min_price']) || isset($criteria['max_price'])) {
            $range = [];
            if (isset($criteria['min_price'])) {
                $range['gte'] = (float) $criteria['min_price'];
            }
            if (isset($criteria['max_price'])) {
                $range['lte'] = (float) $criteria['max_price'];
            }
            $query['bool']['filter'][] = ['range' => ['price' => $range]];
        }

        if (isset($criteria['is_on_sale'])) {
            $query['bool']['filter'][] = [
                'term' => ['is_on_sale' => (bool) $criteria['is_on_sale']],
            ];
        }

        // should：可选匹配（增加评分权重）
        if (!empty($criteria['brand'])) {
            $query['bool']['should'][] = [
                'term' => ['brand' => $criteria['brand']],
            ];
        }

        if (!empty($criteria['tags'])) {
            $query['bool']['should'][] = [
                'terms' => ['tags' => (array) $criteria['tags']],
            ];
        }

        // must_not：必须不匹配
        if (!empty($criteria['exclude_brands'])) {
            foreach ((array) $criteria['exclude_brands'] as $brand) {
                $query['bool']['must_not'][] = [
                    'term' => ['brand' => $brand],
                ];
            }
        }

        $page = (int) ($criteria['page'] ?? 1);
        $size = (int) ($criteria['size'] ?? 20);

        // 构建排序
        $sort = $this->buildSort($criteria);

        $params = [
            'index' => 'products',
            'body'  => [
                'query' => $query,
                'from'  => ($page - 1) * $size,
                'size'  => $size,
                'sort'  => $sort,
                'track_total_hits' => true,
            ],
        ];

        $response = $this->esClient->search($params);

        return $this->parseSearchResponse($response);
    }

    /**
     * 构建排序条件
     */
    private function buildSort(array $criteria): array
    {
        $sort = [];

        $sortField = $criteria['sort'] ?? '_score';
        $sortOrder = $criteria['order'] ?? 'desc';

        return match ($sortField) {
            'price'       => [['price' => ['order' => $sortOrder]]],
            'sales_count' => [['sales_count' => ['order' => $sortOrder]]],
            'rating'      => [['rating' => ['order' => $sortOrder]]],
            'created_at'  => [['created_at' => ['order' => $sortOrder]]],
            'distance'    => [
                '_geo_distance' => [
                    'location'   => $criteria['location'] ?? [0, 0],
                    'order'      => $sortOrder,
                    'unit'       => 'km',
                    'distance_type' => 'arc',
                ],
            ],
            default        => [['_score' => ['order' => 'desc']]],
        };
    }

    /**
     * 解析搜索响应
     */
    private function parseSearchResponse(array $response): array
    {
        $hits = $response['hits'] ?? [];

        return [
            'total'      => $hits['total']['value'] ?? 0,
            'max_score'  => $hits['max_score'] ?? 0,
            'items'      => array_map(function (array $hit): array {
                return [
                    'id'        => $hit['_id'],
                    'score'     => $hit['_score'] ?? 0,
                    'source'    => $hit['_source'],
                    'highlight' => $hit['highlight'] ?? [],
                ];
            }, $hits['hits'] ?? []),
        ];
    }
}

高亮搜索 ​

php
<?php

declare(strict_types=1);

namespace App\Repository;

use Elasticsearch\Client;

class ProductHighlightRepository
{
    public function __construct(
        private readonly Client $esClient
    ) {}

    /**
     * 带高亮的搜索结果
     */
    public function searchWithHighlight(
        string $keyword,
        int $page = 1,
        int $size = 20
    ): array {
        $params = [
            'index' => 'products',
            'body'  => [
                'query' => [
                    'bool' => [
                        'should' => [
                            [
                                'match' => [
                                    'title' => [
                                        'query' => $keyword,
                                        'boost' => 3,
                                    ],
                                ],
                            ],
                            [
                                'match' => [
                                    'description' => [
                                        'query' => $keyword,
                                        'boost' => 1,
                                    ],
                                ],
                            ],
                        ],
                        'minimum_should_match' => 1,
                    ],
                ],
                'highlight' => [
                    'pre_tags'  => ['<em class="highlight">'],
                    'post_tags' => ['</em>'],
                    'fields'    => [
                        'title' => [
                            'fragment_size'       => 100,
                            'number_of_fragments'  => 3,
                            'fragment_offset'     => 0,
                        ],
                        'description' => [
                            'fragment_size'       => 200,
                            'number_of_fragments'  => 2,
                            'fragment_offset'     => 0,
                        ],
                    ],
                ],
                'from' => ($page - 1) * $size,
                'size' => $size,
            ],
        ];

        $response = $this->esClient->search($params);
        $hits = $response['hits']['hits'] ?? [];

        $results = [];
        foreach ($hits as $hit) {
            $results[] = [
                'id'         => $hit['_id'],
                'score'      => $hit['_score'],
                'source'     => $hit['_source'],
                'highlights' => $this->extractHighlights($hit),
            ];
        }

        return [
            'total' => $response['hits']['total']['value'] ?? 0,
            'items' => $results,
        ];
    }

    /**
     * 提取高亮内容
     */
    private function extractHighlights(array $hit): array
    {
        $highlights = [];
        $highlightFields = $hit['highlight'] ?? [];

        foreach ($highlightFields as $field => $fragments) {
            $highlights[$field] = $fragments;
        }

        return $highlights;
    }
}

索引管理 ​

php
<?php

declare(strict_types=1);

namespace App\Service;

use Elasticsearch\Client;
use Elasticsearch\Common\Exceptions\Missing404Exception;

class IndexManagementService
{
    public function __construct(
        private readonly Client $esClient
    ) {}

    /**
     * 创建索引
     */
    public function createIndex(string $indexName, array $settings = [], array $mappings = []): array
    {
        $params = [
            'index' => $indexName,
            'body'  => [],
        ];

        if (!empty($settings)) {
            $params['body']['settings'] = $settings;
        }

        if (!empty($mappings)) {
            $params['body']['mappings'] = $mappings;
        }

        return $this->esClient->indices()->create($params);
    }

    /**
     * 检查索引是否存在
     */
    public function indexExists(string $indexName): bool
    {
        return $this->esClient->indices()->exists(['index' => $indexName]);
    }

    /**
     * 删除索引
     */
    public function deleteIndex(string $indexName): bool
    {
        try {
            $this->esClient->indices()->delete(['index' => $indexName]);
            return true;
        } catch (Missing404Exception $e) {
            return false;
        }
    }

    /**
     * 获取索引映射
     */
    public function getMapping(string $indexName): array
    {
        return $this->esClient->indices()->getMapping(['index' => $indexName]);
    }

    /**
     * 更新索引设置
     */
    public function updateSettings(string $indexName, array $settings): array
    {
        return $this->esClient->indices()->putSettings([
            'index' => $indexName,
            'body'  => ['settings' => $settings],
        ]);
    }

    /**
     * 添加新字段映射
     */
    public function addMapping(string $indexName, array $properties): array
    {
        return $this->esClient->indices()->putMapping([
            'index' => $indexName,
            'body'  => ['properties' => $properties],
        ]);
    }

    /**
     * 重建索引（Reindex）
     */
    public function reindex(
        string $sourceIndex,
        string $destIndex,
        ?array $query = null,
        int $batchSize = 1000
    ): array {
        $body = [
            'source' => [
                'index' => $sourceIndex,
                'size'  => $batchSize,
            ],
            'dest' => [
                'index' => $destIndex,
            ],
        ];

        if ($query !== null) {
            $body['source']['query'] = $query;
        }

        return $this->esClient->reindex(['body' => $body]);
    }

    /**
     * 获取索引统计信息
     */
    public function getIndexStats(string $indexName): array
    {
        return $this->esClient->indices()->stats(['index' => $indexName]);
    }
}

注意事项 ​

错误处理 ​

php
<?php

declare(strict_types=1);

namespace App\Exception;

use Elasticsearch\Common\Exceptions;
use Throwable;

class ElasticsearchErrorHandler
{
    /**
     * 处理 Elasticsearch 异常
     */
    public static function handle(Throwable $e): string
    {
        return match (true) {
            $e instanceof Exceptions\Missing404Exception
                => '文档或索引不存在: ' . $e->getMessage(),

            $e instanceof Exceptions\BadRequest400Exception
                => '请求参数错误: ' . $e->getMessage(),

            $e instanceof Exceptions\Conflict409Exception
                => '版本冲突: ' . $e->getMessage(),

            $e instanceof Exceptions\Forbidden403Exception
                => '权限不足: ' . $e->getMessage(),

            $e instanceof Exceptions\Unauthorized401Exception
                => '认证失败: ' . $e->getMessage(),

            $e instanceof Exceptions\TimeoutException
                => '请求超时: ' . $e->getMessage(),

            $e instanceof Exceptions\NoNodesAvailableException
                => '无可用 ES 节点，请检查集群状态',

            $e instanceof Exceptions\MaxRetriesException
                => '超过最大重试次数: ' . $e->getMessage(),

            default
                => 'Elasticsearch 错误 [' . get_class($e) . ']: ' . $e->getMessage(),
        };
    }

    /**
     * 是否为可重试错误
     */
    public static function isRetryable(Throwable $e): bool
    {
        return $e instanceof Exceptions\TimeoutException
            || $e instanceof Exceptions\NoNodesAvailableException
            || $e instanceof Exceptions\MaxRetriesException
            || $e instanceof Exceptions\Conflict409Exception;
    }
}

性能注意事项 ​

php
<?php

declare(strict_types=1);

namespace App\Repository;

use Elasticsearch\Client;

class ProductRepository
{
    // ========================================
    // 错误示范
    // ========================================

    /**
     * BAD: 循环逐条写入（大量网络开销）
     */
    public function badBatchInsert(Client $client, array $products): void
    {
        foreach ($products as $product) {
            $client->index([
                'index' => 'products',
                'id'    => $product['id'],
                'body'  => $product,
            ]);
        }
    }

    // ========================================
    // 正确示范
    // ========================================

    /**
     * GOOD: 使用 bulk API 批量写入
     */
    public function goodBatchInsert(Client $client, array $products): void
    {
        $body = [];
        foreach ($products as $product) {
            $body[] = ['index' => ['_index' => 'products', '_id' => $product['id']]];
            $body[] = $product;
        }

        $client->bulk(['body' => $body]);
    }

    /**
     * GOOD: 使用 _source 过滤减少返回数据量
     */
    public function goodSearch(Client $client, string $keyword): array
    {
        return $client->search([
            'index' => 'products',
            'body'  => [
                '_source' => ['product_id', 'title', 'price'],  // 只返回需要的字段
                'query'   => ['match' => ['title' => $keyword]],
                'size'    => 20,
            ],
        ]);
    }

    /**
     * GOOD: 使用 scroll API 深度分页
     */
    public function scrollSearch(Client $client, string $keyword, int $totalLimit = 10000): array
    {
        $params = [
            'index' => 'products',
            'body'  => [
                'query' => ['match' => ['title' => $keyword]],
                'size'  => 1000,        // 每批大小
                'scroll' => '5m',       // 游标保留时间
            ],
        ];

        $response = $client->search($params);
        $scrollId = $response['_scroll_id'];
        $allHits = [];

        while (true) {
            $hits = $response['hits']['hits'] ?? [];
            if (empty($hits)) {
                break;
            }

            foreach ($hits as $hit) {
                $allHits[] = $hit['_source'];
            }

            if (count($allHits) >= $totalLimit) {
                break;
            }

            $response = $client->scroll([
                'scroll_id' => $scrollId,
                'scroll'    => '5m',
            ]);
            $scrollId = $response['_scroll_id'];
        }

        // 清除 scroll 游标（重要！）
        try {
            $client->clearScroll(['scroll_id' => $scrollId]);
        } catch (\Throwable $e) {
            // 忽略清除错误
        }

        return $allHits;
    }

    /**
     * GOOD: 使用 search_after 替代深度分页
     */
    public function searchAfter(Client $client, string $keyword, array $sort = [], ?array $after = null): array
    {
        $body = [
            'query' => ['match' => ['title' => $keyword]],
            'size'  => 20,
            'sort'  => $sort ?: ['created_at' => 'desc', '_id' => 'asc'],
        ];

        if ($after !== null) {
            $body['search_after'] = $after;
        }

        $response = $client->search([
            'index' => 'products',
            'body'  => $body,
        ]);

        $hits = $response['hits']['hits'];
        $lastSort = [];

        if (!empty($hits)) {
            $lastSort = end($hits)['sort'];
        }

        return [
            'items'     => array_column($hits, '_source'),
            'last_sort' => $lastSort,
            'total'     => $response['hits']['total']['value'],
        ];
    }
}

最佳实践 ​

  1. 连接复用：使用单例模式管理 Elasticsearch Client，避免频繁创建连接
  2. 批量写入：所有写操作尽量使用 _bulk API，每批 1000~5000 条
  3. 字段过滤：搜索时使用 _source 指定返回字段，减少网络传输量
  4. 错误重试：对超时和连接错误实现指数退避重试策略
  5. 版本控制：使用乐观锁（seq_no + primary_term）处理并发更新
  6. 深度分页：避免 from + size 深度分页，使用 scroll 或 search_after

下一节

继续学习：索引与查询

参考链接 ​