Elasticsearch Java API简介

加入依赖

我本地的Elasticsearch的版本是2.1.0，因此加入相应的maven依赖

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

创建Client

Elasticsearch Client分为Node Client和TransportClient。

• Node Client：节点本身也是Elasticsearch集群的节点，也进入Elasticsearch集群和别的Elasticsearch集群中的节点一样
• TransportClient：轻量级的Client，使用Netty线程池，Socket连接到ES集群。本身不加入到集群，只作为请求的处理

一般我们使用TransportClient。创建Client的实例如下：

    private TransportClient client = null;

    @Before
    public void createElaCLient() throws UnknownHostException {
        //如果集群是默认名称的话可以不设置集群名称
        Settings settings = Settings.settingsBuilder().put("cluster.name","elasticsearch").build();
        client = TransportClient.builder().settings(settings).build().addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("master"),9300));
    }

    /**
     * 关闭ela客户端
     */
    @After
    public void closeElaClient(){
        if(client != null){
            client.close();
        }
    }

client.transport.sniff嗅探功能

你可以设置client.transport.sniff为true来使客户端去嗅探整个集群的状态，把集群中其它机器的ip地址加到客户端中，这样做的好处是一般你不用手动设置集群里所有集群的ip到连接客户端，它会自动帮你添加，并且自动发现新加入集群的机器。代码实例如下：

    private TransportClient client = null;

    @Before
    public void createElaCLient() throws UnknownHostException {
        //如果集群是默认名称的话可以不设置集群名称
        Settings settings = Settings.settingsBuilder().put("cluster.name","elasticsearch").put("client.transport.sniff",true).build();
        client = TransportClient.builder().settings(settings).build().addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("master"),9300));
    }

注意：当ES服务器监听使用内网服务器IP而访问使用外网IP时，不要使用client.transport.sniff为true，在自动发现时会使用内网IP进行通信，导致无法连接到ES服务器，而直接使用addTransportAddress方法进行指定ES服务器

测试Client连接到Elasticsearch集群

代码如下：

@Test
    public void testConnection(){
        List<DiscoveryNode> discoveryList = client.connectedNodes();
        for(DiscoveryNode node : discoveryList){
            System.out.println(node.getName());
        }
    }

创建/删除Index和Type信息

    /**
     * 创建索引
     */
    @Test
    public void createIndex(){
        if(client != null){
            client.admin().indices().create(new CreateIndexRequest("test_index")).actionGet();
        }
    }

    /**
     * 清除索引
     */
    @Test
    public void clearIndex(){
        IndicesExistsResponse indicesExistsResponse = client.admin().indices().exists(new IndicesExistsRequest("test_index")).actionGet();
        if(indicesExistsResponse.isExists()){
            client.admin().indices().delete(new DeleteIndexRequest("test_index")).actionGet();
        }
    }

    /**
     * 定义索引的映射类型（mapping）
     */
    @Test
    public void defineIndexTypeMapping(){
        try {
            XContentBuilder builder = XContentFactory.jsonBuilder();
            builder.startObject()
                    .startObject("test")
                    .startObject("properties")
                    .startObject("id").field("type","long").field("store","yes").endObject()
                    .startObject("name").field("type","string").field("store","yes").field("index","not_analyzed").endObject()
                    .endObject()
                    .endObject()
                    .endObject();
            PutMappingRequest mappingRequest = Requests.putMappingRequest("test_index").type("test").source(builder);
            client.admin().indices().putMapping(mappingRequest).actionGet();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 删除index下的某个type
     */
    @Test
    public void deleteType(){
        if(client != null){
            client.prepareDelete().setIndex("test_index").setType("test").execute().actionGet();
        }
    }

这里自定义了某个Type的索引映射（Mapping），默认ES会自动处理数据类型的映射：针对整型映射为long，浮点数为double，字符串映射为string，时间为date，true或false为boolean。

注意：针对字符串，ES默认会做“analyzed”处理，即先做分词、去掉stop words等处理再index。如果你需要把一个字符串做为整体被索引到，需要把这个字段这样设置：field(“index”, “not_analyzed”)。

索引数据

    /**
     * 批量索引
     */
    @Test
    public void indexData(){
        BulkRequestBuilder requestBuilder = client.prepareBulk();
        for(Person person : personList){
            String obj = getIndexDataFromHotspotData(person);
            if(obj != null){
                requestBuilder.add(client.prepareIndex("test_index","test",String.valueOf(person.getId())).setRefresh(true).setSource(obj));
            }
        }
        BulkResponse bulkResponse = requestBuilder.execute().actionGet();
        if(bulkResponse.hasFailures()){
            Iterator<BulkItemResponse> it = bulkResponse.iterator();
            while(it.hasNext()){
                BulkItemResponse itemResponse = it.next();
                if(itemResponse.isFailed()){
                    System.out.println(itemResponse.getFailureMessage());
                }
            }
        }
    }

    /**
     * 单个索引数据
     * @return
     */
    @Test
    public void indexHotspotData() {
        String jsonSource = getIndexDataFromHotspotData(new Person(1004,"jim"));
        if (jsonSource != null) {
            IndexRequestBuilder requestBuilder = client.prepareIndex("test_index",
                    "test").setRefresh(true);
            requestBuilder.setSource(jsonSource)
                    .execute().actionGet();
        }
    }
    public String getIndexDataFromHotspotData(Person p){
        String result = null;
        if(p != null){
            try {
                XContentBuilder builder = XContentFactory.jsonBuilder();
                builder.startObject().field("id",p.getId()).field("name",p.getName()).endObject();
                result = builder.string();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return result;
    }

查询数据

ES支持分页查询获取数据，也可以一次性获取大量数据，需要使用Scroll Search，QueryBuilder是一个查询条件

    public List<Long> searchData(QueryBuilder builder){
        List<Long> ids = new ArrayList<>();
        SearchResponse response = client.prepareSearch("test_index").setTypes("test").setQuery(builder).setSize(10).execute().actionGet();
        SearchHits hits = response.getHits();
        for(SearchHit hit : hits){
            Long id = (Long) hit.getSource().get("id");
            ids.add(id);
        }
        return ids;
    }