当前位置: 首页 > 面试题库 >

使用Rest High Level Client检索数据或将数据插入Elastic Search时发生SocketTimeoutException

赫连宏伯
2023-03-14
问题内容

我在SocketTimeoutException从Elastic检索数据/从Elastic插入数据时面对。周围有这种情况10-30 request/second。这些请求是get / put的组合。

这是我的弹性配置:

  • 3 master nodes 每个 4GB RAM
  • 2 data nodes 每个 8GM RAM
  • 连接到以上数据节点的Azure负载平衡器(似乎仅在其中打开了9200端口)。而且Java客户端连接到此负载均衡器,因为它只是公开的。
  • 弹性版:7.2.0
  • 休息高级客户:
        <dependency>
        <groupId>org.elasticsearch.client</groupId>
        <artifactId>elasticsearch-rest-high-level-client</artifactId>
        <version>7.2.0</version>
    </dependency>

    <dependency>
        <groupId>org.elasticsearch</groupId>
        <artifactId>elasticsearch</artifactId>
        <version>7.2.0</version>
    </dependency>

索引信息:

  • 索引分片:2
  • 索引副本:1
  • 索引总栏位:10000
  • 来自kibana的索引大小:总计27.2 MB-和Primaries: 12.2MB
  • 索引结构:
        {
      "dev-index": {
        "mappings": {
          "properties": {
            "dataObj": {
              "type": "object",
              "enabled": false
            },
            "generatedID": {
              "type": "keyword"
            },
            "transNames": { //it's array of string
              "type": "keyword"
            }
          }
        }
      }
    }
  • 动态映射已禁用。

以下是我的elastic Config文件。在这里,我有两个连接bean,一个用于读取,另一个用于写入弹性。

ElasticConfig.java:

    @Configuration
    public class ElasticConfig {

        @Value("${elastic.host}")
        private String elasticHost;

        @Value("${elastic.port}")
        private int elasticPort;

        @Value("${elastic.user}")
        private String elasticUser;

        @Value("${elastic.pass}")
        private String elasticPass;

        @Value("${elastic-timeout:20}")
        private int timeout;

        @Bean(destroyMethod = "close")
        @Qualifier("readClient")
        public RestHighLevelClient readClient(){

            final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
            credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(elasticUser, elasticPass));

            RestClientBuilder builder = RestClient
                    .builder(new HttpHost(elasticHost, elasticPort))
                    .setHttpClientConfigCallback(httpClientBuilder -> 
                            httpClientBuilder
                                    .setDefaultCredentialsProvider(credentialsProvider)
                                    .setDefaultIOReactorConfig(IOReactorConfig.custom().setIoThreadCount(5).build())
                    );

            builder.setRequestConfigCallback(requestConfigBuilder -> 
                    requestConfigBuilder
                            .setConnectTimeout(10000)
                            .setSocketTimeout(60000)
                            .setConnectionRequestTimeout(0)
            );

            RestHighLevelClient restClient = new RestHighLevelClient(builder);
            return restClient;
        }

        @Bean(destroyMethod = "close")
        @Qualifier("writeClient")
        public RestHighLevelClient writeClient(){

            final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
            credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(elasticUser, elasticPass));

            RestClientBuilder builder = RestClient
                    .builder(new HttpHost(elasticHost, elasticPort))
                    .setHttpClientConfigCallback(httpClientBuilder -> 
                            httpClientBuilder
                                    .setDefaultCredentialsProvider(credentialsProvider)
                                    .setDefaultIOReactorConfig(IOReactorConfig.custom().setIoThreadCount(5).build())
                    );

            builder.setRequestConfigCallback(requestConfigBuilder -> 
                    requestConfigBuilder
                            .setConnectTimeout(10000)
                            .setSocketTimeout(60000)
                            .setConnectionRequestTimeout(0)
            );

            RestHighLevelClient restClient = new RestHighLevelClient(builder);
            return restClient;
        }

    }

这是调用弹性函数的函数,如果弹性数据可用,它将接受它,否则将生成数据并放入弹性。

    public Object getData(Request request) {

        DataObj elasticResult = elasticService.getData(request);
        if(elasticResult!=null){
            return elasticResult;
        }
        else{
            //code to generate data
            DataObj generatedData = getData();//some function which will generated data
            //put above data into elastic by Async call.
            elasticAsync.putData(generatedData);
            return generatedData;
        }
    }

