video_server/app/Repositories/Service/ElasticSearchService.php

<?php
/**
 * Elasticsearch 相关操作
 * @author jim <jim458734573@gmail.com>
 * @date   2020-01-22 22:37:21
 */
declare(strict_types=1);

namespace App\Repositories\Service;

use Hyperf\Guzzle\ClientFactory;

class ElasticSearchService
{
    /**
     * 在指定index 和 type 下面存储文档
     * @param int $documentId
     * @param array $document
     * @param string $typeName
     * @param string $indexName
     * @return bool
     */
    public function save($documentId, $document, $typeName = 'video', $indexName = 'dafanhao')
    {
        $url = "/{$indexName}/{$typeName}/{$documentId}";
        $result = $this->doRequest($url, 'POST', $document);
        $result = json_decode($result);
        if ($result->_version) {
            return  true;
        }
        return false;
    }

    /**
     * 更新指定文档 $document 可只传 更新的部分
     * @param int $documentId
     * @param array $document
     * @param string $typeName
     * @param string $indexName
     * @return bool
     */
    public function update($documentId, $document, $typeName = 'video', $indexName = 'dafanhao')
    {
        $url = "/{$indexName}/{$typeName}/{$documentId}/_update";
        $document = array(
            'doc' => $document
        );
        $result = $this->doRequest($url, 'POST', $document);
        $result = json_decode($result);
        if ($result->_version) {
            return  true;
        }
        return false;
    }

    /**
     * 查询
     * @param array $query 查询条件
     * @param string $typeName
     * @param string $indexName
     * @return mixed|string
     */
    public function search($query, $typeName = 'video', $indexName = 'dafanhao')
    {
        foreach ($query['query'] as $key => $item) {
            if (empty($item)) {
                unset($query['query'][$key]);
            }
        }
        $url = "/{$indexName}/{$typeName}/_search";
        $result = $this->doRequest($url, 'GET', $query);
        $result = json_decode($result, true);
        return $result;
    }

    /**
     * 统计
     * @param $query
     * @param string $typeName
     * @param string $indexName
     * @return mixed|string
     */
    public function count($query, $typeName = 'video', $indexName = 'dafanhao')
    {
        foreach ($query['query'] as $key => $item) {
            if (empty($item)) {
                unset($query['query'][$key]);
            }
        }
        $url = "/{$indexName}/{$typeName}/_count";
        $result = $this->doRequest($url, 'GET', $query);
        $result = json_decode($result);
        return $result;
    }

    /**
     * 获取指定的 document
     * @param $id
     * @param string $typeName
     * @param string $indexName
     * @return array
     */
    public function get($id, $typeName = 'video', $indexName = 'dafanhao')
    {
        $url = "/{$indexName}/{$typeName}/".$id;
        $result = $this->doRequest($url, 'GET');
        $result = json_decode($result, true);
        if ($result['found']>0) {
            return $result['_source'];
        }
        return array();
    }

    /**
     * 删除指定文档 注意 当 documentId 为空时，删除所有.
     * @param string $documentId
     * @param string $typeName
     * @param string $indexName
     * @return bool
     */
    public function delete($documentId = '', $typeName = 'video', $indexName = 'dafanhao')
    {
        $url = "/{$indexName}/{$typeName}";
        if ($documentId) {
            $url .='/'.$documentId;
            $result =   $this->doRequest($url, 'DELETE');
        } else {
            $url .='/_delete_by_query?conflicts=proceed';
            $filter = new \stdClass();
            $result =   $this->doRequest($url, 'POST', array('query'=>array('match_all'=>$filter)));
        }
        $result = json_decode($result);
        if ($result->result=='deleted' ||  isset($result->deleted)) {
            return true;
        }
        return false;
    }

    /**
     * 发起请求
     * @param string $url 请求地址
     * @param string $method 请求方法
     * @param array $data 请求参数
     * @param int $timeout 超时时间
     * @return string
     */
    private function doRequest($url, $method, $data = array(), $timeout = 40)
    {
        $esConnection = env('ELASTICSEARCH');
        if ($esConnection) {
            $url = $esConnection.$url;
        } else {
            $url = 'http://localhost:9200'.$url;
        }
        if (!in_array($method, array('DELETE','PUT','GET','POST'))) {
            return '';
        }
        $options = array(
            'timeout' => $timeout,    // 超时，单位秒.
            'verify' => false,  // 是否校验SSL证书.
            'http_errors' => false, // 如果跟随http异常头部是否中断.
            'debug' => false,    // 是否启用调试模式.
            'allow_redirects' => true, // 不跟随301／302跳转.
        );
        $client = make(ClientFactory::class)->create($options);
        try {
            $params = [
                'headers' => [
                    'Content-Type'=>'application/json',
                    'X-HTTP-Method-Override'=>$method,
                ],
                'json'=>$data,
            ];
            $result = $client->request($method, $url, $params);
            $code = $result->getStatusCode();
            if ($code == 200) {
                return $result->getBody()->getContents();
            }
        } catch (\Exception $e) {
        }
        return '';
    }
}