缘由

开发同学有个数据同步的需求,正好我这边有闲,就接过来做了,
主要是将 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. 假如要再一步提升,使用分布式任务,如何处理


↙↙↙阅读原文可查看相关链接,并与作者交流