ElasticService.java getData函数:

    @Service
    public class ElasticService {

        @Value("${elastic.index}")
        private String elasticIndex;

        @Autowired
        @Qualifier("readClient")
        private RestHighLevelClient readClient;

        public DataObj getData(Request request){
            String generatedId = request.getGeneratedID();

            GetRequest getRequest = new GetRequest()
                    .index(elasticIndex)   //elastic index name
                    .id(generatedId);   //retrieving by index id from elastic _id field (as key-value)

            DataObj result = null;
            try {
                GetResponse response = readClient.get(getRequest, RequestOptions.DEFAULT);
                if(response.isExists()) {
                    ObjectMapper objectMapper = new ObjectMapper();
                    result = objectMapper.readValue(response.getSourceAsString(), DataObj.class);
                }
            }  catch (Exception e) {
                LOGGER.error("Exception occurred during  fetch from elastic !!!! " + ,e);
            }
            return result;
        }

    }

ElasticAsync.java异步放置数据功能:

    @Service
    public class ElasticAsync {

        private static final Logger LOGGER = Logger.getLogger(ElasticAsync.class.getName());

        @Value("${elastic.index}")
        private String elasticIndex;

        @Autowired
        @Qualifier("writeClient")
        private RestHighLevelClient writeClient;

        @Async
        public void putData(DataObj generatedData){
         ElasticVO updatedRequest = toElasticVO(generatedData);//ElasticVO matches to the structure of index given above.

            try {
                ObjectMapper objectMapper = new ObjectMapper();
                String jsonString = objectMapper.writeValueAsString(updatedRequest);

                IndexRequest request = new IndexRequest(elasticIndex);
                request.id(generatedData.getGeneratedID());
                request.source(jsonString, XContentType.JSON);
                request.setRefreshPolicy(WriteRequest.RefreshPolicy.NONE);
                request.timeout(TimeValue.timeValueSeconds(5));
                IndexResponse indexResponse = writeClient.index(request, RequestOptions.DEFAULT);
                LOGGER.info("response id: " + indexResponse.getId());

                }

            } catch (Exception e) {
                LOGGER.error("Exception occurred during saving into elastic !!!!",e);
            }


        }

    }

这是将数据保存到弹性模块时发生异常时堆栈跟踪的一部分:

    2019-07-19 07:32:19.997 ERROR [data-retrieval,341e6ecc5b10f3be,1eeb0722983062b2,true] 1 --- [askExecutor-894] a.c.s.a.service.impl.ElasticAsync        : Exception occurred during saving into elastic !!!!

    java.net.SocketTimeoutException: 60,000 milliseconds timeout on connection http-outgoing-34 [ACTIVE]
    at org.elasticsearch.client.RestClient.extractAndWrapCause(RestClient.java:789) ~[elasticsearch-rest-client-7.2.0.jar!/:7.2.0]
        at org.elasticsearch.client.RestClient.performRequest(RestClient.java:225) ~[elasticsearch-rest-client-7.2.0.jar!/:7.2.0]
        at org.elasticsearch.client.RestClient.performRequest(RestClient.java:212) ~[elasticsearch-rest-client-7.2.0.jar!/:7.2.0]
        at org.elasticsearch.client.RestHighLevelClient.internalPerformRequest(RestHighLevelClient.java:1448) ~[elasticsearch-rest-high-level-client-7.2.0.jar!/:7.2.0]
        at org.elasticsearch.client.RestHighLevelClient.performRequest(RestHighLevelClient.java:1418) ~[elasticsearch-rest-high-level-client-7.2.0.jar!/:7.2.0]
        at org.elasticsearch.client.RestHighLevelClient.performRequestAndParseEntity(RestHighLevelClient.java:1388) ~[elasticsearch-rest-high-level-client-7.2.0.jar!/:7.2.0]
        at org.elasticsearch.client.RestHighLevelClient.index(RestHighLevelClient.java:836) ~[elasticsearch-rest-high-level-client-7.2.0.jar!/:7.2.0]


    Caused by: java.net.SocketTimeoutException: 60,000 milliseconds timeout on connection http-outgoing-34 [ACTIVE]
        at org.apache.http.nio.protocol.HttpAsyncRequestExecutor.timeout(HttpAsyncRequestExecutor.java:387) ~[httpcore-nio-4.4.11.jar!/:4.4.11]
        at org.apache.http.impl.nio.client.InternalIODispatch.onTimeout(InternalIODispatch.java:92) ~[httpasyncclient-4.1.3.jar!/:4.1.3]
        at org.apache.http.impl.nio.client.InternalIODispatch.onTimeout(InternalIODispatch.java:39) ~[httpasyncclient-4.1.3.jar!/:4.1.3]
        at org.apache.http.impl.nio.reactor.AbstractIODispatch.timeout(AbstractIODispatch.java:175) ~[httpcore-nio-4.4.11.jar!/:4.4.11]
        at org.apache.http.impl.nio.reactor.BaseIOReactor.sessionTimedOut(BaseIOReactor.java:263) ~[httpcore-nio-4.4.11.jar!/:4.4.11]
        at org.apache.http.impl.nio.reactor.AbstractIOReactor.timeoutCheck(AbstractIOReactor.java:492) ~[httpcore-nio-4.4.11.jar!/:4.4.11]
        at org.apache.http.impl.nio.reactor.BaseIOReactor.validate(BaseIOReactor.java:213) ~[httpcore-nio-4.4.11.jar!/:4.4.11]
        at org.apache.http.impl.nio.reactor.AbstractIOReactor.execute(AbstractIOReactor.java:280) ~[httpcore-nio-4.4.11.jar!/:4.4.11]
        at org.apache.http.impl.nio.reactor.BaseIOReactor.execute(BaseIOReactor.java:104) ~[httpcore-nio-4.4.11.jar!/:4.4.11]
        at org.apache.http.impl.nio.reactor.AbstractMultiworkerIOReactor$Worker.run(AbstractMultiworkerIOReactor.java:591) ~[httpcore-nio-4.4.11.jar!/:4.4.11]
        ... 1 common frames omitted

