elasticsearch——Java API的使用

雨点打透心脏的1/2处 2022-06-08 12:26 179阅读 0赞

[前一篇有关ES的文章][ES]介绍了使用Rest方式调用ES的聚合API。Rest API使用了HTTP协议，按理来说，可以直接使用类似HttpClient的工具直接调用Rest API。虽然笔者并没有尝试过，但稍微想想一下就知道这种方法是可行的。这种方法主要有下面几个弊端：

1.  需要开启ES的Http服务和端口。ES提供的Http服务功能非常全面，没有提供权限控制，防护也比较脆弱。一旦遭到破解，则数据面临极大的风险。所以，建议在生产中关闭Http服务，或者自己增加一层代理来实现权限控制。
2.  调用比较困难。Rest API的核心是url和post数据，url直接需传入字符串，这样就不能使用IDE的查错功能。需要记忆的东西太多，不确定时就要去查API，影响开发效率。
3.  Http协议的一大特点是无连接性。也就是每一次请求都需要建立新的连接，我们知道tcp连接是比较耗时的过程。从性能的角度来说，直接使用Rest API也是不合适的。

ES所提供的Http服务适合用作集群状态和数据的监控，而不适合直接用于数据操作。ES提供了多种语言（包括Java、Python、PHP、Ruby等）版本的Client API，可以使用这些Client API编程实现数据操作功能。作为一个Java语言编程者，本文主要介绍使用Java版本的Client来操作数据。使用Java API需要依赖ES所提供的jar包，我们使用[maven][]来下载所需的依赖包，maven依赖定义如下：

<dependency>
        <groupId>org.elasticsearch</groupId>
        <artifactId>elasticsearch</artifactId>
        <version>1.5.0</version>
    </dependency>

version表示依赖包的版本，可以输入任意存在的版本，本文的示例中使用1.5.0版的API。注意，建议API的版本与ES集群所使用的版本保持一致，以免出现因版本不一致而导致的冲突。

本文的主要内容包括：

1.  介绍两类Client，解释它们的之间的差异；
2.  使用Client进行index、document和聚合相关的操作。

### 1. Client ###

ES中所有的Java API调用都要使用Client对象，ES为API调用者提供了两类Client对象：NodeClient和TransportClient。下面来讲讲这两类Client的差异和使用场景。

#### 1.1 NodeClient ####

NodeClient是一种嵌入式节点客户端。它首先在客户端启动一个节点（Node），并加入同名集群内。这个节点可以保存数据，并且数据能够被索引。然后从这个节点中获取Client，这类Client就是NodeClient。NodeClient无需指明ES服务端的地址，操作的数据位于启动的节点所在的集群中。下面是获得NodeClient的代码：

import org.elasticsearch.client.Client;
    import org.elasticsearch.client.node.NodeClient;
    import org.elasticsearch.node.Node;
    import static org.elasticsearch.node.NodeBuilder.nodeBuilder;
    
    public class MyNodeClient { 
    
        public static void main(String[] args) {
            // 启动一个本地节点，并加入子网内的ES集群
            Node node = nodeBuilder()
                        .clusterName("elasticsearch") // 要加入的集群名为elasticsearch
                        // .client(true) //如果设置为true，则该节点不会保存数据
                        .data(true) // 本嵌入式节点可以保存数据
                        .node(); // 构建并启动本节点
    
            // 获得一个Client对象，该对象可以对子网内的“elasticsearch”集群进行相关操作。
            Client nodeClient = node.client();
        }
    }

运行这段代码之后，可以看到工程中新增了一个data文件夹，这是因为`data(true)`将Node设置为可以存放数据的节点，数据正是放在了data文件夹下。

![新增加的data文件夹][data]

NodeClient适合用作单元或集成测试，而不适合用于生产环境。

#### 1.2 TransportClient ####

TransportClient连接远端的ES集群，其本身并不会加入集群。创建TransportClient的代码如下：

