详解spring-boot集成elasticsearch及其简单应用

介绍

记录将elasticsearch集成到spring boot的过程，以及一些简单的应用和helper类使用。

接入方式

使用spring-boot中的spring-data-elasticsearch,可以使用两种内置客户端接入

1、节点客户端(node client)：

配置文件中设置为local:false,节点客户端以无数据节点(node-master或node-client)身份加入集群，换言之，它自己不存储任何数据，但是它知道数据在集群中的具体位置，并且能够直接转发请求到对应的节点上。

2、传输客户端(transport client)：

配置文件中设置为local:true,这个更轻量的传输客户端能够发送请求到远程集群。它自己不加入集群，只是简单转发请求给集群中的节点。

两个java客户端都通过9300端口与集群交互，使用elasticsearch传输协议(elasticsearch transport protocol)。集群中的节点之间也通过9300端口进行通信。如果此端口未开放，你的节点将不能组成集群。

环境

版本兼容

请一定注意版本兼容问题。这关系到很多maven依赖。spring data elasticsearch spring boot version matrix

搭建环境

spring boot: 1.4.1.release

spring-data-elasticsearch: 用了最基础的spring-boot-starter-data-elasticsearch，选择高版本时需要对于提高es服务版本
elasticsearch: 2.3.0

maven依赖

<parent>
   <groupid>org.springframework.boot</groupid>
   <artifactid>spring-boot-starter-parent</artifactid>
   <version>1.4.1.release</version>
   <relativepath/> <!-- lookup parent from repository -->
</parent>
<dependency>
   <groupid>org.springframework.boot</groupid>
   <artifactid>spring-boot-starter-data-elasticsearch</artifactid>
 </dependency>

配置文件

bootstrap.yml

spring:
 data:
  elasticsearch:
   # 集群名
   cluster-name: syncwt-es
   # 连接节点,注意在集群中通信都是9300端口，否则会报错无法连接上！
   cluster-nodes: localhost:9300,119.29.38.169:9300
   # 是否本地连接
   local: false
   repositories:
    # 仓库中数据存储
    enabled: true

调试

启动

启动项目，日志出现以下说明代表成功。并且没有报错。

 2017-03-30 19:35:23.078  info 20881 --- [           main] o.s.d.e.c.transportclientfactorybean     : adding transport node : localhost:9300

知识点

在elasticsearch中，文档归属于一种类型(type),而这些类型存在于索引(index)中，我们可以画一些简单的对比图来类比传统关系型数据库：

elasticsearch集群可以包含多个索引(indices)（数据库），每一个索引可以包含多个类型(types)（表），每一个类型包含多个文档(documents)（行），然后每个文档包含多个字段(fields)（列）

relational db -> databases -> tables -> rows -> columns
elasticsearch -> indices  -> types -> documents -> fields

demo

customer.java

/*
 * copyright 2012-2013 the original author or authors.
 *
 * licensed under the apache license, version 2.0 (the "license");
 * you may not use this file except in compliance with the license.
 * you may obtain a copy of the license at
 *
 *
 * unless required by applicable law or agreed to in writing, software
 * distributed under the license is distributed on an "as is" basis,
 * without warranties or conditions of any kind, either express or implied.
 * see the license for the specific language governing permissions and
 * limitations under the license.
 */

package com.syncwt.www.common.es;

import org.springframework.data.annotation.id;
import org.springframework.data.elasticsearch.annotations.document;

@document(indexname = "es-customer", type = "customer", shards = 2, replicas = 1, refreshinterval = "-1")
public class customer {

  @id
  private string id;

  private string firstname;

  private string lastname;

  public customer() {
  }

  public customer(string firstname, string lastname) {
    this.firstname = firstname;
    this.lastname = lastname;
  }

  public string getid() {
    return this.id;
  }

  public void setid(string id) {
    this.id = id;
  }

  public string getfirstname() {
    return this.firstname;
  }

  public void setfirstname(string firstname) {
    this.firstname = firstname;
  }

  public string getlastname() {
    return this.lastname;
  }

  public void setlastname(string lastname) {
    this.lastname = lastname;
  }

  @override
  public string tostring() {
    return string.format("customer[id=%s, firstname='%s', lastname='%s']", this.id,
        this.firstname, this.lastname);
  }

}

customerrepository.java

