介绍
记录将elasticsearch集成到spring boot的过程，以及一些简单的应用和helper类使用。
接入方式
使用spring-boot中的spring-data-elasticsearch,可以使用两种内置客户端接入
- 节点客户端(node client)：
配置文件中设置为local:false,节点客户端以无数据节点(node-master或node-client)身份加入集群，换言之，它自己不存储任何数据，但是它知道数据在集群中的具体位置，并且能够直接转发请求到对应的节点上。 - 传输客户端(Transport client)：
配置文件中设置为local:true,这个更轻量的传输客户端能够发送请求到远程集群。它自己不加入集群，只是简单转发请求给集群中的节点。
两个Java客户端都通过9300端口与集群交互，使用Elasticsearch传输协议(Elasticsearch Transport Protocol)。集群中的节点之间也通过9300端口进行通信。如果此端口未开放，你的节点将不能组成集群。
环境
版本兼容
请一定注意版本兼容问题。这关系到很多maven依赖。Spring Data Elasticsearch Spring Boot version matrix
版本兼容列表.png
搭建环境
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
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* 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
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* 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
* @Author wanwt@senthink.com
* @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 业务层
* @Author wanwt@senthink.com
* @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查询方法的封装
- 封装数据库基本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.
}
- 分页排序查询
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));
- 计数
public interface UserRepository extends CrudRepository<User, Long> {
Long countByLastname(String lastname);
}
- 删除
public interface UserRepository extends CrudRepository<User, Long> {
Long deleteByLastname(String lastname);
List<User> removeByLastname(String lastname);
}
- 自定义查询方法自动注入
- 声明一个接口继承
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");
}
}
- 支持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(…);
}
- 支持异步查询
@Async
Future<User> findByFirstname(String firstname);
@Async
CompletableFuture<User> findOneByFirstname(String firstname);
@Async
ListenableFuture<User> findOneByLastname(String lastname);
支持原生es JavaAPI
- 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);
- 利用
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;
}
}
}
- 获取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();
}
总结
-
spring-data-elasticsearch对es有很好的支持，但很多高版本在spring-boot中不是很友好。所以，除了spring-boot自动配置的方法，最好掌握代码动态配置方法。 - 为了操作的便利性,我们往往需要动态索引，因为同一个索引（固定）是无法满足集群中多业务的。所以后续封装一个
EsUtil类作为基本操作公交类
参考
Spring Data ElasticSearch
Spring Data Elasticsearch介绍
elasticsearch中文指南
Spring Data Elasticsearch