这是在将数据检索到弹性期间发生异常时堆栈跟踪的某些部分:

    2019-07-19 07:22:37.844 ERROR [data-retrieval,104cf6b2ab5b3349,b302d3d3cd7ebc84,true] 1 --- [o-8080-exec-346] a.c.s.a.service.impl.ElasticService      : Exception occurred during  fetch from elastic !!!! 

    java.net.SocketTimeoutException: 60,000 milliseconds timeout on connection http-outgoing-30 [ACTIVE]
        at org.elasticsearch.client.RestClient.extractAndWrapCause(RestClient.java:789) ~[elasticsearch-rest-client-7.1.1.jar!/:7.1.1]
        at org.elasticsearch.client.RestClient.performRequest(RestClient.java:225) ~[elasticsearch-rest-client-7.1.1.jar!/:7.1.1]
        at org.elasticsearch.client.RestClient.performRequest(RestClient.java:212) ~[elasticsearch-rest-client-7.1.1.jar!/:7.1.1]
        at org.elasticsearch.client.RestHighLevelClient.internalPerformRequest(RestHighLevelClient.java:1433) ~[elasticsearch-rest-high-level-client-7.1.1.jar!/:7.1.1]
        at org.elasticsearch.client.RestHighLevelClient.performRequest(RestHighLevelClient.java:1403) ~[elasticsearch-rest-high-level-client-7.1.1.jar!/:7.1.1]
        at org.elasticsearch.client.RestHighLevelClient.performRequestAndParseEntity(RestHighLevelClient.java:1373) ~[elasticsearch-rest-high-level-client-7.1.1.jar!/:7.1.1]
        at org.elasticsearch.client.RestHighLevelClient.get(RestHighLevelClient.java:699) ~[elasticsearch-rest-high-level-client-7.1.1.jar!/:7.1.1]



    Caused by: java.net.SocketTimeoutException: 60,000 milliseconds timeout on connection http-outgoing-30 [ACTIVE]
            at org.apache.http.nio.protocol.HttpAsyncRequestExecutor.timeout(HttpAsyncRequestExecutor.java:387) ~[httpcore-nio-4.4.11.jar!/:4.4.11]
        at org.apache.http.impl.nio.client.InternalIODispatch.onTimeout(InternalIODispatch.java:92) ~[httpasyncclient-4.1.3.jar!/:4.1.3]
        at org.apache.http.impl.nio.client.InternalIODispatch.onTimeout(InternalIODispatch.java:39) ~[httpasyncclient-4.1.3.jar!/:4.1.3]
        at org.apache.http.impl.nio.reactor.AbstractIODispatch.timeout(AbstractIODispatch.java:175) ~[httpcore-nio-4.4.11.jar!/:4.4.11]
        at org.apache.http.impl.nio.reactor.BaseIOReactor.sessionTimedOut(BaseIOReactor.java:263) ~[httpcore-nio-4.4.11.jar!/:4.4.11]
        at org.apache.http.impl.nio.reactor.AbstractIOReactor.timeoutCheck(AbstractIOReactor.java:492) ~[httpcore-nio-4.4.11.jar!/:4.4.11]
        at org.apache.http.impl.nio.reactor.BaseIOReactor.validate(BaseIOReactor.java:213) ~[httpcore-nio-4.4.11.jar!/:4.4.11]
        at org.apache.http.impl.nio.reactor.AbstractIOReactor.execute(AbstractIOReactor.java:280) ~[httpcore-nio-4.4.11.jar!/:4.4.11]
        at org.apache.http.impl.nio.reactor.BaseIOReactor.execute(BaseIOReactor.java:104) ~[httpcore-nio-4.4.11.jar!/:4.4.11]
        at org.apache.http.impl.nio.reactor.AbstractMultiworkerIOReactor$Worker.run(AbstractMultiworkerIOReactor.java:591) ~[httpcore-nio-4.4.11.jar!/:4.4.11]
        ... 1 common frames omitted

