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

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

您可能感兴趣的文章:

• Spring Boot集成ElasticSearch实现搜索引擎的示例
• SpringBoot整合ElasticSearch实践
• Spring Boot整合Elasticsearch实现全文搜索引擎案例解析
• SpringBoot整合ElasticSearch的示例代码
• SpringBoot整合Elasticsearch并实现CRUD操作
• Spring Boot与Kotlin 整合全文搜索引擎Elasticsearch的示例代码
• Spring Boot整合ElasticSearch实现多版本兼容的方法详解