缘由

开发同学有个数据同步的需求，正好我这边有闲，就接过来做了，
主要是将 mysql 的几张表，同步至 elasticsearch，全量同步

数据量

测试环境：4600 条
生产环境：3000 万条

大套路

首先，es 集群环境搭建
然后，编写 java 代码，完成同步功能

代码设计套路

因为当前几张表都是基于 accountId 的关联的，而 accountId 是自增的
主要分以下几步：

1. 查出库表中最小的 id 主键
2. 查出库表中最大的 id 主键
3. 按每页 1000 条，算出总页数
4. 按批查询 mysql，一批 10 页
5. 将查到的数据转化为 es 对象，使用线程池将数据插入到 es

集群环境搭建

这里借鉴了搜索和其他业务线的开发同学搭建的方式，完成搭建，可另写帖子完成

第一版 60 秒

@Override
public void memberAccountInfoSync() {
    int pageSize = 1000;
    long channelMinId = queryChannelMinId();
    long channelMaxId = queryChannelMaxId();
    int totalCount = queryTotalCount(channelMinId, channelMaxId);
    PageTaskUtil.handlePageTaskForBizInvoke(new PageTaskUtil.PageTaskForBizInvoke<ESMemberAccountInfo>() {

        @Override
        public List queryPageData(int pageNo, int pageSize) {
            long pageStartId = getPageStartId(channelMinId, pageSize, pageNo);
            long pageEndId = getPageEndId(pageStartId, pageSize);
            List<CpsBaseAccountInfo> cpsBaseAccountInfoList = cpsBaseAccountService.listAccountIdsWithInitialValue((byte) CURR_CHANNEL.getCode(), pageStartId, pageEndId);
            List<Long> accountIdList = cpsBaseAccountInfoList.stream().map(CpsBaseAccountInfo::getAccountId).collect(Collectors.toList());
            // 性能优化，如果accountIdList为空，则进入下一次循环
            if (CollectionUtils.isEmpty(accountIdList)) {
                return new ArrayList(0);
            }
            // 捞出其他表数据
            List<ESMemberAccountInfo> esMemberAccountInfoList = new ArrayList<>();

            cpsBaseAccountInfoList.forEach(one -> {
                long accountId = one.getAccountId();
                ESMemberAccountInfo e = new ESMemberAccountInfo();

                // 合并数据到es对象

                esMemberAccountInfoList.add(e);
            });

            return esMemberAccountInfoList;
        }

        @Override
        public void pageTask(List<ESMemberAccountInfo> pageList, Object res) {
            // 如果集合为空，直接返回
            if (CollectionUtils.isEmpty(pageList)) {
                return;
            }

            CountDownLatch countDownLatch = new CountDownLatch(pageList.size());
            for (ESMemberAccountInfo esMemberAccountInfo : pageList) {
                final Long accountId = esMemberAccountInfo.getAccountId();
                esSyncThreadPoolTaskExecutor.execute(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            // 查询es中是否存在该用户
                            if (!isExistAccountInES(esMemberAccountInfo)) {
                                insertAccountInfoToES(esMemberAccountInfo);
                            }
                        } catch (Exception e) {
                            log.error("es accountInfo sync error, accountId: {}", accountId);
                        } finally {
                            countDownLatch.countDown();
                        }
                    }
                });
            }
            try {
                countDownLatch.await();
            } catch (Exception e) {
                log.error("es sync error" + e.getMessage());
            }
        }
    }, totalCount, pageSize, 10, "esSync", null);
}

第一版，因为不知道 es 有批量插入的功能，是一条一条的往 es 中插入数据的，然后
在插入操作之前还查了一下 es 中是否存在该数据，
不存在则插入，整个功能完成后，测试环境用时近 60 秒

第二版 30 秒