/*
 * copyright 2012-2013 the original author or authors.
 *
 * licensed under the apache license, version 2.0 (the "license");
 * you may not use this file except in compliance with the license.
 * you may obtain a copy of the license at
 *
 *
 * unless required by applicable law or agreed to in writing, software
 * distributed under the license is distributed on an "as is" basis,
 * without warranties or conditions of any kind, either express or implied.
 * see the license for the specific language governing permissions and
 * limitations under the license.
 */

package com.syncwt.www.common.es;

import org.springframework.data.elasticsearch.repository.elasticsearchrepository;

import java.util.list;

public interface customerrepository extends elasticsearchrepository<customer, string> {

  public list<customer> findbyfirstname(string firstname);

  public list<customer> findbylastname(string lastname);

}

customercontroller.java

package com.syncwt.www.web;

import com.syncwt.www.response.message;
import com.syncwt.www.service.customerservice;
import org.springframework.beans.factory.annotation.autowired;
import org.springframework.web.bind.annotation.requestmapping;
import org.springframework.web.bind.annotation.requestmethod;
import org.springframework.web.bind.annotation.restcontroller;

import java.io.ioexception;

/**
 * @version v0.0.1
 * @description customercontroller
 * @creation date 2017年03月30日 下午8:21
 * @modificationhistory who    when     what
 * --------  ----------  -----------------------------------
 */
@restcontroller
public class customercontroller {
  @autowired
  private customerservice customerservice;


  @requestmapping(value = "/test", method = requestmethod.get)
  public message test() throws ioexception {
    customerservice.savecustomers();
    customerservice.fetchallcustomers();
    customerservice.fetchindividualcustomers();
    return message.success;
  }
}

customerservice.java

package com.syncwt.www.service;

import com.syncwt.www.common.es.customer;
import com.syncwt.www.common.es.customerrepository;
import org.springframework.beans.factory.annotation.autowired;
import org.springframework.stereotype.service;

import java.io.ioexception;

/**
 * @version v0.0.1
 * @description 业务层
 * @creation date 2017年03月30日 下午8:19
 * @modificationhistory who    when     what
 * --------  ----------  -----------------------------------
 */
@service
public class customerservice {

  @autowired
  private customerrepository repository;

  public void savecustomers() throws ioexception {
    repository.save(new customer("alice", "smith"));
    repository.save(new customer("bob", "smith"));
  }

  public void fetchallcustomers() throws ioexception {
    system.out.println("customers found with findall():");
    system.out.println("-------------------------------");
    for (customer customer : repository.findall()) {
      system.out.println(customer);
    }
  }

  public void fetchindividualcustomers() {
    system.out.println("customer found with findbyfirstname('alice'):");
    system.out.println("--------------------------------");
    system.out.println(repository.findbyfirstname("alice"));

    system.out.println("customers found with findbylastname('smith'):");
    system.out.println("--------------------------------");
    for (customer customer : repository.findbylastname("smith")) {
      system.out.println(customer);
    }
  }
}

spring对es的操作方法

spring-data-elasticsearch查询方法的封装

1、封装数据库基本crud(创建（create）、更新（update）、读取（retrieve）和删除（delete）)

public interface crudrepository<t, id extends serializable>
 extends repository<t, id> {

 <s extends t> s save(s entity);

 t findone(id primarykey);    

 iterable<t> findall();     

 long count();          

 void delete(t entity);     

 boolean exists(id primarykey); 

 // … more functionality omitted.
}

2、分页排序查询

public interface pagingandsortingrepository<t, id extends serializable>
 extends crudrepository<t, id> {

 iterable<t> findall(sort sort);

 page<t> findall(pageable pageable);
}

//accessing the second page by a page size of 20
pagingandsortingrepository<user, long> repository = // … get access to a bean
page<user> users = repository.findall(new pagerequest(1, 20));

3、计数

public interface userrepository extends crudrepository<user, long> {

 long countbylastname(string lastname);
}

4、删除

public interface userrepository extends crudrepository<user, long> {

 long deletebylastname(string lastname);

 list<user> removebylastname(string lastname);

}

5、自定义查询方法自动注入

声明一个接口继承repository<t, id>

interface personrepository extends repository<person, long> { … }

接口中自定义方法，在方法名中包含t中字段名

查询关键字包括find…by, read…by, query…by, count…by, and get…by,熟悉直接可以用and and or连接

interface personrepository extends repository<person, long> {
list<person> findbylastname(string lastname);
}

