datasync / yoho-datasync-fullsync · Commits

Go to a project

Authored by chaogeng
Commit 364040f0f2d63189a5d77fc2e859502e20dc372b 2 parents 829f8960 6e4c6eae

Merge branch 'test6.8.9' of http://git.yoho.cn/datasync/yoho-datasync-fullsync into test6.8.9

Inline Side-by-side
Showing 6 changed files with 229 additions and 17 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);
}
}
... ...
spring:
datasource:
url: jdbc:mysql://192.168.102.219:3306/yh_pcms?useUnicode=true&amp;characterEncoding=UTF-8
username: yh_test
password: 9nm0icOwt6bMHjMusIfMLw==
driver-class-name: com.mysql.jdbc.Driver
primary:
url: jdbc:mysql://write.cms.yohoops.org:3306/yh_grass?useUnicode=true&amp;characterEncoding=UTF-8
username: yh_vpc_bak
password: +BfhVxZQ4LuPKt+QxSy9naMwEu/zaKV31I9S8xDJIUA=
driver-class-name: com.mysql.jdbc.Driver
secondary:
url: jdbc:mysql://write.cms.yohoops.org:3306/yh_pcms?useUnicode=true&amp;characterEncoding=UTF-8
username: yh_vpc_bak
password: +BfhVxZQ4LuPKt+QxSy9naMwEu/zaKV31I9S8xDJIUA=
driver-class-name: com.mysql.jdbc.Driver
logging:
config: classpath:logback-boot.xml
\ No newline at end of file
... ...
... ... @@ -11,3 +11,6 @@ spring:
username: yh_test
password: 9nm0icOwt6bMHjMusIfMLw==
driver-class-name: com.mysql.jdbc.Driver
logging:
config: classpath:logback-boot.xml
\ No newline at end of file
... ...
<?xml version="1.0" encoding="UTF-8"?>
<!-- 从高到地低 OFF 、 FATAL 、 ERROR 、 WARN 、 INFO 、 DEBUG 、 TRACE 、 ALL -->
<!-- 日志输出规则 根据当前ROOT 级别，日志输出时，级别高于root默认的级别时 会输出 -->
<!-- 以下 每个配置的 filter 是过滤掉输出文件里面，会出现高级别文件，依然出现低级别的日志信息，通过filter 过滤只记录本级别的日志-->
<!-- 属性描述 scan：性设置为true时，配置文件如果发生改变，将会被重新加载，默认值为true scanPeriod:设置监测配置文件是否有修改的时间间隔，如果没有给出时间单位，默认单位是毫秒。当scan为true时，此属性生效。默认的时间间隔为1分钟。
debug:当此属性设置为true时，将打印出logback内部日志信息，实时查看logback运行状态。默认值为false。 -->
<configuration scan="true" scanPeriod="60 seconds" debug="false">
<!-- 日志最大的历史 7天 -->
<property name="maxHistory" value="7"/>
<!-- 日志最大的文件大小 10MB-->
<property name="maxFileSize" value="10MB"/>
<!-- 彩色日志 -->
<!-- 彩色日志依赖的渲染类 -->
<conversionRule conversionWord="clr" converterClass="org.springframework.boot.logging.logback.ColorConverter" />
<conversionRule conversionWord="wex" converterClass="org.springframework.boot.logging.logback.WhitespaceThrowableProxyConverter" />
<conversionRule conversionWord="wEx" converterClass="org.springframework.boot.logging.logback.ExtendedWhitespaceThrowableProxyConverter" />
<!-- 彩色日志格式 -->
<property name="CONSOLE_LOG_PATTERN" value="${CONSOLE_LOG_PATTERN:-%clr(%d{yyyy-MM-dd HH:mm:ss.SSS}){faint} %clr(${LOG_LEVEL_PATTERN:-%5p}) %clr(${PID:- }){magenta} %clr(---){faint} %clr([%15.15t]){faint} %clr(%-40.40logger{39}){cyan} %clr(:){faint} %m%n${LOG_EXCEPTION_CONVERSION_WORD:-%wEx}}"/>
<!-- ConsoleAppender 控制台输出日志 -->
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<!-- 对日志进行格式化 -->
<encoder>
<Pattern>${CONSOLE_LOG_PATTERN}</Pattern>
<!-- 设置字符集 -->
<charset>UTF-8</charset>
</encoder>
</appender>
<!-- DEBUG、INFO、WARN、ERROR级别日志 appender -->
<appender name="DEBUG" class="ch.qos.logback.core.rolling.RollingFileAppender">
<file>/Data/logs/yoho-datasync-fullsync/debug.log</file>
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
<!-- daily rollover -->
<fileNamePattern>/Data/logs/yoho-datasync-fullsync/archived/debug.%d{yyyy-MM-dd}.%i.log</fileNamePattern>
<timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
<!-- or whenever the file size reaches 100MB -->
<maxFileSize>${maxFileSize}</maxFileSize>
</timeBasedFileNamingAndTriggeringPolicy>
<!-- keep 30 days' worth of history -->
<maxHistory>${maxHistory}</maxHistory>
</rollingPolicy>
<encoder>
<Pattern>${CONSOLE_LOG_PATTERN}</Pattern>
<!-- 设置字符集 -->
<charset>UTF-8</charset>
</encoder>
</appender>
<!-- INFO级别日志 appender -->
<appender name="INFO" class="ch.qos.logback.core.rolling.RollingFileAppender">
<file>/Data/logs/yoho-datasync-fullsync/info.log</file>
<!-- 过滤器，过滤掉 TRACE 和 DEBUG 级别的日志 -->
<filter class="ch.qos.logback.classic.filter.ThresholdFilter">
<level>INFO</level>
</filter>
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
<!-- daily rollover -->
<fileNamePattern>/Data/logs/yoho-datasync-fullsync/archived/info.%d{yyyy-MM-dd}.%i.log</fileNamePattern>
<timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
<!-- or whenever the file size reaches 100MB -->
<maxFileSize>${maxFileSize}</maxFileSize>
</timeBasedFileNamingAndTriggeringPolicy>
<!-- keep 30 days' worth of history -->
<maxHistory>${maxHistory}</maxHistory>
</rollingPolicy>
<encoder>
<Pattern>${CONSOLE_LOG_PATTERN}</Pattern>
<!-- 设置字符集 -->
<charset>UTF-8</charset>
</encoder>
</appender>
<!-- WARN级别日志 appender -->
<appender name="WARN" class="ch.qos.logback.core.rolling.RollingFileAppender">
<file>/Data/logs/yoho-datasync-fullsync/warn.log</file>
<!-- 过滤器，过滤掉 TRACE 和 DEBUG 和 INFO 级别的日志 -->
<filter class="ch.qos.logback.classic.filter.ThresholdFilter">
<level>WARN</level>
</filter>
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
<!-- daily rollover -->
<fileNamePattern>/Data/logs/yoho-datasync-fullsync/archived/warn.%d{yyyy-MM-dd}.%i.log</fileNamePattern>
<timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
<!-- or whenever the file size reaches 100MB -->
<maxFileSize>${maxFileSize}</maxFileSize>
</timeBasedFileNamingAndTriggeringPolicy>
<!-- keep 30 days' worth of history -->
<maxHistory>${maxHistory}</maxHistory>
</rollingPolicy>
<encoder>
<Pattern>${CONSOLE_LOG_PATTERN}</Pattern>
<!-- 设置字符集 -->
<charset>UTF-8</charset>
</encoder>
</appender>
<!-- 数据检测相关日志 -->
<appender name="DATA_CHECKERROR_LOG" class="ch.qos.logback.core.rolling.RollingFileAppender">
<file>/Data/logs/yoho-datasync-fullsync/datacheck-error.log</file>
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
<!-- daily rollover -->
<fileNamePattern>/Data/logs/yoho-datasync-fullsync/archived/datacheck-error.%d{yyyy-MM-dd}.%i.log</fileNamePattern>
<timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
<!-- or whenever the file size reaches 100MB -->
<maxFileSize>${maxFileSize}</maxFileSize>
</timeBasedFileNamingAndTriggeringPolicy>
<!-- keep 30 days' worth of history -->
<maxHistory>${maxHistory}</maxHistory>
</rollingPolicy>
<encoder>
<Pattern>${CONSOLE_LOG_PATTERN}</Pattern>
<!-- 设置字符集 -->
<charset>UTF-8</charset>
</encoder>
</appender>
<!-- 数据检测相关日志 -->
<logger name="DATA_CHECKERROR" additivity="false">
<level value="INFO"/>
<appender-ref ref="DATA_CHECKERROR_LOG"/>
</logger>
<!-- root级别 DEBUG -->
<root level="INFO">
<!-- 控制台输出 -->
<appender-ref ref="STDOUT" />
<!-- 文件输出 -->
<appender-ref ref="DEBUG" />
<appender-ref ref="WARN" />
</root>
</configuration>
\ No newline at end of file
... ...