import org.elasticsearch.client.Client;
    import org.elasticsearch.client.transport.TransportClient;
    import org.elasticsearch.common.settings.Settings;
    import org.elasticsearch.common.transport.InetSocketTransportAddress;
    import org.elasticsearch.common.transport.TransportAddress;
    
    import static org.elasticsearch.common.settings.ImmutableSettings.settingsBuilder;
    
    public class MyTransportClient { 
    
        public static void main(String[] args) {
            // 配置信息
            Settings esSetting = settingsBuilder()
                                    .put("cluster.name", "elasticsearch")
                                    .build();
            TransportClient transportClient = new TransportClient(esSetting);
    
            // 添加连接地址
            TransportAddress address = new InetSocketTransportAddress("192.168.1.110", 9300);
            TransportAddress address2 = new InetSocketTransportAddress("192.168.1.111", 9300);
            transportClient.addTransportAddress(address);
            transportClient.addTransportAddress(address2);
        }
    }

TransportClient适合用于生产环境中。

### 2. Index操作 ###

本小节介绍如果使用Java API创建和删除索引。

#### 2.1 创建索引 ####

废话先不说，上代码先。下面的方法创建一个索引，并同时创建一个mapping。mapping可以传入符合格式要求的json字符串。一般情况下，我们可以使用下面的方式来生成所需的json字符串。

1.  手动拼接json字符串
2.  使用类似jackson的工具将对象转换为相应的json字符串
3.  使用ES内置的XContentFactory.jsonBuilder()来创建json字符串。

本文的示例中均使用ES自带的XContentFactory.jsonBuilder()来构建json字符串。

/** * 创建一个索引 * @param indexName 索引名 */
    public void createIndex(String indexName) {
        try {
            CreateIndexResponse indexResponse = this.client
                                    .admin()
                                    .indices()
                                    .prepareCreate(indexName)
                                    .get();
    
            System.out.println(indexResponse.isAcknowledged()); // true表示创建成功
        } catch (ElasticsearchException e) {
            e.printStackTrace();
        }
    }

如果需要再索引上新建mapping，可通过下面的代码来实现。

