介紹

記錄將elasticsearch集成到spring boot的過程，以及一些簡單的應(yīng)用和helper類使用纵苛。

接入方式

使用spring-boot中的spring-data-elasticsearch,可以使用兩種內(nèi)置客戶端接入

• 節(jié)點客戶端(node client)：
  配置文件中設(shè)置為local:false,節(jié)點客戶端以無數(shù)據(jù)節(jié)點(node-master或node-client)身份加入集群想幻，換言之，它自己不存儲任何數(shù)據(jù)，但是它知道數(shù)據(jù)在集群中的具體位置，并且能夠直接轉(zhuǎn)發(fā)請求到對應(yīng)的節(jié)點上惰匙。
• 傳輸客戶端(Transport client)：
  配置文件中設(shè)置為local:true,這個更輕量的傳輸客戶端能夠發(fā)送請求到遠程集群。它自己不加入集群铃将，只是簡單轉(zhuǎn)發(fā)請求給集群中的節(jié)點徽曲。

兩個Java客戶端都通過9300端口與集群交互，使用Elasticsearch傳輸協(xié)議(Elasticsearch Transport Protocol)麸塞。集群中的節(jié)點之間也通過9300端口進行通信。如果此端口未開放涧衙，你的節(jié)點將不能組成集群哪工。

環(huán)境

版本兼容

請一定注意版本兼容問題。這關(guān)系到很多maven依賴弧哎。Spring Data Elasticsearch Spring Boot version matrix

版本兼容列表.png

搭建環(huán)境

Spring boot: 1.4.1.RELEASE
spring-data-elasticsearch: 用了最基礎(chǔ)的spring-boot-starter-data-elasticsearch雁比，選擇高版本時需要對于提高es服務(wù)版本
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
      # 連接節(jié)點,注意在集群中通信都是9300端口，否則會報錯無法連接上撤嫩！
      cluster-nodes: localhost:9300,119.29.38.169:9300
      # 是否本地連接
      local: false
      repositories:
        # 倉庫中數(shù)據(jù)存儲
        enabled: true

調(diào)試

啟動

啟動項目偎捎，日志出現(xiàn)以下說明代表成功。并且沒有報錯序攘。

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

知識點

在Elasticsearch中茴她，文檔歸屬于一種類型(type),而這些類型存在于索引(index)中，我們可以畫一些簡單的對比圖來類比傳統(tǒng)關(guān)系型數(shù)據(jù)庫：
Elasticsearch集群可以包含多個索引(indices)（數(shù)據(jù)庫）程奠，每一個索引可以包含多個類型(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 業(yè)務(wù)層
 * @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查詢方法的封裝

1. 封裝數(shù)據(jù)庫基本CRUD(創(chuàng)建（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. 計數(shù)

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中字段名
  查詢關(guān)鍵字包括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的配置文件將自動注入
• 注入調(diào)用

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構(gòu)建查詢

@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進行大結(jié)果集查詢

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實例進行節(jié)點操作,可以自行封裝Util方法

@Autowired
private ElasticsearchTemplate elasticsearchTemplate;

public void searchHelper() throws IOException {

        //節(jié)點客戶端
        // 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();

    }

總結(jié)

1. spring-data-elasticsearch對es有很好的支持申尼，但很多高版本在spring-boot中不是很友好。所以垫桂，除了spring-boot自動配置的方法师幕，最好掌握代碼動態(tài)配置方法。
2. 為了操作的便利性,我們往往需要動態(tài)索引诬滩，因為同一個索引（固定）是無法滿足集群中多業(yè)務(wù)的们衙。所以后續(xù)封裝一個EsUtil類作為基本操作公交類

參考

Spring Data ElasticSearch
Spring Data Elasticsearch介紹
elasticsearch中文指南
Spring Data Elasticsearch