没错这又是一次血案，不过是在测试环境发现的，还好不是上次线上导致的CPU300%，这次及时发现，避免了线上血案，不过我还是要复盘一下的，下面就看看我的分析，看看有没有什么错误的地方

场景描述

场景很简单，就是把我MySQL中一个表的数据全部迁移到ElasticSearch的一个新索引上(因为老的索引字段匹配不上，随着业务迭代MySQL的字段类型发生了变化，ES又没有办法动态更新，所以最简单的方法就是同步一次数据啦) 然后我就想当然的写了一段同步脚本，然后满怀信心的去开发环境自测，没问题啊，10W多条数据很快啊，当时直接就同步完了，一条数据不差，心里暗想，这次真简单，然后就提测，去测试环境跑数据了，然后血案就发生了

跑着跑着，测试姐姐就说测试环境不能用了，我想着我只是同步一个表的数据，就算有错应该也只是影响那个页面的查询吧，然后我就打开了测试环境页面，好家伙，每个页面都报错了，都是can not found url，再结合我们的ShenYU网关，就想到应该是节点崩了，没有Controller注册到网关上，所以请求就没办法映射。然后我就去容器里面看报错嘛，但是错误显示的是超时：ElasticSearch连接超时，ZK连接超时等等等等。比较茫然，这个错误没有太大的信息呀，师兄提醒我是不是内存崩了，但是我并没有看到OOM的提醒呀，所以我就顺着这个思路开始了排查

思路验证

想到可能是OOM的原因，为了测验，我就把代码的异步批量处理改成了同步单条索引，然后去测试环境验证(为什么不去开发环境？我下面会说到) 发现如果是同步单条跑的话，一点问题没有就是速度有点慢，那不行，生产几千万呢，算下来的话，几天也同步不完，但是我从这就确定了应该是内存的问题：是内存溢出了？还是内存泄漏了？接下来，看看我经过排查过后得到的重点代码(罪魁祸首)

//拉取5000条数据
//这条SQL就是查询主键id > redis中已经同步过后的ID 的后5000条数据
//为了减轻MySQL压力
List<xxxDO> aDOS = xxxRepository.selectById(Integer.valueOf(redisService.get(idKey)));
//同步数据中间逻辑省略
//开始索引
xxxxEsService.indexEs(aDOS);


