PHP小码农

Appearance

Elasticsearch 最佳实践 ​

在生产环境中使用 Elasticsearch 需要考虑索引设计、性能优化、集群运维和数据同步等多个维度。本节将系统性地总结 Elasticsearch 在 PHP 项目中的最佳实践，帮助团队建立可维护、高性能的搜索基础设施。

前置知识

阅读本节前，建议先了解：Elasticsearch 基础、PHP 客户端操作 和 索引与查询进阶

索引设计策略 ​

索引划分原则 ​

合理的索引设计是 Elasticsearch 性能的基础。核心原则是：一个索引对应一类文档。

// 好的设计：按业务实体分索引
products_v1          -> 商品数据
articles_v1          -> 文章数据
orders_2024_01       -> 订单数据（按月分片）
logs_app_2024_06_15  -> 应用日志（按日分片）

// 差的设计：一个大索引包含所有数据
all_data             -> 混合商品、文章、订单...

索引命名规范 ​

php
<?php

declare(strict_types=1);

namespace App\Constants;

class ElasticsearchIndexConstants
{
    // 索引命名格式：{业务}_{数据类型}_{版本}
    // 对于时间序列数据：{业务}_{数据类型}_{yyyy_MM}

    // 业务索引（带版本号，支持零停机重建）
    public const INDEX_PRODUCTS    = 'ecommerce_products_v1';
    public const INDEX_ARTICLES    = 'cms_articles_v1';
    public const INDEX_USERS       = 'user_profiles_v1';

    // 时间序列索引（按月自动创建）
    public const INDEX_ORDERS_PREFIX = 'ecommerce_orders_';
    public const INDEX_LOGS_PREFIX   = 'app_logs_';

    // 别名（应用程序始终通过别名访问）
    public const ALIAS_PRODUCTS = 'products_read';
    public const ALIAS_PRODUCTS_WRITE = 'products_write';
    public const ALIAS_ORDERS  = 'orders_latest';

    /**
     * 获取当月订单索引名
     */
    public static function getOrderIndexName(string $date = ''): string
    {
        $date = $date ?: date('Y_m');
        return self::INDEX_ORDERS_PREFIX . $date;
    }

    /**
     * 获取当日日志索引名
     */
    public static function getLogIndexName(string $date = ''): string
    {
        $date = $date ?: date('Y_m_d');
        return self::INDEX_LOGS_PREFIX . $date;
    }
}

分片数量规划 ​

分片数量直接影响查询性能和集群扩展能力。

php
<?php

declare(strict_types=1);

namespace App\Service;

class ShardPlanner
{
    /**
     * 计算合理的分片数量
     *
     * 原则：
     * 1. 每个分片大小建议 10~50GB
     * 2. 单节点分片数不超过 20 个（避免资源竞争）
     * 3. 预留扩展空间（考虑未来数据增长）
     * 4. 分片数一旦确定不可更改，需要合理预估
     */
    public static function calculateShards(
        float $expectedDataGB,
        float $growthRate = 0.2,
        int $plannedYears = 2,
        int $targetShardSizeGB = 30
    ): int {
        // 预估最终数据量
        $finalSizeGB = $expectedDataGB * (1 + $growthRate) ** $plannedYears;
        // 计算分片数（向上取整）
        return (int) ceil($finalSizeGB / $targetShardSizeGB);
    }

    /**
     * 常见场景分片建议
     */
    public static function getRecommendation(string $scenario): array
    {
        return match ($scenario) {
            'small_product' => [
                'description'   => '小型电商（< 10万商品）',
                'shards'        => 3,
                'replicas'      => 1,
                'refresh'       => '1s',
                'estimated_size'=> '~2GB',
            ],
            'medium_product' => [
                'description'   => '中型电商（10~100万商品）',
                'shards'        => 5,
                'replicas'      => 1,
                'refresh'       => '1s',
                'estimated_size'=> '~20GB',
            ],
            'large_product' => [
                'description'   => '大型电商（> 100万商品）',
                'shards'        => 10,
                'replicas'      => 1,
                'refresh'       => '5s',
                'estimated_size'=> '~100GB',
            ],
            'log_daily' => [
                'description'   => '日志索引（按天滚动）',
                'shards'        => 3,
                'replicas'      => 1,
                'refresh'       => '5s',
                'estimated_size'=> '~10GB/day',
                'rollover'      => 'max_size: 50gb, max_age: 7d',
            ],
            default => [
                'description' => '默认配置',
                'shards'       => 3,
                'replicas'     => 1,
                'refresh'      => '1s',
            ],
        };
    }
}

