详解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类作为基本操作公交类

来源：http://www.jianshu.com/p/35f9f867f48f?utm_source=tuicool&utm_medium=referral#