快捷導(dǎo)航

Java多線程實(shí)現(xiàn)第三方數(shù)據(jù)同步

更新時(shí)間：2022年08月10日 15:04:31 作者：我高考零分!

這篇文章主要為大家詳細(xì)介紹了Java多線程實(shí)現(xiàn)第三方數(shù)據(jù)同步，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下

本文實(shí)例為大家分享了Java多線程實(shí)現(xiàn)第三方數(shù)據(jù)同步的具體代碼，供大家參考，具體內(nèi)容如下

一、場景

最近的一項(xiàng)開發(fā)任務(wù)是同步第三方數(shù)據(jù)，而第三方數(shù)據(jù)一般有存量數(shù)據(jù)和增量數(shù)據(jù)，存量數(shù)據(jù)有100w+。在得知此需求時(shí)，進(jìn)行了一定的信息檢索和工具學(xué)習(xí)，提前獲取存量數(shù)據(jù)到目標(biāo)庫，再使用kettle進(jìn)行存量數(shù)據(jù)轉(zhuǎn)換；增量數(shù)據(jù)則根據(jù)業(yè)務(wù)方規(guī)定的請求時(shí)間，通過定時(shí)任務(wù)去獲取增量數(shù)據(jù)并進(jìn)行數(shù)據(jù)轉(zhuǎn)換。在數(shù)據(jù)獲取和轉(zhuǎn)換時(shí)，我們應(yīng)該要記錄每一次的請求信息，便于溯源和數(shù)據(jù)對賬!!!

二、獲取數(shù)據(jù)的方式

2.1 遞歸方式

使用遞歸方式時(shí)，要求數(shù)據(jù)量少，否則會(huì)出現(xiàn)棧溢出或堆溢出!!!并且遞歸方式是單線程，所以會(huì)導(dǎo)致同步速度很慢!!!

/**
?? ? * 數(shù)據(jù)同步 - 遞歸方式
?? ? * 此處存量數(shù)據(jù)只需要請求到數(shù)據(jù)并保存數(shù)據(jù)庫即可，后期通過kettle進(jìn)行轉(zhuǎn)換。
?? ? * Data為自定義實(shí)體類，這里僅做示例!!!
*/
?? ?private void fetchAndSaveDB(int pageIndex, int pageSize) throws Exception {
? ? ? ? log.info("【數(shù)據(jù)同步 - 存量】，第{}次同步,", pageIndex);
? ? ? ? List<Data> datas= getDataByPage(pageIndex,pageSize);
? ? ? ? if (CollectionUtils.isNotEmpty(datas)) {
? ? ? ? ? ? dataService.saveOrUpdateBatch(datas);
? ? ? ? ? ? log.info("【數(shù)據(jù)同步 - 存量】，第{}次同步,同步成功", pageIndex);
? ? ? ? ? ? if (datas.size() < pageSize) {
? ? ? ? ? ? ? ? log.info("【數(shù)據(jù)同步 - 存量】,第{}次同步,獲取數(shù)據(jù)小于每頁獲取條數(shù),證明已全部同步完畢!!!", pageIndex);
? ? ? ? ? ? ? ? return;
? ? ? ? ? ? }
? ? ? ? ? ? // 遞歸操作-直到數(shù)據(jù)同步完畢
? ? ? ? ? ? fetchAndSaveDB(pageIndex + 1, pageSize);
? ? ? ? } else {
? ? ? ? ? ? log.info("【數(shù)據(jù)同步 - 存量】,第{}次同步,獲取數(shù)據(jù)為空,證明已全部同步完畢!!!", pageIndex);
? ? ? ? ? ? return;
? ? ? ? }
?? ?}
?? ?/**?
?? ? * 獲取分頁數(shù)據(jù),Data為自定義實(shí)體類，這里僅做示例!!!
?? ? */
? ? private List<Data> getDataByPage(int pageIndex, int pageSize) throws Exception {
? ? ?? ?//通過feign調(diào)用第三方接口獲取數(shù)據(jù)
? ? ? ? String data = dataFeignService.fetchAllData(pageSize, pageIndex);
? ? ? ? JSONObject jsonObject = JSONObject.parseObject(data);
? ? ? ? JSONArray datalist = jsonObject.getJSONArray("datalist");
? ? ? ? List<Data> datas = datalist.toJavaList(Data.class);
? ? ? ? return datas;
? ? }