分片数不可修改

索引创建后，主分片数量无法修改（副本分片数可以随时调整）。如果分片数规划不当，只能通过 Reindex 到新索引来解决。因此，规划时宁多勿少，但要避免过多（单节点建议不超过 20 个分片）。

性能优化 ​

写入优化 ​

php
<?php

declare(strict_types=1);

namespace App\Service;

use Elasticsearch\Client;

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

    /**
     * 批量写入前的优化配置
     * 在大批量数据导入时调用
     */
    public function prepareForBulkWrite(string $index): void
    {
        $this->esClient->indices()->putSettings([
            'index' => $index,
            'body'  => [
                'settings' => [
                    'refresh_interval'       => '-1',       // 禁用自动刷新
                    'number_of_replicas'     => 0,           // 暂时关闭副本
                    'translog.durability'    => 'async',     // 异步写入 translog
                    'translog.sync_interval' => '30s',       // translog 同步间隔
                    'index.indexing.slowlog.threshold.index.warn' => '10s',
                    'index.indexing.slowlog.level' => 'INFO',
                ],
            ],
        ]);
    }

    /**
     * 批量写入完成后恢复配置
     */
    public function restoreAfterBulkWrite(string $index, int $replicas = 1): void
    {
        // 先手动刷新，确保数据可搜索
        $this->esClient->indices()->refresh(['index' => $index]);

        $this->esClient->indices()->putSettings([
            'index' => $index,
            'body'  => [
                'settings' => [
                    'refresh_interval'    => '1s',
                    'number_of_replicas'  => $replicas,
                    'translog.durability' => 'request',
                ],
            ],
        ]);

        // 强制合并 Segment（减少段数，提升后续查询性能）
        $this->esClient->indices()->forcemerge([
            'index' => $index,
            'body'  => [
                'max_num_segments' => 1,
            ],
        ]);
    }

    /**
     * 优化的批量写入方法
     */
    public function optimizedBulkImport(string $index, iterable $data, int $batchSize = 3000): array
    {
        $this->prepareForBulkWrite($index);

        $stats = ['total' => 0, 'success' => 0, 'errors' => 0];
        $batch = [];
        $count = 0;

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

            $batch[] = ['index' => ['_index' => $index, '_id' => $id]];
            $batch[] = $doc;
            $count++;

            if ($count >= $batchSize) {
                $result = $this->esClient->bulk([
                    'body'    => $batch,
                    'timeout' => '60s',
                ]);

                foreach ($result['items'] as $item) {
                    $action = array_key_first($item);
                    $status = $item[$action]['status'] ?? 0;
                    if ($status >= 200 && $status < 300) {
                        $stats['success']++;
                    } else {
                        $stats['errors']++;
                    }
                }

                $stats['total'] += $count;
                $batch = [];
                $count = 0;
            }
        }

        // 写入最后一批
        if (!empty($batch)) {
            $result = $this->esClient->bulk(['body' => $batch, 'timeout' => '60s']);
            foreach ($result['items'] as $item) {
                $action = array_key_first($item);
                $status = $item[$action]['status'] ?? 0;
                if ($status >= 200 && $status < 300) {
                    $stats['success']++;
                } else {
                    $stats['errors']++;
                }
            }
            $stats['total'] += $count;
        }

        $this->restoreAfterBulkWrite($index);

        return $stats;
    }
}

查询优化 ​

php
<?php

declare(strict_types=1);

