ES 客户端 RestHighLevelClient Connection reset by peer 亲测有效 2022-11-05( 二 ) _生活百科

优化&解决因为这个项目是助农app（抖音+淘宝，杂交版app），app首页下拉获取最新推荐视频，需要通过第三方极光拉取推荐视频，存入本地数据库mysql后；一次拉取20条，然后通过线程池异步写入ES，大概架构如下

文章插图
初步判定，可能是app随着用户体量上去，线程池异步逐条写入ES，造成ES大量的写请求，然后将架构改为如图

文章插图
就这样以为问题就解决了呢，第二天去服务器里拉下日志，发现之前的问题还是存在，就上百度， github ， ES官网，很多人遇到相同问题，各种解决方案，对我们这套架构来说，都不管用。大概意思是ES客户端如果没有设置KeepAlive的话，默认为-1就是永不过期，然后就设置ES客户端的KeepAlive时间，问题还是复现，这边最终做了如下修改，从此ES客户端连接关闭问题就没有复现过
我们是通过Nginx反向代理，构建一个ES集群，架构如下

文章插图
问题修复步骤一修改客户端tcp keepalive，系统默认为7200，修改为300（单位秒， 5分钟）

文章插图
修改为300

文章插图
[root@ybchen-1 ~]# cat /proc/sys/net/ipv4/tcp_keepalive_time7200[root@ybchen-1 ~]#[root@ybchen-1 ~]#[root@ybchen-1 ~]#[root@ybchen-1 ~]# cat /proc/sys/net/ipv4/tcp_keepalive_time7200[root@ybchen-1 ~]# echo 300 > /proc/sys/net/ipv4/tcp_keepalive_time[root@ybchen-1 ~]# cat /proc/sys/net/ipv4/tcp_keepalive_time300[root@ybchen-1 ~]#[root@ybchen-1 ~]#[root@ybchen-1 ~]#[root@ybchen-1 ~]# 步骤二修改ES客户端的KeepAlive时间，并添加线程池大小等参数
=========ES配置类-开始=============@Data@Component@ConfigurationProperties(prefix = "elastic.search")public class EsConfiguration {//主机private String host;//端口private int port;//集群名称private String clusterName;//访问协议，如:httpprivate String schema;//用户名private String userName;//密码private String password;}=========ES配置类-结束=========================ES客户端-开始=======================@AutowiredEsConfiguration esConfiguration;private static RestHighLevelClient restHighLevelClient;public RestHighLevelClient getRestClient() {if (null != restHighLevelClient) {return restHighLevelClient;}List<HttpHost> httpHosts = new ArrayList<>();//填充数据httpHosts.add(new HttpHost(esConfiguration.getHost(), esConfiguration.getPort()));//填充host节点RestClientBuilder builder = RestClient.builder(httpHosts.toArray(new HttpHost[0]));if (StringUtils.isNotBlank(esConfiguration.getUserName()) && StringUtils.isNotBlank(esConfiguration.getPassword())) {//填充用户名密码final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(esConfiguration.getUserName(), esConfiguration.getPassword()));//异步链接延时配置builder.setRequestConfigCallback(requestConfigBuilder ->requestConfigBuilder.setConnectTimeout(5000) //5秒.setSocketTimeout(5000).setConnectionRequestTimeout(5000));//异步链接数配置builder.setHttpClientConfigCallback(httpClientBuilder -> {//最大连接数100个httpClientBuilder.setMaxConnTotal(100);//最大路由连接数httpClientBuilder.setMaxConnPerRoute(100);httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);// 设置KeepAlive为5分钟的时间，不设置默认为-1，也就是持续连接，然而这会受到外界的影响比如Firewall ，会将TCP连接单方面断开，从而会导致Connection reset by peer的报错// 参考github解决方案：https://github.com/TFdream/Elasticsearch-learning/issues/30httpClientBuilder.setKeepAliveStrategy((response, context) -> TimeUnit.MINUTES.toMillis(3)).setDefaultIOReactorConfig(IOReactorConfig.custom().setIoThreadCount(1).setSoKeepAlive(true).build());return httpClientBuilder;});}restHighLevelClient = new RestHighLevelClient(builder);return restHighLevelClient;}==============ES客户端-结束========================== ES客户端使用示例 ===============@Data@ApiModel("新增/更新ES通用数据结构")public class DocSearchVo implements Serializable {@ApiModelProperty(value = "https://www.huyubaike.com/biancheng/文档ID")private String docId;@ApiModelProperty(value = "https://www.huyubaike.com/biancheng/文档JSON")private String docStrJson;}/*** 批量添加文档** @param indexName* @param docSearchVoList* @return*/public boolean batchIndexDoc(String indexName, List<DocSearchVo> docSearchVoList) {RestHighLevelClient client = getRestClient();BulkRequest request = new BulkRequest();for (DocSearchVo docSearchVo : docSearchVoList) {IndexRequest indexRequest = new IndexRequest(indexName).id(docSearchVo.getDocId()).source(docSearchVo.getDocStrJson(), XContentType.JSON);request.add(indexRequest);}client.bulkAsync(request, RequestOptions.DEFAULT, new ActionListener<BulkResponse>() {@Overridepublic void onResponse(BulkResponse bulkItemResponses) {log.debug("异步批量添加文档成功，indexName：{}，docSearchVoList：{}", indexName, docSearchVoList);}@Overridepublic void onFailure(Exception e) {log.error("异步批量添加文档失败，indexName：{} ， docSearchVoList：{}，错误信息：{}", indexName, docSearchVoList, e);}});return true;}