2.2 多線程方式

由于遞歸方式是單線程，考慮到數(shù)據(jù)的龐大,且易造成內(nèi)存溢出，因此將遞歸更換成多線程方式，不僅避免了內(nèi)存溢出的情況，且速度大大的提升!!!

public void synAllData() {
? ? ??? ?// 定義原子變量 - 頁數(shù)
? ? ?? ?AtomicInteger pageIndex = new AtomicInteger(0);
? ? ??? ?// 創(chuàng)建線程池
? ? ??? ?ExecutorService fixedThreadPool = Executors.newFixedThreadPool(10);

? ? ? ? // 100萬數(shù)據(jù)
? ? ? ? int total = 1000000;//數(shù)據(jù)總量
? ? ? ? int times = total / 1000;
? ? ? ? if (total % 1000!= 0) {
? ? ? ? ? ? times = times + 1;
? ? ? ? }
? ? ? ? LocalDateTime beginLocalDateTime = LocalDateTime.now();
? ? ? ? log.info("【數(shù)據(jù)同步 - 存量】開始同步時(shí)間：{}", beginLocalDateTime.format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
? ? ? ? for (int index = 1; index <= times; index++) {
? ? ? ? ? ? fixedThreadPool.submit(new Runnable() {
? ? ? ? ? ? ? ? @Override
? ? ? ? ? ? ? ? public void run() {
? ? ? ? ? ? ? ? ? ? try {
? ? ? ? ? ? ? ? ? ? ? ? multiFetchAndSaveDB(pageIndex.incrementAndGet(), 1000);
? ? ? ? ? ? ? ? ? ? } catch (Exception e) {
? ? ? ? ? ? ? ? ? ? ? ? log.error("并發(fā)獲取并保存數(shù)據(jù)異常：{}", e);
? ? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? }
? ? ? ? ? ? });
? ? ? ? }
? ? ? ? LocalDateTime endLocalDateTime = LocalDateTime.now();
? ? ? ? log.info("【數(shù)據(jù)同步 - 存量】同步結(jié)束時(shí)間：{},總共耗時(shí)：{}分鐘",
? ? ? ? ? ? ? ? endLocalDateTime.format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")),
? ? ? ? ? ? ? ? Duration.between(beginLocalDateTime, endLocalDateTime).toMinutes());
? ? }
?? ?/**
? ? ?* 數(shù)據(jù)同步 - 【多線程方式】
? ? ?*
? ? ?* @throws Exception
? ? ?*/
? ? private void multiFetchAndSaveDB(int pageIndex, int pageSize) throws Exception {
? ? ? ? log.info("【數(shù)據(jù)同步 - 存量】，第{}次同步,", pageIndex);
? ? ? ? List<Data> datas= getDataByPage(pageIndex, pageSize);//getDataByPage()同上2.1
? ? ? ? if (CollectionUtils.isNotEmpty(datas)) {
? ? ? ? ? ? log.info("【數(shù)據(jù)同步 - 存量】，第{}次同步,同步成功", pageIndex);
? ? ? ? ? ? if (datas.size() < pageSize) {
? ? ? ? ? ? ? ? log.info("【數(shù)據(jù)同步 - 存量】,第{}次同步,獲取數(shù)據(jù)小于每頁獲取條數(shù),證明已全部同步完畢!!!", pageIndex);
? ? ? ? ? ? ? ? return;
? ? ? ? ? ? }
? ? ? ? } else {
? ? ? ? ? ? log.info("【數(shù)據(jù)同步 - 存量】,第{}次同步,獲取數(shù)據(jù)為空,證明已全部同步完畢!!!", pageIndex);
? ? ? ? ? ? return;
? ? ? ? }

? ? }