private void insertAccountInfoToESBatch(List<ESMemberAccountInfo> esMemberAccountInfoList) throws Exception  {
    BulkRequest bulkRequest = new BulkRequest();
    for (ESMemberAccountInfo esMemberAccountInfo: esMemberAccountInfoList) {
        UpdateRequest updateRequest = new UpdateRequest(ESIndexEnum.ES_INDEX_BASE_ACCOUNT_INFO.getIndex(), ESIndexEnum.ES_INDEX_BASE_ACCOUNT_INFO.getType(),
                esMemberAccountInfo.getAccountId().toString() + esMemberAccountInfo.getBizChannel().toString());
        updateRequest.doc(JSON.parseObject(JSON.toJSONString(esMemberAccountInfo)));
        updateRequest.docAsUpsert(true);
        bulkRequest.add(updateRequest);
    }
    client.getRhlClient().bulk(bulkRequest, RequestOptions.DEFAULT);
}

第二版，改为批量插入 es，并且移除查询请求，测试环境用时近 30 秒

第三版 6 秒

@Override
public void memberAccountInfoSync() {
    int pageSize = 1000;
    long channelMinId = queryChannelMinId();
    long channelMaxId = queryChannelMaxId();
    log.info("es全量同步任务，channelMinId is: {}, channelMaxId is: {}", channelMinId, channelMaxId);
    int totalCount = queryTotalCount(channelMinId, channelMaxId);

    int pageCount = totalCount % pageSize == 0 ? totalCount / pageSize : totalCount / pageSize + 1;
    log.info("es全量同步任务，pageCount is: {}", pageCount);

    // druid数据库连接池的大小
    final int dbConnection = 10;
    final int repeatTimes = pageCount % dbConnection == 0 ? pageCount / dbConnection : pageCount / dbConnection + 1;
    log.info("es全量同步任务，repeatTimes is: {}", repeatTimes);

    for (int i=1; i<repeatTimes+1; i++) {

        CountDownLatch countDownLatch = new CountDownLatch(dbConnection);
        for (int j=1; j<dbConnection+1; j++) {
            log.info("es全量同步任务，channelMinId is: {}", channelMinId);
            Runnable r = syncTask(channelMinId, pageSize, j + (i-1)*dbConnection, countDownLatch);
            esSyncThreadPoolTaskExecutor.execute(r);
        }

        try {
            countDownLatch.await();
        } catch (Exception e) {
            log.error("es全量同步任务 countDownLatch.await() error" + e.getMessage());
        }
    }
}

在第二版的基础上，对查询数据的动作也进行异步批量处理，所以整个不使用开发提供的帮助类了，自己写，但保留了批量查询处理的思想，测试环境用时近 6 秒

第四版 3 秒

@Override
public void memberAccountInfoSync() {
    int pageSize = 2000;
    long channelMinId = queryChannelMinId();
    long channelMaxId = queryChannelMaxId();
    int totalCount = queryTotalCount(channelMinId, channelMaxId);

    int pageCount = totalCount % pageSize == 0 ? totalCount / pageSize : totalCount / pageSize + 1;
    log.info("es全量同步任务，pageCount: {}", pageCount);

    CountDownLatch c = new CountDownLatch(pageCount);
    // druid数据库连接池的大小，不要超过最大值40
    esSyncThreadPoolTaskExecutor.setCorePoolSize(30);
    esSyncThreadPoolTaskExecutor.setMaxPoolSize(30);

    for (int i=0; i<pageCount; i++) {
        try {
            esSyncThreadPoolTaskExecutor.execute(syncTask(CURR_CHANNEL, channelMinId, channelMaxId, pageSize, i, c));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    try {
        c.await();
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}

因为第三版仍然使用了批量处理的思想，在同一批任务中，是以最后一个任务跑完了时间为准的，所以会导致一些性能浪费，
在这里直接通过线程池本身提供的队列功能，通过参数控制线程池大小，避免将数据库连接池耗尽
这里需要注意的是，任务数最好要小于线程池的队列大小，避免生产者过快，将线程池队列塞满导致异常
同时，每页数量不能过大，这可能导致在写 es 时，操作时间过长，导致超过 es 客户端默认的最大连接时间

其他补充

实际在上到预发环境的时候，测试了下，写 3000 万的数据，因为网络的关系，导致需要 15 个小时才能跑完；
优化方法是将应用与 es 布在同一个局域网里面，结果是 17 分钟完成

总结

主要使用批量操作，线程池异步操作完成了单机优化

问题

1. 假如要再一步提升，使用分布式任务，如何处理