[大数据]（一）springboot集成ES

springboot集成ES

1. 集成之前需要注意ES版本和Springboot的版本对应，这个可以再springboot官网查看。此处我选择的es版本为：7.10.1 springboot版本2.3.12.RELEASE。
2. 引入pom.xml

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

3. 使用RestHighLevelClient，代码如下：

    @Bean
    RestHighLevelClient restHighLevelClient() {
        List<EsConfigProperty.Cluster> clusters = esConfigProperty.getClusters();
        if(CollectionUtils.isEmpty(clusters)){
            return null;
       }
       EsConfigProperty.Cluster cluster = clusters.stream().findFirst().get();
       String host = hostPortName.substring(0, hostPortName.lastIndexOf(":"));
        int port = Integer.parseInt(hostPortName.substring(hostPortName.lastIndexOf(":") + 1));
        // 初始化 RestClient, hostName 和 port 填写集群的内网 VIP 地址与端口
       RestClientBuilder builder = RestClient.builder(new HttpHost(host, port, scheme))
                    .setRequestConfigCallback(builder1 -> {
                       builder1.setConnectTimeout(cluster.getConnTimeout());
                        builder1.setSocketTimeout(cluster.getSocketTimeout());
                       builder1.setConnectionRequestTimeout(cluster.getConnectionRequestTimeout());
                        return builder1;
                   });
       //保活策略
        builder.setHttpClientConfigCallback(httpClientBuilder -> httpClientBuilder
                .setDefaultIOReactorConfig(IOReactorConfig.custom()
                        .setSoKeepAlive(true)
                        .build()));
         // 设置认证信息
        if(StringUtils.isNotEmpty(cluster.getUsername())&&StringUtils.isNotEmpty(cluster.getPassword())) {
            //如果没配置密码就可以不用下面这两部
            final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
           credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(cluster.getUsername(), cluster.getPassword()));
           builder.setHttpClientConfigCallback(httpAsyncClientBuilder -> {
                httpAsyncClientBuilder.disableAuthCaching();
                return httpAsyncClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
           });
        }
        RestHighLevelClient client = new RestHighLevelClient(builder);
        return client;
   }
    @Bean
    public ElasticsearchOperations elasticsearchTemplate() {
        return new ElasticsearchRestTemplate(restHighLevelClient());
    }

4. 注意事项：RestHighLevelClient 长时间没有流量访问，隔段时间访问可能出现socket超时或者8,000milliseconds timeout on connection http-outgoing-1[ACTIVE]等错误。解决方案就是在集成RestHighLevelClient 是设置保活策略，即上面的代码的的：

同时客户端的探活时间修改为：net.ipv4.tcp_keepalive_time = 1800，可以参考：https://www.cnblogs.com/kevingrace/p/6656095.html 修改参数。

1. 新增商品索引类：
   其中Document注解中indexName表示索引名称 shards表示索引数量，其他注解再下篇文章中会介绍

@Data
@AllArgsConstructor
@NoArgsConstructor
@Builder
@Document(indexName = EsConst.PRODUCT_INDEX_NAME,shards = 5)
public class ProductIndex implements Serializable {
    /**
     * 商品ID
     */
    @Id
    private Long skuId;
    /**
     * sku状态 0删除  1正常
     */
    @Field(type=FieldType.Integer)
    private Integer skuState;

    /**
     * 商品标题
     */
    @MultiField(mainField = @Field(type=FieldType.Text,analyzer = "ik_max_word",searchAnalyzer = "ik_max_word"),
    otherFields = @InnerField(type=FieldType.Text,suffix = "pinyin",analyzer = "pinyin"))
    private String skuTitle;
}

6. 使用ElasticsearchOperations 进行索引的操作，此处是批量新增或修改，索引搜索比较复杂，下篇文章单独讲。

 @Autowired
private ElasticsearchOperations elasticsearchTemplate
/**
  * 批量新增或修改索引  存在就修改，不存在就新增
  **/
 public void batchUpdateProductIndex(List<ProductIndex> productIndexList){
       try{
           if(CollectionUtil.isNotEmpty(productIndexList)){
               if(!elasticsearchTemplate.indexOps(ProductIndex.class).exists()){
                   IndexOperations indexOps = elasticsearchTemplate.indexOps(ProductIndex.class);
                   indexOps.create();
                   Document document= indexOps.createMapping(ProductIndex.class);
                   indexOps.putMapping(document);
               }
               elasticsearchTemplate.save(productIndexList);
           }
       }catch (Exception e){
           log.error("batchUpdateProductIndex error {}",e);
       }
    }