ES 批量插入数据 bulk

Posted 2021-09-05 AresCarry

引入依赖

 <dependency>
            <groupId>org.elasticsearch.client</groupId>
            <artifactId>elasticsearch-rest-high-level-client</artifactId>
            <version>7.4.0</version>
            <exclusions>
                <exclusion>
                    <groupId>org.elasticsearch</groupId>
                    <artifactId>elasticsearch</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.elasticsearch.client</groupId>
                    <artifactId>elasticsearch-rest-client</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.elasticsearch.client</groupId>
            <artifactId>elasticsearch-rest-client</artifactId>
            <version>7.4.0</version>
        </dependency>
        <dependency>
            <groupId>org.elasticsearch</groupId>
            <artifactId>elasticsearch</artifactId>
            <version>7.4.0</version>
        </dependency>

ES客户端配置类

package com.ares.es;

import com.ctrip.framework.apollo.ConfigService;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.elasticsearch.client.HttpAsyncResponseConsumerFactory;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class ElasticsearchConfig {

    public static final RequestOptions COMMON_OPTIONS;

    String key = "ELASTICSEARCH";

    private String username = "x'x'x'x";
    private String passwrod = "x'x'x'x'x'x";
    private String hostName= "es-cn-x'x'x'x'x.elasticsearch.aliyuncs.com";
    private Integer hostport= 9200;

    static {
        RequestOptions.Builder builder = RequestOptions.DEFAULT.toBuilder();

        // 默认缓存限制为100MB，此处修改为30MB。
        builder.setHttpAsyncResponseConsumerFactory(
                new HttpAsyncResponseConsumerFactory
                        .HeapBufferedResponseConsumerFactory(30 * 1024 * 1024));
        COMMON_OPTIONS = builder.build();
    }

    @Bean("esClient")
    public RestHighLevelClient createRestHighLevelClient() {
        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, passwrod));
        // 通过builder创建rest client，配置http client的HttpClientConfigCallback。
        // 单击所创建的Elasticsearch实例ID，在基本信息页面获取公网地址，即为ES集群地址。
        RestClientBuilder builder = RestClient.builder(new HttpHost(hostName, hostport))
                .setHttpClientConfigCallback(httpClientBuilder ->
                        httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider));

        //防止 java.net.SocketTimeoutException: 30,000 milliseconds timeout on connection http-outgoing-3
        RestClientBuilder.RequestConfigCallback configCallback = new RestClientBuilder.RequestConfigCallback() {
            @Override
            public org.apache.http.client.config.RequestConfig.Builder customizeRequestConfig(org.apache.http.client.config.RequestConfig.Builder requestConfigBuilder) {
                return requestConfigBuilder
                        .setConnectTimeout(5000*1000)
                        .setSocketTimeout(6000 * 1000);//更改客户端的超时限制默认30秒现在改为6000s
            }
        };
        builder.setRequestConfigCallback(configCallback);

        // RestHighLevelClient实例通过REST low-level client builder进行构造。
        RestHighLevelClient highClient = new RestHighLevelClient(builder);
        return highClient;
    }
}

批量插入

  @Autowired
    private RestHighLevelClient esClient;
    
    public void bulkPutIndex(List<Map<String, Object>> list) throws IOException {
        if (CollectionUtil.isEmpty(list)) {
            return;
        }
        String index = "media_play_time_test";
        String type = "_doc";
        int size = list.size();
        BulkRequest request = new BulkRequest();
        for (int i = 0; i < size; i++) {
            Map<String, Object> map = list.get(i);
            request.add(new IndexRequest(index, type).source(map, XContentType.JSON));
        }
        esClient.bulk(request, RequestOptions.DEFAULT);
    }