我浏览了一些相关的博客,他们提到这个问题可能是由于RAMElastic的cluster配置所致。然后,因为只有两个数据节点,所以我将分片从5更改为2。还将Data节点的内存从4GB增加到8GB,因为我知道弹性只会使用50%totalRAM。异常的发生率有所降低,但问题仍然存在。

解决这个问题的可能方法是什么?从Java/弹性配置的角度来看,我经常缺少这种东西SocketTimeoutException吗?如果您需要有关配置的更多详细信息,请告诉我。


问题答案:

我们遇到了同样的问题,经过一番挖掘后,我发现了根本原因:客户端和弹性服务器之间的防火墙配置不匹配,tcp的内核配置仍然有效。

防火墙在3600秒后删除空闲连接。问题是tcp keep alive的内核参数设置为7200秒(RedHat 6.x / 7.x中的默认设置):

sysctl -n net.ipv4.tcp_keepalive_time
7200

因此,在发送保持活动探测之前,连接已断开。弹性http客户端中的asyncHttpClient似乎无法很好地处理掉线的连接,它只是等待套接字超时。

因此,请检查您的客户端和服务器之间是否有任何网络设备(负载平衡器,防火墙,代理服务器等)具有会话超时或类似超时,并增加该超时或降低tcp_keep_alive内核参数



 类似资料:
  • 我不理解高级REST客户机(它取代了传输客户机)和Spring Data ElasticSearch之间的关系。 有一些教程(https://www.baeldung.com/spring-data-elasticsearch-tutorial)展示了如何使用像getSomethingById()这样的自动实现方法进行JPA风格的查询。他们利用这个接口: 所以如果我想使用最新的官方客户端,对我来说

  • 问题内容: 我是Elasticearch的新手,并且已经尝试了2天,将一些数据插入Elasticearch。我在Google上发现有很多页面可以帮助创建索引(我不清楚“ index”,换句话说是“插入”吗?)然后很多地方给出了curl命令,但我确实没有知道在哪里执行这些代码行以插入数据。例: 我正在使用Window 7,并且已安装Java并成功运行elasticsearch。任何人都可以向我展示有

  • 我正在尝试从我创建的一个简单的联系人表单中插入数据。我正在使用phpMyAdmin 下面是PHP(出于安全考虑,我删除了define语句,但我可以毫无问题地建立到数据库的链接。) 当有人点击论坛上的submit按钮后,代码就会运行,据我所知,这是正常工作的。 该错误发生在php代码的末尾。正在输出“SLQ Insert Statement FAILD”,但未发布mysql错误。 我的数据库/表的设

  • 问题内容: 我一直在关注本教程(http://blog.florian-hopf.de/2014/07/scrapy-and- elasticsearch.html )并使用这个易碎的elasticsearch管道(https://github.com/knockrentals/scrapy -elasticsearch ),并能够将数据从scrapy提取到JSON文件,并在本地主机上启动并运行E