namespace App\Service;

use Elasticsearch\Client;

class QueryOptimizer
{
    /**
     * 查询优化要点清单
     *
     * 1. 使用 filter 替代 must（filter 不计算评分，可被缓存）
     * 2. 使用 _source 过滤返回字段，减少网络传输
     * 3. 避免 wildcard 前缀通配符查询
     * 4. 避免 script 查询（性能差）
     * 5. 使用 routing 将相关数据路由到同一分片
     * 6. 避免 depth > 2 的嵌套聚合
     * 7. 对大数据集使用 search_after 代替 from+size
     */
    public function getOptimizedQuery(array $criteria): array
    {
        $query = ['bool' => []];

        // 全文搜索使用 must（需要评分）
        if (!empty($criteria['keyword'])) {
            $query['bool']['must'][] = [
                'multi_match' => [
                    'query'  => $criteria['keyword'],
                    'fields' => ['title^3', 'description^1'],
                    'type'   => 'best_fields',
                ],
            ];
        } else {
            $query['bool']['must'][] = ['match_all' => (object)[]];
        }

        // 精确过滤使用 filter（不需要评分，可缓存）
        $filters = [];
        if (!empty($criteria['category'])) {
            $filters[] = ['term' => ['category' => $criteria['category']]];
        }
        if (!empty($criteria['brand'])) {
            $filters[] = ['term' => ['brand' => $criteria['brand']]];
        }
        if (isset($criteria['min_price'])) {
            $filters[] = ['range' => [
                'price' => ['gte' => (float) $criteria['min_price']],
            ]];
        }

        if (!empty($filters)) {
            $query['bool']['filter'] = $filters;
        }

        return $query;
    }

    /**
     * 使用 routing 优化查询（同一用户的订单在同一分片上）
     */
    public function searchWithRouting(Client $client, string $index, string $userId): array
    {
        return $client->search([
            'index'   => $index,
            'routing' => $userId,   // 只查询该用户所在分片
            'body'    => [
                'query' => [
                    'term' => ['user_id' => $userId],
                ],
            ],
        ]);
    }
}

字段级优化 ​

php
<?php

// Mapping 优化建议

// 1. 不需要搜索/聚合的字段，设置 enabled: false 或 index: false
"title": { "type": "text" },
"image_url": { "type": "keyword", "index": false },  // 只存储，不索引
"raw_data": { "type": "object", "enabled": false },   // 完全不索引不解析

// 2. 不需要评分的字段使用 doc_values: false（减少内存占用）
// 仅适用于不需要排序和聚合的字段
"session_id": { "type": "keyword", "doc_values": false }

// 3. 不需要精确值的 text 字段关闭 norms
"description": {
    "type": "text",
    "norms": false,           // 不需要根据字段长度计算评分
    "index_options": "freqs"  // 不需要位置信息
}

// 4. 使用 ignore_above 限制 keyword 长度
"tags": { "type": "keyword", "ignore_above": 256 }

// 5. 数值类型选择最小够用的类型
"age": { "type": "byte" },        // 0~127，而非 integer
"is_active": { "type": "boolean" }, // 而非 integer
"stock": { "type": "integer" },     // 而非 long

集群运维 ​

集群监控 ​

php
<?php

declare(strict_types=1);

namespace App\Service;

