datasync / yoho-datasync-fullsync · Commits

Go to a project

Authored by 张帅
Commit 7173aca7442505d4f0a6b61e6e88d4d9df132b5c 1 parent 07d2dfd6

fix

Inline Side-by-side
Showing 4 changed files with 31 additions and 116 deletions
... ... @@ -10,6 +10,7 @@ import java.util.List;
@Repository
public interface PublicArticleRepository extends JpaRepository<PublicArticle, Integer> {
@Query("select article from PublicArticle article where article.relateId in(?1) and article.articleType in (?2)")
List<PublicArticle> findAllByRelateIdInAndArticleTypeIn(List<Integer> relateIds, List<Integer> types);
int countByRelateIdInAndArticleTypeIn(List<Integer> relateIds, List<Integer> types);
... ...
... ... @@ -83,23 +83,15 @@ public class GrassArticleSyncServiceImpl implements IGrassArticleSyncService {
@Override
public String syncArticle(Long startTime, Long endTime) {
logger.info("syncArticle begin, startTime is {}, endTime is {}", startTime, endTime);
ExecutorService es = Executors.newFixedThreadPool(2);
int pageSize = 100;
int total = grassArticleRepository.countByCreateTimeAfterAndCreateTimeBeforeAndArticleTypeInAndAuthStatusIn(startTime, endTime,needSyncArticleType,needSyncArticleStatus);
logger.info("syncArticle get grassArticle count is {}", total);
int totalPage = total % pageSize ==0 ? total / pageSize : (total / pageSize)+1;
try{
//每次查询100条，增加detail
for (int index=0; index < totalPage; index++) {
int indexPage = index;
es.execute(() -> {
syncArticle(startTime, endTime, indexPage, pageSize);
});
}
}finally {
es.shutdown();
for (int index=0; index < totalPage; index++) {
syncArticle(startTime, endTime, index, pageSize);
}
return String.valueOf(total);
}
... ...
... ... @@ -162,25 +162,17 @@ 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.newFixedThreadPool(2);
int pageSize = 100;
Long startTimeLong = (long)startTime * 1000;
Long endTimeLong = (long)endTime * 1000;
int total = grassArticleRepository.countByCreateTimeAfterAndCreateTimeBeforeAndArticleTypeInAndAuthStatusIn(startTimeLong, endTimeLong,needSyncArticleType,needSyncArticleStatus);
logger.info("syncPraiseByArticle get grassArticle count is {}", total);
int totalPage = total % pageSize ==0 ? total / pageSize : (total / pageSize)+1;
try{
//每次查询100条，增加detail
for (int index=0; index < totalPage; index++) {
int indexPage = index;
es.execute(() -> {
syncPraiseByArticlePage(startTimeLong, endTimeLong, indexPage, pageSize);
});
}
}finally {
es.shutdown();
for (int index=0; index < totalPage; index++) {
syncPraiseByArticlePage(startTimeLong, endTimeLong, index, pageSize);
}
}
private void syncPraiseByArticlePage(Long startTime, Long endTime, int indexPage, int pageSize){
... ... @@ -290,23 +282,15 @@ 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.newFixedThreadPool(2);
int pageSize = 100;
int total = grassArticlePraiseRepository.countByCreateTimeBetween(startTime, endTime);
logger.info("syncPraiseBySelf get grassArticlePraise count is {}", total);
int totalPage = total % pageSize ==0 ? total / pageSize : (total / pageSize)+1;
try{
//每次查询100条，增加detail
for (int index=0; index < totalPage; index++) {
int indexPage = index;
es.execute(() -> {
syncPraiseBySelfPage(startTime, endTime, indexPage, pageSize);
});
}
}finally {
es.shutdown();
for (int index=0; index < totalPage; index++) {
syncPraiseBySelfPage(startTime, endTime, index, pageSize);
}
}
private void syncPraiseBySelfPage(int startTime, int endTime, int indexPage, int pageSize){
... ... @@ -340,25 +324,17 @@ 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.newFixedThreadPool(2);
Long startTimeLong = (long)startTime * 1000;
Long endTimeLong = (long)endTime * 1000;
int pageSize = 100;
int total = grassArticleRepository.countByCreateTimeAfterAndCreateTimeBeforeAndArticleTypeInAndAuthStatusIn(startTimeLong, endTimeLong,needSyncArticleType,needSyncArticleStatus);
logger.info("syncFavoriteByArticle get grassArticle count is {}", total);
int totalPage = total % pageSize ==0 ? total / pageSize : (total / pageSize)+1;
try{
//每次查询100条，增加detail
for (int index=0; index < totalPage; index++) {
int indexPage = index;
es.execute(() -> {
syncFavoriteByArticlePage(startTimeLong, endTimeLong, indexPage, pageSize);
});
}
}finally {
es.shutdown();
for (int index=0; index < totalPage; index++) {
syncFavoriteByArticlePage(startTimeLong, endTimeLong, index, pageSize);
}
}
private void syncFavoriteByArticlePage(Long startTime, Long endTime, int indexPage, int pageSize){
... ... @@ -421,23 +397,15 @@ 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.newFixedThreadPool(2);
int pageSize = 100;
int total = userFavoriteArticleRepository.countByCreateTimeBetween(startTime, endTime);
logger.info("syncFavoriteBySelf get grassArticlePraise count is {}", total);
int totalPage = total % pageSize ==0 ? total / pageSize : (total / pageSize)+1;
try{
//每次查询100条，增加detail
for (int index=0; index < totalPage; index++) {
int indexPage = index;
es.execute(() -> {
syncFavoriteBySelfPage(startTime, endTime, indexPage, pageSize);
});
}
}finally {
es.shutdown();
for (int index=0; index < totalPage; index++) {
syncFavoriteBySelfPage(startTime, endTime, index, pageSize);
}
}
private void syncFavoriteBySelfPage(int startTime, int endTime, int indexPage, int pageSize){
... ... @@ -471,23 +439,15 @@ 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.newFixedThreadPool(2);
int pageSize = 100;
int total = grassUserAttentionRepository.countByCreateTimeBetween(startTime, endTime);
logger.info("syncUserAttention get attention count is {}", total);
int totalPage = total % pageSize ==0 ? total / pageSize : (total / pageSize)+1;
try{
//每次查询100条，增加detail
for (int index=0; index < totalPage; index++) {
int indexPage = index;
es.execute(() -> {
syncUserAttentionPage(startTime, endTime, indexPage, pageSize);
});
}
}finally {
es.shutdown();
for (int index=0; index < totalPage; index++) {
syncUserAttentionPage(startTime, endTime, index, pageSize);
}
}
private void syncUserAttentionPage(int startTime, int endTime, int indexPage, int pageSize){
... ... @@ -594,24 +554,17 @@ 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.newFixedThreadPool(2);
int pageSize = 100;
Long startTimeLong = (long)startTime * 1000;
Long endTimeLong = (long)endTime * 1000;
int total = grassArticleRepository.countByCreateTimeAfterAndCreateTimeBeforeAndArticleTypeInAndAuthStatusIn(startTimeLong, endTimeLong,needSyncArticleType,needSyncArticleStatus);
logger.info("syncCommentsByArticle get grassArticle count is {}", total);
int totalPage = total % pageSize ==0 ? total / pageSize : (total / pageSize)+1;
try{
//每次查询100条，增加detail
for (int index=0; index < totalPage; index++) {
int indexPage = index;
es.execute(() -> {
syncCommentsByArticlePage(startTimeLong, endTimeLong, indexPage, pageSize);
});
}
}finally {
es.shutdown();
for (int index=0; index < totalPage; index++) {
syncCommentsByArticlePage(startTimeLong, endTimeLong, index, pageSize);
}
}
/**
... ... @@ -624,21 +577,13 @@ 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.newFixedThreadPool(2);
int pageSize = 100;
int total = grassArticleCommentRepository.countByCreateTimeBetween(startTime, endTime);
logger.info("syncCommentsBySelf get grassArticlePraise count is {}", total);
int totalPage = total % pageSize ==0 ? total / pageSize : (total / pageSize)+1;
try{
//每次查询100条，增加detail
for (int index=0; index < totalPage; index++) {
int indexPage = index;
es.execute(() -> {
syncCommentsBySelfPage(startTime, endTime, indexPage, pageSize);
});
}
}finally {
es.shutdown();
for (int index=0; index < totalPage; index++) {
syncCommentsBySelfPage(startTime, endTime, index, pageSize);
}
}
... ... @@ -927,21 +872,13 @@ 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.newFixedThreadPool(2);
int pageSize = 100;
int total = grassCommentPraiseRepository.countByCreateTimeBetween(startTime, endTime);
logger.info("syncCommentPraise get commentPraise count is {}", total);
int totalPage = total % pageSize ==0 ? total / pageSize : (total / pageSize)+1;
try{
//每次查询100条，增加detail
for (int index=0; index < totalPage; index++) {
int indexPage = index;
es.execute(() -> {
syncCommentPraisePage(startTime, endTime, indexPage, pageSize);
});
}
}finally {
es.shutdown();
for (int index=0; index < totalPage; index++) {
syncCommentPraisePage(startTime, endTime, index, pageSize);
}
}
... ...
... ... @@ -80,20 +80,12 @@ 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.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;
try{
//每次查询100条，增加detail
for (int index=0; index < totalPage; index++) {
int indexPage = index;
es.execute(() -> {
syncUserAchieveDataPage(startTime, endTime, indexPage);
});
}
}finally {
es.shutdown();
for (int index=0; index < totalPage; index++) {
syncUserAchieveDataPage(startTime, endTime, index);
}
}
... ... @@ -171,21 +163,14 @@ 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.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;
try{
//每次查询100条，增加detail
for (int index=0; index < totalPage; index++) {
int indexPage = index;
es.execute(() -> {
syncVirtualUserDataPage(startTime, endTime, indexPage);
});
}
}finally {
es.shutdown();
for (int index=0; index < totalPage; index++) {
syncVirtualUserDataPage(startTime, endTime, index);
}
}
private void syncVirtualUserDataPage(Integer startTime, Integer endTime, int indexPage){
... ...