springboot 整合 elasticsearch

springboot 整合 elasticsearch

pom.xml 引入

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.0.6.RELEASE</version>
    <relativePath/> <!-- lookup parent from repository -->
</parent>

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>


<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-elasticsearch</artifactId>
</dependency>

elasticsearch版本

當前使用的 爲 6.6.1 , elasticsearch 7.7.1 版本報錯
Received handshake message from unsupported version: [5.0.0] minimal compatible version is: [6.8.0]

yml文件引入

spring:
  data:
    elasticsearch:
      cluster-nodes: 127.0.0.1:9300
      cluster-name: my-application ## 表示 集羣名字，在es的elasticsearch.yml中查看
      repositories:
        enabled: true

定義對象信息

//dist_account 索引名
// data 類型
@Data
@Document(indexName = "dist_account", type = "data")
public class AccountEs {
    /**
     *
     */
    @Id
    private Integer id;
    private Integer type;
    private String coinCode;
    private Date createTime;
    private Date updateTime;
    private String userCode;
    private String username;
    private String feeAmount;
}

定義接口信息

public interface AccountRepository extends ElasticsearchRepository<AccountEs, Integer> {

}

定義掃描es接口信息

@Configuration
@EnableElasticsearchRepositories(basePackages = "my.elasticsearch")
public class ElasticsearchConfig {
}

執行查詢操做

@Autowired
private AccountRepository accountRepository;

/**
 * 執行普通的es查詢
 */
public void queryEs(String userCode, String coinCode, List<Integer> typeList, Date startTime, Date endTime){
    
    BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
    //wildcardQuery 表示模糊匹配 至關於 like
    boolQueryBuilder.must(QueryBuilders.wildcardQuery("userCode", String.format("%s%s%s","*",userCode, "*")));
    //termQuery 表示徹底匹配 至關於 =
    boolQueryBuilder.must(QueryBuilders.termQuery("coinCode",coinCode));
    //termsQuery 表示徹底匹配，至關於in
    boolQueryBuilder.must(QueryBuilders.termsQuery("type", typeList));
    
    //當前 根據時間範圍查找
    RangeQueryBuilder rangeQueryBuilder = QueryBuilders.rangeQuery("createTime");
    if (startTime != null){
        boolQueryBuilder.must(rangeQueryBuilder.gte(startTime.getTime()));
    }
    
    if (endTime != null){
        boolQueryBuilder.must(rangeQueryBuilder.lte(endTime.getTime()));
    }
    
    NativeSearchQueryBuilder nativeSearchQueryBuilder = new NativeSearchQueryBuilder();
    nativeSearchQueryBuilder.withQuery(boolQueryBuilder);
    
    //排序 按照id進行倒敘排序
    SortBuilder sortBuilder = SortBuilders.fieldSort("id").order(SortOrder.DESC);
    nativeSearchQueryBuilder.withSort(sortBuilder);
    
    //分頁 es 是從0開始計算分頁
    PageRequest pageRequest = PageRequest.of(0, 10);
    nativeSearchQueryBuilder.withPageable(pageRequest);
    
    NativeSearchQuery nativeSearchQuery = nativeSearchQueryBuilder.build();
    AggregatedPage<AccountEs> resultPage = (AggregatedPage<AccountEs>) accountRepository.search(nativeSearchQuery);
    
    //獲取列表
    resultPage.getContent()
    //獲取總數
    resultPage.getTotalElements()
}

es 聚合查詢

BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
... 參數 分頁 排序 與普通查詢一致

//對字段 feeAmount 進行sum操做  sum_feeAmount 表示查詢參數信息
AbstractAggregationBuilder sum = AggregationBuilders.sum("sum_feeAmount").field("feeAmount");
nativeSearchQueryBuilder.addAggregation(sum);

... 獲取列表和總數與普通查詢一致

InternalSum internalSum = (InternalSum)resultPage.getAggregation("sum_feeAmount");
//表示進行sum操做的結果
double result = internalSum.getValue();

es 聚合分組查詢

//在聚合的基礎上添加分組 當前按照 coinCode 字段進行分組,統計feeAmount 字段的總數

BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
... 參數 分頁 排序 與普通查詢一致

AbstractAggregationBuilder sum = AggregationBuilders.sum("sum_feeAmount").field("feeAmount");
AbstractAggregationBuilder group = AggregationBuilders.terms("coinCode_group").field("coinCode").subAggregation(sum);

... 獲取列表和總數與普通查詢一致

StringTerms stringTerms = resultPage.getAggregations().get("coinCode_group");
List<StringTerms.Bucket> list = stringTerms.getBuckets();
for (StringTerms.Bucket b : list) {
    //表示獲取的是分組的名字
    String goupName = b.getKeyAsString()
    //表示的是每一個分組下 統計的 feeAmount 的總數
    InternalSum internalSum = b.getAggregations().get("sum_feeAmount");
    double result = internalSum.getValue();
}

es 根據遊標查詢

@Autowired
private ElasticsearchTemplate elasticsearchTemplate;


... 參數 分頁 排序 與普通查詢一致

//第一次查詢
//scrollId 進行10秒緩存，超過這個時間則根據 scrollId獲取不到數據
Page<AccountEs> page = elasticsearchTemplate.startScroll(10000, nativeSearchQuery, AccountEs.class)

//最終的結果列表
List<AccountEs> resultList = new ArrayList<>();
while (true){
    //獲取查詢結果
    List<AccountEs> accountEsList = page.getContent();
    
    if (!CollectionUtils.isEmpty(accountEsList)){
        resultList.addAll(accountEsList);
        String scrollId = ((ScrolledPage) page).getScrollId();
        //根據scrollId 進行查詢
        page = elasticsearchTemplate.continueScroll(scrollId, 10000, AccountEs.class)
    } else {
        break;
    }
}

錯誤信息

nested exception is org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'elasticsearchClient' defined in class path resource [org/springframework/boot/autoconfigure/data/elasticsearch/ElasticsearchAutoConfiguration.class]: Bean instantiation via factory method failed; nested exception is org.springframework.beans.BeanInstantiationException: Failed to instantiate [org.elasticsearch.client.transport.TransportClient]: Factory method 'elasticsearchClient' threw exception; nested exception is java.lang.IllegalStateException: availableProcessors is already set to [12], rejecting [12]

解決方式 在main方法中設置下環境變量

System.setProperty("es.set.netty.runtime.available.processors", "false");