use Elasticsearch\Client;

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

    /**
     * 获取集群健康状态
     */
    public function getClusterHealth(): array
    {
        return $this->esClient->cluster()->health();
    }

    /**
     * 获取节点状态
     */
    public function getNodeStats(): array
    {
        return $this->esClient->nodes()->stats([
            'metric' => 'jvm,os,process,fs,indices',
        ]);
    }

    /**
     * 获取索引状态
     */
    public function getIndexStats(string $index = '_all'): array
    {
        return $this->esClient->indices()->stats([
            'index'  => $index,
            'metric' => 'store,docs,indexing,search',
        ]);
    }

    /**
     * 获取 pending tasks
     */
    public function getPendingTasks(): array
    {
        return $this->esClient->cluster()->pendingTasks();
    }

    /**
     * 综合健康检查
     */
    public function healthCheck(): array
    {
        $health = $this->getClusterHealth();
        $nodes = $this->getNodeStats();
        $pending = $this->getPendingTasks();

        $status = $health['status'] ?? 'unknown';
        $nodeCount = $health['number_of_nodes'] ?? 0;
        $dataNodeCount = $health['number_of_data_nodes'] ?? 0;
        $unassignedShards = $health['unassigned_shards'] ?? 0;
        $pendingTasks = $pending['pending_task_count'] ?? 0;

        $isHealthy = $status === 'green'
            && $unassignedShards === 0
            && $pendingTasks < 100;

        // 检查 JVM 堆使用率
        $heapWarning = false;
        foreach ($nodes['nodes'] ?? [] as $nodeId => $node) {
            $heapPercent = $node['jvm']['mem']['heap_used_percent'] ?? 0;
            if ($heapPercent > 85) {
                $heapWarning = true;
                break;
            }
        }

        // 检查磁盘使用率
        $diskWarning = false;
        foreach ($nodes['nodes'] ?? [] as $nodeId => $node) {
            foreach ($node['fs']['data'] ?? [] as $fs) {
                $diskPercent = $fs['disk']['disk_used_percent'] ?? 0;
                if ($diskPercent > 85) {
                    $diskWarning = true;
                    break 2;
                }
            }
        }

        return [
            'healthy'           => $isHealthy,
            'status'            => $status,
            'node_count'        => $nodeCount,
            'data_node_count'   => $dataNodeCount,
            'unassigned_shards' => $unassignedShards,
            'pending_tasks'     => $pendingTasks,
            'heap_warning'      => $heapWarning,
            'disk_warning'      => $diskWarning,
            'checked_at'        => date('Y-m-d H:i:s'),
        ];
    }
}

索引维护 ​

php
<?php

declare(strict_types=1);

namespace App\Service;

use Elasticsearch\Client;

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

    /**
     * Force Merge（段合并，减少查询开销）
     * 建议在低峰期执行
     */
    public function forceMerge(string $index, int $maxSegments = 1): array
    {
        return $this->esClient->indices()->forcemerge([
            'index' => $index,
            'body'  => [
                'max_num_segments' => $maxSegments,
            ],
        ]);
    }

    /**
     * 清除索引缓存
     */
    public function clearCache(string $index = '_all'): array
    {
        return $this->esClient->indices()->clearCache([
            'index' => $index,
        ]);
    }

    /**
     * 刷新索引（立即生效）
     */
    public function refresh(string $index): array
    {
        return $this->esClient->indices()->refresh(['index' => $index]);
    }

    /**
     * 清理旧的日志索引
     */
    public function cleanupOldIndices(string $prefix, int $keepDays = 30): array
    {
        $stats = $this->esClient->indices()->stats(['index' => "{$prefix}*"]);
        $indices = array_keys($stats['indices'] ?? []);
        $deleted = [];
        $cutoff = strtotime("-{$keepDays} days");

        foreach ($indices as $index) {
            // 从索引名中提取日期
            if (preg_match('/(\d{4}_\d{2}_\d{2})$/', $index, $matches)) {
                $indexDate = str_replace('_', '-', $matches[1]);
                if (strtotime($indexDate) < $cutoff) {
                    $this->esClient->indices()->delete(['index' => $index]);
                    $deleted[] = $index;
                }
            }
        }

        return ['deleted_indices' => $deleted];
    }

    /**
     * Reroute（手动移动分片）
     */
    public function rerouteShard(string $index, int $shardId, string $fromNode, string $toNode): array
    {
        return $this->esClient->cluster()->reroute([
            'body' => [
                'commands' => [
                    [
                        'move' => [
                            'index'     => $index,
                            'shard'     => $shardId,
                            'from_node' => $fromNode,
                            'to_node'   => $toNode,
                        ],
                    ],
                ],
            ],
        ]);
    }
}

