hyperf之elasticsearch封装

<?php declare (strict_types = 1); namespace App\Elasticsearch; use Elasticsearch\ClientBuilder; use Hyperf\Elasticsearch\ClientBuilderFactory; use Hyperf\Utils\ApplicationContext; use Elasticsearch\Client; use Hyperf\Guzzle\RingPHP\PoolHandler; use Swoole\Coroutine; use Throwable; /** * elasticsearch数据处理. * Class ElasticSearch * @package App\ElasticSearch */ class Elasticsearch { /** * 获取ES客户端实例. * @param false|bool $factory 是否从工厂构建实例 * @return null|Client */ public static function client($factory = false) : ?Client { try { if ($factory) { $builder = ApplicationContext::getContainer()->get(ClientBuilderFactory::class)->create(); } else { $builder = ClientBuilder::create(); if (Coroutine::getCid() > 0) { $handler = make(PoolHandler::class, [ 'option' => [ 'max_connections' => config('elasticsearch.max_connections'), ], ]); $builder->setHandler($handler); } } return $builder->setHosts([config('elasticsearch.host')])->build(); } catch (Throwable $e) { logger()->error(__METHOD__, [$e->getCode(), $e->getMessage()]); return null; } } /** * Get Info. * @param null|Client $client 客户端实例 * @param array $params 参数 Associative array of parameters * @return array|callable */ public static function info(?Client $client, $params = []) { return $client->info($params); } /** * 判断索引是否存在. * @param null|Client $client 客户端实例 * @param string $index 索引名 * @return bool **/ public static function indexExists(?Client $client, string $index): bool { return $client->indices()->exists(['index' => $index]); } /** * 创建索引. * @param null|Client $client 客户端实例 * @param string $index 索引名 * @param int $shardsNum 分片数量 * @param int $replicasNum 副本备份数量 * @return bool **/ public static function createIndex(?Client $client, string $index, int $shardsNum = 0, int $replicasNum = 0): bool { try { $params = [ 'index' => $index, 'body' => [ 'settings' => [ 'number_of_shards' => $shardsNum ?: config('elasticsearch.shards_num', 1), 'number_of_replicas' => $replicasNum ?: config('elasticsearch.replicas_num', 1), ] ] ]; $response = $client->indices()->create($params); if (empty($response['index'])) { logger()->error(__METHOD__, [$response]); return false; } } catch (Throwable $e) { logger()->error(__METHOD__, [$e->getCode(), $e->getMessage()]); return false; } return true; } /** * 更新索引. * @param null|Client $client 客户端实例 * @param string $index 索引名 * @param int $shardsNum 分片数量 * @param int $replicasNum 副本备份数量 * @return bool **/ public static function updateIndex(?Client $client, string $index, int $shardsNum = 0, int $replicasNum = 0): bool { try { $params = [ 'index' => $index, 'body' => [ 'settings' => [ 'number_of_shards' => $shardsNum ?: config('elasticsearch.shards_num', 1), 'number_of_replicas' => $replicasNum ?: config('elasticsearch.replicas_num', 1), ] ] ]; $response = $client->indices()->putSettings($params); if (empty($response['acknowledged'])) { logger()->error(__METHOD__, [$response]); return false; } } catch (Throwable $e) { logger()->error(__METHOD__, [$e->getCode(), $e->getMessage()]); return false; } return true; } /** * 删除索引. * @param null|Client $client 客户端实例 * @param string $index 索引名 * @return bool */ public static function deleteIndex(?Client $client, string $index): bool { try{ $response = $client->indices()->delete(['index' => $index]); if (empty($response['acknowledged'])) { logger()->error(__METHOD__, [$response]); return false; } } catch (Throwable $e) { logger()->error(__METHOD__, [$e->getCode(), $e->getMessage()]); return false; } return true; } /** * 插入一条文档. * @param null|Client $client 客户端实例 * @param string $index 索引名 * @param array $data 记录数据 * @param null|string $type 类型:默认=索引名 * @return bool */ public static function insert(?Client $client, string $index, array $data, string $type = null): bool { if (! $type) { $type = $index; } try{ $params = [ 'index' => $index, 'type' => $type, 'body' => $data ]; if (isset($data['id']) && ! isset($params['id'])) { $params['id'] = $data['id']; } if (isset($data['_id']) && ! isset($params['id'])) { $params['id'] = $data['_id']; } $client->index($params); // TODO:暂时缺少返回判断逻辑,暂时死信写入 } catch (Throwable $e) { logger()->error(__METHOD__, [$e->getCode(), $e->getMessage()]); return false; } return true; } /** * 聚合查询. * @param null|Client $client 客户端实例 * @param string $index 索引名 * @param string $type 类型 * @param array $fields 查询字段 * @param array $filter 过滤范围 * @param array $must 必要条件(必须全满足) * @param array $mustNot 禁止条件 * @param array $should 需要条件(至少满足一) * @param int $page 页码 * @param int $limit 每页记录数 * @param array $sort 排序条件 * @param array $uniqueFields 唯一字段(distinct) * @return array */ public static function select(?Client $client, string $index, string $type, array $fields, array $filter = [], array $must = [], array $mustNot = [], array $should = [], int $page = 1, int $limit = 10, $sort = [['_score' => ['order' => 'desc']]], $uniqueFields = []): array { $offset = max(0, ($page - 1) * $limit); $params = [ 'index' => $index, 'type' => $type, 'body' => [ 'from' => $offset, 'size' => $limit, '_source' => $fields, 'query' => ['bool' => ['filter' => $filter]], // 默认范围查询 'track_total_hits' => true, 'sort' => $sort ] ]; if ($must) { $params['body']['query']['bool']['must'] = $must; } if ($mustNot) { $params['body']['query']['bool']['must_not'] = $mustNot; } if ($should) { $params['body']['query']['bool']['should'] = $should; $params['body']['query']['bool']['minimum_should_match'] = 1; } if ($uniqueFields) { $params['body']['collapse']= $uniqueFields; $params['body']['aggs']= [ 'total_size' => [ 'cardinality' => $uniqueFields ] ]; }