保证注入了elasticsearch配置

在bootstrap.yml中写入了spring-data-elasticsearch的配置文件将自动注入

注入调用

public class someclient {

 @autowired
 private personrepository repository;

 public void dosomething() {
  list<person> persons = repository.findbylastname("matthews");
 }
}

6、支持java8 stream查询和sql语句查询

@query("select u from user u")
stream<user> findallbycustomqueryandstream();

stream<user> readallbyfirstnamenotnull();

@query("select u from user u")
stream<user> streamallpaged(pageable pageable);

try (stream<user> stream = repository.findallbycustomqueryandstream()) {
 stream.foreach(…);
}

7、支持异步查询

@async
future<user> findbyfirstname(string firstname);        

@async
completablefuture<user> findonebyfirstname(string firstname);

@async
listenablefuture<user> findonebylastname(string lastname);

支持原生es javaapi

1、nativesearchquerybuilder构建查询

@autowired
private elasticsearchtemplate elasticsearchtemplate;

searchquery searchquery = new nativesearchquerybuilder()
  .withquery(matchallquery())
  .withfilter(boolfilter().must(termfilter("id", documentid)))
  .build();

page<sampleentity> sampleentities =
  elasticsearchtemplate.queryforpage(searchquery,sampleentity.class);

2、利用scan和scroll进行大结果集查询

searchquery searchquery = new nativesearchquerybuilder()
  .withquery(matchallquery())
  .withindices("test-index")
  .withtypes("test-type")
  .withpageable(new pagerequest(0,1))
  .build();
string scrollid = elasticsearchtemplate.scan(searchquery,1000,false);
list<sampleentity> sampleentities = new arraylist<sampleentity>();
boolean hasrecords = true;
while (hasrecords){
  page<sampleentity> page = elasticsearchtemplate.scroll(scrollid, 5000l , new resultsmapper<sampleentity>()
  {
    @override
    public page<sampleentity> mapresults(searchresponse response) {
      list<sampleentity> chunk = new arraylist<sampleentity>();
      for(searchhit searchhit : response.gethits()){
        if(response.gethits().gethits().length <= 0) {
          return null;
        }
        sampleentity user = new sampleentity();
        user.setid(searchhit.getid());
        user.setmessage((string)searchhit.getsource().get("message"));
        chunk.add(user);
      }
      return new pageimpl<sampleentity>(chunk);
    }
  });
  if(page != null) {
    sampleentities.addall(page.getcontent());
    hasrecords = page.hasnextpage();
  }
  else{
    hasrecords = false;
  }
  }
}

3、获取client实例进行节点操作,可以自行封装util方法

@autowired
private elasticsearchtemplate elasticsearchtemplate;

public void searchhelper() throws ioexception {

    //节点客户端
    // on startup
//    node node = nodebuilder().clustername("syncwt-es").client(true).node();
//    client nodeclient = node.client();

    //传输客户端
//    settings settings = settings.settingsbuilder().build();
//    client transportclient = transportclient.builder().settings(settings).build();

    client transportclient = elasticsearchtemplate.getclient();

    customer customer = new customer("alice", "smith");

    // instance a json mapper
    objectmapper mapper = new objectmapper(); // create once, reuse

    // generate json
    string json = mapper.writevalueasstring(customer);
    system.out.println("--------------------------------jackson mapper");
    system.out.println(json);

    xcontentbuilder builder = jsonbuilder()
        .startobject()
        .field("firstname", "alice")
        .field("latname", "smith")
        .endobject();
    system.out.println("--------------------------------jsonbuilder");
    system.out.println(builder.string());

    indexresponse response = transportclient.prepareindex("es-customer", "customer")
        .setsource(jsonbuilder()
            .startobject()
            .field("firstname", "alice")
            .field("latname", "smith")
            .endobject()
        )
        .execute()
        .actionget();

    system.out.println("--------------------------------response");
    system.out.println(response.tostring());

    // on shutdown
//    node.close();
//    nodeclient.close();
    transportclient.close();

  }

总结

4、spring-data-elasticsearch对es有很好的支持，但很多高版本在spring-boot中不是很友好。所以，除了spring-boot自动配置的方法，最好掌握代码动态配置方法。

5、为了操作的便利性,我们往往需要动态索引，因为同一个索引（固定）是无法满足集群中多业务的。所以后续封装一个esutil类作为基本操作公交类

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。