数据同步策略 ​

MySQL -> Elasticsearch 数据同步 ​

php
<?php

declare(strict_types=1);

namespace App\Service;

use Elasticsearch\Client;
use PDO;

class DataSyncService
{
    public function __construct(
        private readonly Client $esClient,
        private readonly PDO $pdo
    ) {}

    /**
     * 全量同步（适用于初始导入）
     */
    public function fullSync(string $index, string $table, int $batchSize = 2000): array
    {
        $offset = 0;
        $stats = ['total' => 0, 'indexed' => 0, 'errors' => 0];

        while (true) {
            $stmt = $this->pdo->query(
                "SELECT * FROM {$table} LIMIT {$offset}, {$batchSize}"
            );
            $rows = $stmt->fetchAll(PDO::FETCH_ASSOC);

            if (empty($rows)) {
                break;
            }

            $body = [];
            foreach ($rows as $row) {
                $id = (string) $row['id'];
                unset($row['id']);
                $body[] = ['index' => ['_index' => $index, '_id' => $id]];
                $body[] = $this->transformDocument($row);
            }

            $result = $this->esClient->bulk(['body' => $body]);

            foreach ($result['items'] as $item) {
                $action = array_key_first($item);
                $status = $item[$action]['status'] ?? 0;
                if ($status >= 200 && $status < 300) {
                    $stats['indexed']++;
                } else {
                    $stats['errors']++;
                }
            }

            $stats['total'] += count($rows);
            $offset += $batchSize;
        }

        $this->esClient->indices()->refresh(['index' => $index]);

        return $stats;
    }

    /**
     * 增量同步（基于 updated_at 时间戳）
     */
    public function incrementalSync(
        string $index,
        string $table,
        string $lastSyncTime,
        int $batchSize = 1000
    ): array {
        $stats = ['updated' => 0, 'deleted' => 0];
        $offset = 0;

        while (true) {
            $stmt = $this->pdo->prepare(
                "SELECT * FROM {$table} WHERE updated_at > ? LIMIT ?, ?"
            );
            $stmt->execute([$lastSyncTime, $offset, $batchSize]);
            $rows = $stmt->fetchAll(PDO::FETCH_ASSOC);

            if (empty($rows)) {
                break;
            }

            $body = [];
            foreach ($rows as $row) {
                $id = (string) $row['id'];
                $body[] = ['update' => ['_index' => $index, '_id' => $id]];
                $body[] = ['doc' => $this->transformDocument($row), 'doc_as_upsert' => true];
            }

            $result = $this->esClient->bulk(['body' => $body]);
            $stats['updated'] += count($rows);
            $offset += $batchSize;
        }

        // 同步已删除的记录
        $stmt = $this->pdo->prepare(
            "SELECT id FROM {$table}_deleted WHERE deleted_at > ?"
        );
        $stmt->execute([$lastSyncTime]);
        $deletedIds = $stmt->fetchAll(PDO::FETCH_COLUMN);

        if (!empty($deletedIds)) {
            $body = [];
            foreach ($deletedIds as $id) {
                $body[] = ['delete' => ['_index' => $index, '_id' => (string) $id]];
            }
            $this->esClient->bulk(['body' => $body]);
            $stats['deleted'] = count($deletedIds);
        }

        return $stats;
    }

    /**
     * 基于 Canal/binlog 的实时同步（推荐方案）
     * 此处展示 PHP 端消费 binlog 事件的伪代码
     */
    public function binlogSyncListener(): void
    {
        // 伪代码：消费 Canal 或 Debezium 推送的 binlog 事件
        // $consumer = new CanalConsumer('canal_host', 11111);
        // $consumer->subscribe('my_database', 'products');
        //
        // while ($event = $consumer->receive()) {
        //     match ($event->type) {
        //         'INSERT' => $this->esClient->index([
        //             'index' => $index,
        //             'id'    => $event->data['id'],
        //             'body'  => $event->data,
        //         ]),
        //         'UPDATE' => $this->esClient->update([
        //             'index' => $index,
        //             'id'    => $event->data['id'],
        //             'body'  => ['doc' => $event->data],
        //         ]),
        //         'DELETE' => $this->esClient->delete([
        //             'index' => $index,
        //             'id'    => $event->data['id'],
        //         ]),
        //     };
        // }
    }

