datasync / yoho-datasync-fullsync · Commits

Go to a project

Authored by 张帅
Commit ca987c0be3a3c0176ff3b279c321cb65695a7ed9 2 parents b1df6d4b 80e533f5

Merge branch 'test6.8.9'

Inline Side-by-side
Showing 3 changed files with 76 additions and 13 deletions
... ... @@ -89,6 +89,7 @@ public class GrassArticleSyncServiceImpl implements IGrassArticleSyncService {
int indexPage = index;
es.execute(() -> {
syncArticle(startTime, endTime, indexPage, pageSize);
});
}
}finally {
... ... @@ -107,6 +108,13 @@ public class GrassArticleSyncServiceImpl implements IGrassArticleSyncService {
syncArticleProduct(publicArticleList);
//同步label 和 article_label表
syncArticleLabel(publicArticleList);
try {
Thread.sleep(200);
} catch (InterruptedException e) {
logger.warn("syncArticle syncArticle error !error{}", e);
}
}
... ... @@ -217,9 +225,8 @@ public class GrassArticleSyncServiceImpl implements IGrassArticleSyncService {
for (GrassArticleBlock grassArticleBlock:grassArticleBlocks) {
//第一个image为默认封面图
if("image".equals(grassArticleBlock.getTemplateKey())){
String dataString = JSONObject.parseObject(grassArticleBlock.getContentData()).getString("data");
JSONObject block = JSONObject.parseObject(dataString);
return block.getString("src");
String dataString = getBlock(grassArticleBlock.getTemplateKey(),grassArticleBlock.getContentData(),true);
return dataString;
}
}
... ...
... ... @@ -157,7 +157,7 @@ public class GrassInteractiveDataSyncServiceImpl implements IGrassInteractiveDat
*/
private void syncPraiseByArticle(Integer startTime, Integer endTime){
logger.info("syncPraiseByArticle begin, startTime is {}, endTime is {}", startTime, endTime);
ExecutorService es = Executors.newSingleThreadExecutor();
ExecutorService es = Executors.newFixedThreadPool(2);
int pageSize = 100;
Long startTimeLong = (long)startTime * 1000;
Long endTimeLong = (long)endTime * 1000;
... ... @@ -170,6 +170,7 @@ public class GrassInteractiveDataSyncServiceImpl implements IGrassInteractiveDat
int indexPage = index;
es.execute(() -> {
syncPraiseByArticlePage(startTimeLong, endTimeLong, indexPage, pageSize);
});
}
}finally {
... ... @@ -191,6 +192,11 @@ public class GrassInteractiveDataSyncServiceImpl implements IGrassInteractiveDat
logger.info("syncPraiseByArticlePage find grassArticlePraise num is {}", grassArticlePraiseList.size());
List<PublicUserPraise> needSyncData = buildPublicUserPraise(publicArticleIdAndGrassIdMap, nowArticleIdAndpublicArticleIdMap,grassArticlePraiseList);
publicUserPraiseRepository.saveAll(needSyncData);
try {
Thread.sleep(200);
} catch (InterruptedException e) {
logger.warn(" sync error{}", e);
}
}
private List<PublicUserPraise> buildPublicUserPraise(Map<Integer, Integer> publicArticleIdAndGrassIdMap,Map<Integer, Integer> nowArticleIdAndpublicArticleIdMap, List<GrassArticlePraise> grassArticlePraiseList){
... ... @@ -238,7 +244,7 @@ public class GrassInteractiveDataSyncServiceImpl implements IGrassInteractiveDat
*/
private void syncPraiseBySelf(Integer startTime, Integer endTime){
logger.info("syncPraiseBySelf begin, startTime is {}, endTime is {}", startTime, endTime);
ExecutorService es = Executors.newSingleThreadExecutor();
ExecutorService es = Executors.newFixedThreadPool(2);
int pageSize = 100;
int total = grassArticlePraiseRepository.countByCreateTimeBetween(startTime, endTime);
logger.info("syncPraiseBySelf get grassArticlePraise count is {}", total);
... ... @@ -249,6 +255,7 @@ public class GrassInteractiveDataSyncServiceImpl implements IGrassInteractiveDat
int indexPage = index;
es.execute(() -> {
syncPraiseBySelfPage(startTime, endTime, indexPage, pageSize);
});
}
}finally {
... ... @@ -270,6 +277,11 @@ public class GrassInteractiveDataSyncServiceImpl implements IGrassInteractiveDat
List<PublicUserPraise> needSyncData = buildPublicUserPraise(publicArticleIdAndGrassIdMap,nowArticleIdAndpublicArticleIdMap, grassArticlePraiseList);
publicUserPraiseRepository.saveAll(needSyncData);
try {
Thread.sleep(200);
} catch (InterruptedException e) {
logger.warn("sync error{}", e);
}
}
... ... @@ -280,7 +292,7 @@ public class GrassInteractiveDataSyncServiceImpl implements IGrassInteractiveDat
*/
private void syncFavoriteByArticle(Integer startTime, Integer endTime){
logger.info("syncFavoriteByArticle begin, startTime is {}, endTime is {}", startTime, endTime);
ExecutorService es = Executors.newSingleThreadExecutor();
ExecutorService es = Executors.newFixedThreadPool(2);
Long startTimeLong = (long)startTime * 1000;
Long endTimeLong = (long)endTime * 1000;
int pageSize = 100;
... ... @@ -293,6 +305,7 @@ public class GrassInteractiveDataSyncServiceImpl implements IGrassInteractiveDat
int indexPage = index;
es.execute(() -> {
syncFavoriteByArticlePage(startTimeLong, endTimeLong, indexPage, pageSize);
});
}
}finally {
... ... @@ -315,6 +328,11 @@ public class GrassInteractiveDataSyncServiceImpl implements IGrassInteractiveDat
logger.info("syncFavoriteByArticlePage find userFavoriteArticle num is {}", userFavoriteArticleList.size());
List<PublicUserFavorite> needSyncData = buildPublicUserFavorite(publicArticleIdAndGrassIdMap,nowArticleIdAndpublicArticleIdMap, userFavoriteArticleList);
publicUserFavoriteRepository.saveAll(needSyncData);
try {
Thread.sleep(200);
} catch (InterruptedException e) {
logger.warn("sync error{}", e);
}
}
... ... @@ -353,7 +371,7 @@ public class GrassInteractiveDataSyncServiceImpl implements IGrassInteractiveDat
*/
private void syncFavoriteBySelf(Integer startTime, Integer endTime){
logger.info("syncFavoriteBySelf begin, startTime is {}, endTime is {}", startTime, endTime);
ExecutorService es = Executors.newSingleThreadExecutor();
ExecutorService es = Executors.newFixedThreadPool(2);
int pageSize = 100;
int total = userFavoriteArticleRepository.countByCreateTimeBetween(startTime, endTime);
logger.info("syncFavoriteBySelf get grassArticlePraise count is {}", total);
... ... @@ -364,6 +382,7 @@ public class GrassInteractiveDataSyncServiceImpl implements IGrassInteractiveDat
int indexPage = index;
es.execute(() -> {
syncFavoriteBySelfPage(startTime, endTime, indexPage, pageSize);
});
}
}finally {
... ... @@ -385,6 +404,11 @@ public class GrassInteractiveDataSyncServiceImpl implements IGrassInteractiveDat
List<PublicUserFavorite> needSyncData = buildPublicUserFavorite(publicArticleIdAndGrassIdMap,nowArticleIdAndpublicArticleIdMap, userFavoriteArticleList);
publicUserFavoriteRepository.saveAll(needSyncData);
try {
Thread.sleep(200);
} catch (InterruptedException e) {
logger.warn("sync error{}", e);
}
}
... ... @@ -395,7 +419,7 @@ public class GrassInteractiveDataSyncServiceImpl implements IGrassInteractiveDat
*/
private void syncUserAttention(Integer startTime, Integer endTime){
logger.info("syncUserAttention begin, startTime is {}, endTime is {}", startTime, endTime);
ExecutorService es = Executors.newSingleThreadExecutor();
ExecutorService es = Executors.newFixedThreadPool(2);
int pageSize = 100;
int total = grassUserAttentionRepository.countByCreateTimeBetween(startTime, endTime);
logger.info("syncUserAttention get attention count is {}", total);
... ... @@ -406,6 +430,7 @@ public class GrassInteractiveDataSyncServiceImpl implements IGrassInteractiveDat
int indexPage = index;
es.execute(() -> {
syncUserAttentionPage(startTime, endTime, indexPage, pageSize);
});
}
}finally {
... ... @@ -420,6 +445,11 @@ public class GrassInteractiveDataSyncServiceImpl implements IGrassInteractiveDat
List<PublicUserAttention> needSyncData = buildPublicUserAttention(grassUserAttentionList);
publicUserAttentionRepository.saveAll(needSyncData);
try {
Thread.sleep(200);
} catch (InterruptedException e) {
logger.warn("sync error{}", e);
}
}
private List<PublicUserAttention> buildPublicUserAttention( List<GrassUserAttention> grassUserAttentionList){
... ... @@ -465,7 +495,7 @@ public class GrassInteractiveDataSyncServiceImpl implements IGrassInteractiveDat
*/
private void syncCommentsByArticle(Integer startTime, Integer endTime){
logger.info("syncCommentsByArticle begin, startTime is {}, endTime is {}", startTime, endTime);
ExecutorService es = Executors.newSingleThreadExecutor();
ExecutorService es = Executors.newFixedThreadPool(2);
int pageSize = 100;
Long startTimeLong = (long)startTime * 1000;
Long endTimeLong = (long)endTime * 1000;
... ... @@ -495,7 +525,7 @@ public class GrassInteractiveDataSyncServiceImpl implements IGrassInteractiveDat
*/
private void syncCommentsBySelf(Integer startTime, Integer endTime){
logger.info("syncCommentsBySelf begin, startTime is {}, endTime is {}", startTime, endTime);
ExecutorService es = Executors.newSingleThreadExecutor();
ExecutorService es = Executors.newFixedThreadPool(2);
int pageSize = 100;
int total = grassArticleCommentRepository.countByCreateTimeBetween(startTime, endTime);
logger.info("syncCommentsBySelf get grassArticlePraise count is {}", total);
... ... @@ -554,6 +584,12 @@ public class GrassInteractiveDataSyncServiceImpl implements IGrassInteractiveDat
buildAndSaveChildrenCommentBySelf(publicArticleIdAndGrassIdMap, grassChildrenArticleComments, relateIdMap);
try {
Thread.sleep(200);
} catch (InterruptedException e) {
logger.warn("sync error{}", e);
}
}
/**
... ... @@ -587,6 +623,11 @@ public class GrassInteractiveDataSyncServiceImpl implements IGrassInteractiveDat
//同步所有层级的子评论
buildAndSaveChildrenComment(publicArticleIdAndGrassIdMap,rootPublicComments, childrenComments, rootIdMap);
try {
Thread.sleep(200);
} catch (InterruptedException e) {
logger.warn("sync error{}", e);
}
}
... ... @@ -784,7 +825,7 @@ public class GrassInteractiveDataSyncServiceImpl implements IGrassInteractiveDat
*/
private void syncCommentPraise(Integer startTime, Integer endTime){
logger.info("syncCommentPraise begin, startTime is {}, endTime is {}", startTime, endTime);
ExecutorService es = Executors.newSingleThreadExecutor();
ExecutorService es = Executors.newFixedThreadPool(2);
int pageSize = 100;
int total = grassCommentPraiseRepository.countByCreateTimeBetween(startTime, endTime);
logger.info("syncCommentPraise get commentPraise count is {}", total);
... ... @@ -820,6 +861,11 @@ public class GrassInteractiveDataSyncServiceImpl implements IGrassInteractiveDat
List<PublicUserPraise> needSyncData = buildPublicUserPraise(relateIdMap, grassCommentPraiseList);
publicUserPraiseRepository.saveAll(needSyncData);
try {
Thread.sleep(200);
} catch (InterruptedException e) {
logger.warn("sync error{}", e);
}
}
private List<PublicUserPraise> buildPublicUserPraise(Map<Integer, Integer> relateIdMap, List<GrassCommentPraise> grassCommentPraiseList){
... ...
... ... @@ -60,7 +60,7 @@ public class GrassUserDataSyncServiceImpl implements IGrassUserDataSyncService {
private void syncUserAchieveData(Integer startTime, Integer endTime){
logger.info("syncUserAchieveData begin, startTime is {}, endTime is {}", startTime, endTime);
ExecutorService es = Executors.newSingleThreadExecutor();
ExecutorService es = Executors.newFixedThreadPool(2);
int total = grassUserAchieveRepository.countByCreateTimeBetween(startTime, endTime);
logger.info("syncUserAchieveData get grassArticlePraise count is {}", total);
int totalPage = total % pageSize ==0 ? total / pageSize : (total / pageSize)+1;
... ... @@ -83,6 +83,11 @@ public class GrassUserDataSyncServiceImpl implements IGrassUserDataSyncService {
List<PublicUserHomePage> needSyncData = buildPublicUserHomePage(grassUserAchieveList);
logger.info("syncUserAchieveDataPage need sync data count is {}", needSyncData.size());
publicUserHomePageRepository.saveAll(needSyncData);
try {
Thread.sleep(200);
} catch (InterruptedException e) {
logger.warn("sync error{}", e);
}
}
private List<PublicUserHomePage> buildPublicUserHomePage( List<GrassUserAchieve> grassUserAchieveList ){
... ... @@ -107,7 +112,7 @@ public class GrassUserDataSyncServiceImpl implements IGrassUserDataSyncService {
private void syncVirtualUserData(Integer startTime, Integer endTime){
logger.info("syncVirtualUserData begin, startTime is {}, endTime is {}", startTime, endTime);
ExecutorService es = Executors.newSingleThreadExecutor();
ExecutorService es = Executors.newFixedThreadPool(2);
int total = grassVirtualUserRepository.countByCreateTimeBetween(startTime, endTime);
logger.info("syncVirtualUserData get grassArticlePraise count is {}", total);
int totalPage = total % pageSize ==0 ? total / pageSize : (total / pageSize)+1;
... ... @@ -130,6 +135,11 @@ public class GrassUserDataSyncServiceImpl implements IGrassUserDataSyncService {
List<PublicVirtualUser> needSyncData = buildPublicVirtualUserPage(grassUserAchieveList);
logger.info("syncUserAchieveDataPage need sync data count is {}", needSyncData.size());
publicVirtualUserRepository.saveAll(needSyncData);
try {
Thread.sleep(200);
} catch (InterruptedException e) {
logger.warn("sync error{}", e);
}
}
... ...