//再来看看这个indexEs方法
aDOS.forEach(aDO -> {
            EsABO esABO = new EsABO();
            //把aDO转换成esABO然后
            esHighLevelRestClient.bulkIndexAsync(Arrays.asList(esABO));
}
//罪魁祸手就是这个bulkIndexAsync接下来看看这个方法

 public void bulkIndexAsync(List<IDocument> dataList) {
        if (CollectionUtils.isEmpty(dataList)) {
            return;
        }
        BulkRequest request = new BulkRequest();
        dataList.forEach(data -> {
            IndexRequest indexRequest = new IndexRequest(indexName,
                    mappingTypeName, data.documentId())
                    .source(JSON.toJSONString(data), XContentType.JSON)
                    .routing(data.routing());
            request.add(indexRequest);
        });

        long startTime = System.currentTimeMillis();
        client.bulkAsync(request, RequestOptions.DEFAULT, new ActionListener() {

            @Override
            public void onResponse(Object o) {
                log.info("es index cost time:{}", System.currentTimeMillis() - startTime);
            }

            @Override
            public void onFailure(Exception e) {
                log.error("es index fail, cost time:{}, Exception:", System.currentTimeMillis() - startTime, e);
            }
        });
    }

解释下，这个方法：就是传过来一条记录，然后这个记录就会被add方法添加到这个BulkRequest里面，然后ESClient会执行这个bulkAsync就去异步的同步数据了。这样看起来没啥问题啊，我下面再放一下这个bulkAsync方法的源码

    public final void bulkAsync(BulkRequest bulkRequest, RequestOptions options, ActionListener<BulkResponse> listener) {
        performRequestAsyncAndParseEntity(bulkRequest, RequestConverters::bulk, options, BulkResponse::fromXContent, listener, emptySet());
    }

它里面就是对我们提交的这个bulkRequest处理的逻辑，看上去没什么问题，我再放一张图，大家看看能不能察觉到什么： 这是我本地跑Jprofiler存储到快照中的几张图 1.同步数据接口未开启的时候

2. 同步数据开启

大家发现什么问题了么？

开启同步数据后，我的堆越来越大，这些也没什么，关键是经过GC之后，堆的大小并没有明显变化，问题就在这了，也就是说我的一个大对象并没有进行GC回收，是那个对象呢？没错就是那个bulkRequest

分析

一开始我拉出来5000条数据塞给这个bulkRequest，可能刚开始他能处理过来，并且进行了GC回收直接回收掉了，看下图

这个时候我们的速度应该能跟得上，但是我们的MySQL源源不断的塞给他5000，很快这个堆就又满了，但是我们的这个bulkRequest还没有同步完，还有强引用(可能现在这个bulkRequest里面有10W数据，但是前5W已经处理完了)，也就是说随着速度跟不上来，导致这个bulkRequest越来愈大

而且以前处理过的也无法销毁，所以导致内存泄漏，进而系统内存OOM崩溃，但是为什么没有OOM，这个我只能猜测可能是速度太快，根本就没来得及然后系统就僵死了，一行日志都没有了，整个系统直接崩溃了

怎么解决？

关键原因就在于及时的回收已经处理过的request，所以采用另外一种批量处理方式

public  void bulkProcessorBatchInsert(List<EsUpdateEntityBO<T>> entityBOList) {
        if (entityBOList.size()==0) {
            return;
        }
        List<IndexRequest> indexRequests=new ArrayList<>();

        //更新的数据
        entityBOList.forEach(e->{
            //获取id
            IndexRequest indexRequest = new IndexRequest();
            indexRequest.routing(e.getRouting());
            indexRequest.index(indexName);
            indexRequest.type(mappingTypeName);
            //更新的id
            indexRequest.id(e.getDocumentId());
            //更新的数据
            String jsonString = JSON.toJSONString(e.getT());
            indexRequest.source(jsonString, XContentType.JSON);
            indexRequests.add(indexRequest);
        });
        indexRequests.forEach(bulkProcessor::add);
    }

这不是和刚才的一样么？其实不一样，因为关键代码在下面 我在建立ES客户端的时候是这样的

private BulkProcessor createBulkProcessor(RestHighLevelClient restHighLevelClient,Boolean async) {
		//中间很多逻辑
        
        BulkProcessor.Builder builder;
        if (async) {
            //异步
            //这里也有很多逻辑
        }
        //到达1000条时刷新
        builder.setBulkActions(2000);
        //内存到达8M时刷新
        builder.setBulkSize(new ByteSizeValue(8L, ByteSizeUnit.MB));
        //设置的刷新间隔1s
        builder.setFlushInterval(TimeValue.timeValueSeconds(1));
        //设置允许执行的并发请求数。
        builder.setConcurrentRequests(8);
        //设置重试策略
        builder.setBackoffPolicy(BackoffPolicy.constantBackoff(TimeValue.timeValueSeconds(1), 3));
        return builder.build();
    }

关键点就在于，我不无限制的处理请求，每当我的bucket攒够2000条，或者我的内存达到了8M，或者已经距离上次处理间隔了1S了我就主动进行刷盘（同步数据）操作，这样的话就能避免上面那种情况了

难道ES官方不知道这个问题么？我看了官方文档，确实没找到他这个 bulkAsync中这个BulkRequest bulkRequest有什么刷盘策略，难道就是无限制增长么？这个我还没有看完他的源码，可能自己的能力也不够，如果有老哥知道，评论区留言 至此用了新的异步批量请求的操作，测试环境很快很安全的同步了数据，而且预发环境的上千万的数据也很快很安全的同步完成，我心中的大石头终于放下来了 分享就到这了，如果有什么不对的地方，请大佬指出，我也好改正学习下 好了，写代码去了

作者：会编程的王学长
链接：https://juejin.cn/post/7090355674655227918