    /**
     * 文档转换（MySQL 字段 -> ES 文档）
     */
    private function transformDocument(array $row): array
    {
        return [
            'title'         => $row['title'] ?? '',
            'description'   => $row['description'] ?? '',
            'price'         => (float) ($row['price'] ?? 0),
            'category'      => $row['category'] ?? '',
            'brand'         => $row['brand'] ?? '',
            'tags'          => json_decode($row['tags'] ?? '[]', true),
            'sales_count'   => (int) ($row['sales_count'] ?? 0),
            'rating'        => (float) ($row['rating'] ?? 0),
            'is_on_sale'    => (bool) ($row['is_on_sale'] ?? true),
            'created_at'    => $row['created_at'] ?? date('Y-m-d H:i:s'),
            'updated_at'    => $row['updated_at'] ?? date('Y-m-d H:i:s'),
        ];
    }
}

同步方案对比 ​

方案实时性复杂度数据一致性适用场景
定时全量同步低（分钟级）低最终一致日志、报表类数据
定时增量同步中（秒~分钟级）中最终一致一般业务数据
Binlog 实时同步高（毫秒级）高近实时一致核心业务数据
双写（DB + ES）高中可能不一致简单场景，不推荐
MQ 异步同步高（毫秒级）中最终一致中高实时性要求

推荐方案

  • 核心业务数据（商品、订单）：使用 Canal/Debezium 消费 MySQL binlog 实现近实时同步
  • 报表/日志数据：使用定时增量同步（每分钟~每小时）
  • 初始导入：使用全量批量导入，配合写入优化参数

注意事项 ​

安全配置 ​

yaml
# 生产环境安全配置清单
# 1. 开启认证
xpack.security.enabled: true

# 2. 启用 HTTPS
xpack.security.http.ssl.enabled: true
xpack.security.http.ssl.key: certs/server.key
xpack.security.http.ssl.certificate: certs/server.crt

# 3. 启用传输层 SSL
xpack.security.transport.ssl.enabled: true
xpack.security.transport.ssl.verification_mode: certificate
xpack.security.transport.ssl.keystore.path: certs/elastic-certificates.p12
xpack.security.transport.ssl.truststore.path: certs/elastic-certificates.p12

# 4. 禁用匿名访问
# 5. 配置基于角色的访问控制（RBAC）
# 6. 开启审计日志
xpack.security.audit.enabled: true

常见错误排查 ​

现象可能原因解决方案
集群 Yellow副本分片未分配检查节点数是否足够，分配分片
集群 Red主分片未分配检查磁盘空间，查看分配失败原因
搜索超时分片过多或查询复杂优化查询，增加超时时间，检查慢查询日志
JVM OOM堆内存不足或查询过大增大堆内存，优化查询，启用 circuit breaker
数据不一致同步延迟或失败检查同步机制，实施对账校验
磁盘满Segment 文件积累定期 Force Merge，清理旧索引

最佳实践 ​

  1. 索引设计：按业务划分索引，合理规划分片数量，使用别名统一访问入口
  2. Mapping：预定义严格 Mapping，禁用动态映射，选择最小够用的数据类型
  3. 写入优化：批量使用 bulk API，大批量导入时关闭 refresh 和副本
  4. 查询优化：filter 替代 must 做精确过滤，限制返回字段，避免深度分页
  5. 数据同步：核心数据使用 binlog 同步，辅助数据使用定时增量同步
  6. 监控告警：监控集群健康状态、JVM 堆使用率、磁盘空间、慢查询日志
  7. 安全防护：开启认证和 HTTPS，配置 RBAC，定期备份索引数据

下一节

继续学习：缓存策略概览

参考链接 ​