ES 批量插入数据 bulk

深碍√TFBOYSˉ_ 2022-09-04 11:48 338阅读 0赞

# 引入依赖 #

<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);
        }