/** * 给索引增加mapping。 * @param index 索引名 * @param type mapping所对应的type */
    public void addMapping(String index, String type) {
        try {
            // 使用XContentBuilder创建Mapping
            XContentBuilder builder = 
                XContentFactory.jsonBuilder()
                                .startObject()
                                    .field("properties")
                                        .startObject()
                                            .field("name")
                                                .startObject()
                                                    .field("index", "not_analyzed")
                                                    .field("type", "string")
                                                .endObject()
                                            .field("age")
                                                .startObject()
                                                    .field("index", "not_analyzed")
                                                    .field("type", "integer")
                                                .endObject()
                                        .endObject()
                                .endObject();
            System.out.println(builder.string());           
            PutMappingRequest mappingRequest = Requests.putMappingRequest(index).source(builder).type(type);
                    this.client.admin().indices().putMapping(mappingRequest).actionGet();
        } catch (ElasticsearchException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

#### 3.1 删除索引 ####

/** * 删除索引 * @param index 要删除的索引名 */
    public void deleteIndex(String index) {
        DeleteIndexResponse deleteIndexResponse = 
            this.client
                .admin()
                .indices()
                .prepareDelete(index)
                .get();
        System.out.println(deleteIndexResponse.isAcknowledged()); // true表示成功
    }

### 3. 文档CURD操作 ###

增删改查是数据的基本操作，同时也是使用频率最高的一类操作。本小节介绍使用Java API来实现document的增删改查。

#### 3.1 新增文档 ####

/** * 创建一个文档 * @param index index * @param type type */
    public void createDoc(String index, String type) {
    
        try {
            // 使用XContentBuilder创建一个doc source
            XContentBuilder builder = 
                XContentFactory.jsonBuilder()
                                .startObject()
                                    .field("name", "zhangsan")
                                    .field("age", "lisi")
                                .endObject();
    
            IndexResponse indexResponse = this.client
                                            .prepareIndex()
                                            .setIndex(index)
                                            .setType(type)
                                            // .setId(id) // 如果没有设置id，则ES会自动生成一个id
                                            .setSource(builder.string())
                                            .get();
            System.out.println(indexResponse.isCreated());
        } catch (ElasticsearchException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

#### 3.2 更新文档 ####

/** * 更新文档 * @param index * @param type * @param id */
    public void updateDoc(String index, String type, String id) {
        try {
            XContentBuilder builder = 
                XContentFactory.jsonBuilder()
                                .startObject()
                                    .field("name", "lisi")
                                    .field("age", 12)
                                .endObject();
    
            UpdateResponse updateResponse = 
                this.client
                    .prepareUpdate()
                    .setIndex(index)
                    .setType(type)
                    .setId(id)
                    .setDoc(builder.string())
                    .get();
            System.out.println(updateResponse.isCreated()); // true表示成功
        } catch (ElasticsearchException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

注意，id参数必须是确定存在的id值，否则会抛出document missing的异常。

#### 3.3 查询文档 ####

查询文档可以是根据id查询，也可以是根据复杂的查询条件查询。根据id的get查询代码如下。

/** * 根据ID查询一条数据记录。 * @param id 要查询数据的ID。 * @return 返回查询出来的记录对象的json字符串。 */
    public String get(String index, String type, String id) {
        GetResponse getResponse = this.client
                                    .prepareGet()   // 准备进行get操作，此时还有真正地执行get操作。（与直接get的区别）
                                    .setIndex(index)  // 要查询的
                                    .setType(type)
                                    .setId(id)
                                    .get();
        return getResponse.getSourceAsString();
    }
    ````
    
    基于复杂查询条件的示例代码如下。
    
    
    
    
    
    <div class="se-preview-section-delimiter"></div>
    
    ```java
    /** * 使用filter方式查询数据。 * @param index 数据所在的索引名 * @param type 数据所在的type * @return */
    public List<String> queryByFilter(String index, String type) {
    
        // 查询名为zhangsan的数据
        FilterBuilder filterBuilder = FilterBuilders.termFilter("name", "zhangsan");
        SearchResponse searchResponse = 
            this.client
                .prepareSearch()  
                .setIndices(index)
                .setTypes(type)
                .setPostFilter(filterBuilder)
                .get();
    
        List<String> docList = new ArrayList<String>();
        SearchHits searchHits = searchResponse.getHits();
        for (SearchHit hit : searchHits) {
            docList.add(hit.getSourceAsString());
        }
        return docList;
    }

#### 3.4 删除文档 ####

下面的代码删除指定id的文档。

/** * 删除一条数据 * @param index * @param type * @param id */
    public void deleteDoc(String index, String type, String id) {
        DeleteResponse deleteResponse  = this.client
                .prepareDelete()  
                .setIndex(index)
                .setType(type)
                .setId(id)
                .get();
        System.out.println(deleteResponse.isFound()); // true表示成功
    }

根据复杂的查询条件来删除文档。

/** * 根据查询条件删除文档。 */
    public void deleteByQuery(String index, String type) {
        try {
            QueryBuilder queryBuilder = QueryBuilders.termQuery("name", "zhangsan");
            DeleteByQueryResponse deleteByQueryResponse = this.client
                    .prepareDeleteByQuery(index)
                    .setTypes(type)
                    .setQuery(queryBuilder)
                    .get();
        } catch (ElasticsearchException e) {
            e.printStackTrace();
        }
    }

### 4. 聚合操作 ###

聚合操作的API稍微比较复杂一点，本文仅以min聚合的示例来说明聚合API的调用方式，其他的聚合API调用步骤类似。

/** * 使用min聚合查询某个字段上最小的值。 * @param index * @param type */
    public void min(String index, String type) {
        SearchResponse response = this.client
                                .prepareSearch(index)
                                .addAggregation(AggregationBuilders.min("min").field("age"))
                                .get();
    
        InternalMin min = response.getAggregations().get("min");
        System.out.println(min.getValue());

`}````参考:https://github.com/xialei199023/blog-demo`

[ES]: http://blog.csdn.net/xialei199023/article/details/48298635
[maven]: http://maven.apache.org/
[data]: /images/20220608/bb0c0da68b3f492396ff9bd5e5619655.png