From 2da3f2ae83c0a27ea3b01b7d201ab4b4103bb9fa Mon Sep 17 00:00:00 2001
From: lqk <1141706615@qq.com>
Date: Thu, 8 Dec 2016 11:17:29 +0800
Subject: [PATCH] add gatherthreadnew
---
.../resources/applicationContext-myBatis.xml | 4 +-
gather_program/sql/pk_control_projects.sql | 2 +-
.../ossean/gather/process/GatherProcess.java | 10 +-
.../gather/process/GatherThreadNew.java | 398 ++++++++++++++++++
.../resource/applicationContext-myBatis.xml | 8 +-
.../src/main/resource/gather_projects.xml | 2 +-
6 files changed, 412 insertions(+), 12 deletions(-)
create mode 100644 gather_program/src/main/java/org/ossean/gather/process/GatherThreadNew.java
diff --git a/gather_program/bin/resources/applicationContext-myBatis.xml b/gather_program/bin/resources/applicationContext-myBatis.xml
index 96b11e1ea..69361cfaa 100644
--- a/gather_program/bin/resources/applicationContext-myBatis.xml
+++ b/gather_program/bin/resources/applicationContext-myBatis.xml
@@ -39,9 +39,9 @@
destroy-method="close">
+ value="jdbc:mysql://172.16.128.30:3306/ossean_production?characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull&autoReconnect=true" />
-
+
diff --git a/gather_program/sql/pk_control_projects.sql b/gather_program/sql/pk_control_projects.sql
index 1c505c75b..a36891a37 100644
--- a/gather_program/sql/pk_control_projects.sql
+++ b/gather_program/sql/pk_control_projects.sql
@@ -20,7 +20,7 @@ SET FOREIGN_KEY_CHECKS=0;
-- ----------------------------
DROP TABLE IF EXISTS `pk_control_projects`;
CREATE TABLE `pk_control_projects` (
- `id` int(11) NOT NULL,
+ `id` int(11) NOT NULL AUTO_INCREMENT,
`url_md5` varchar(255) DEFAULT NULL,
PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=latin1;
diff --git a/gather_program/src/main/java/org/ossean/gather/process/GatherProcess.java b/gather_program/src/main/java/org/ossean/gather/process/GatherProcess.java
index 7fbbbd4f8..f79be45c5 100644
--- a/gather_program/src/main/java/org/ossean/gather/process/GatherProcess.java
+++ b/gather_program/src/main/java/org/ossean/gather/process/GatherProcess.java
@@ -44,9 +44,10 @@ public class GatherProcess {
else if(state == true){
continue;
}
- GatherThread gatherThread = (GatherThread)AppContext.appContext.getBean("gatherThread");
- gatherThread.setParameters(conf,sourceTableName);
- pool.execute(gatherThread);
+ //GatherThread gatherThread = (GatherThread)AppContext.appContext.getBean("gatherThread");
+ GatherThreadNew gatherThreadNew = (GatherThreadNew)AppContext.appContext.getBean("gatherThreadNew");
+ gatherThreadNew.setParameters(conf,sourceTableName);
+ pool.execute(gatherThreadNew);
}
try {
logger.info(".......sleeping......");
@@ -66,7 +67,8 @@ public class GatherProcess {
configureName = args[0].toString();
} else {
logger.error("404 configure!");
- configureName = "relative_memos";
+ //configureName = "relative_memos";
+ configureName = "gather_projects";
//return;
}
ApplicationContext applicationContext = new ClassPathXmlApplicationContext("classpath:/applicationContext*.xml");
diff --git a/gather_program/src/main/java/org/ossean/gather/process/GatherThreadNew.java b/gather_program/src/main/java/org/ossean/gather/process/GatherThreadNew.java
new file mode 100644
index 000000000..4899ca906
--- /dev/null
+++ b/gather_program/src/main/java/org/ossean/gather/process/GatherThreadNew.java
@@ -0,0 +1,398 @@
+package org.ossean.gather.process;
+/**
+ * @author 李乾坤 2016年12月5号第一次修改。修改内容如下：
+ * 1.将获取tag部分抽离为一个单独的可以接收参数的函数handleTags
+ * 2.改变了数据汇总策略。放弃之前把记录全放到set中的方法。采用如下策略：
+ * 1>查询pk_control_*表，查看当前帖子或者项目是否已经有固定Id
+ * 2>查询到固定Id值（sameId），则依据sameId更新数据库中的内容。未查到固定Id进行3>
+ * 3>将当前项目或帖子依据url_md5插入pk_control_*表以生成固定Id（newId），然后根据newId执行插入操作。
+ * */
+import java.util.List;
+
+import javax.annotation.Resource;
+
+import org.apache.log4j.Logger;
+import org.ossean.gather.model.Configure;
+import org.ossean.gather.model.GatherProject;
+import org.ossean.gather.model.JobRequirement;
+import org.ossean.gather.model.PKControlPosts;
+import org.ossean.gather.model.PKControlProjects;
+import org.ossean.gather.model.RelativeMemo;
+import org.ossean.gather.model.Taggings;
+import org.ossean.gather.sourceDao.GatherDao;
+import org.ossean.gather.sourceDao.PKControlPostsDao;
+import org.ossean.gather.sourceDao.PKControlProjectsDao;
+import org.ossean.gather.targetDao.PointerDao;
+import org.ossean.gather.targetDao.TargetDao;
+import org.springframework.context.annotation.Scope;
+import org.springframework.jmx.export.annotation.ManagedResource;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Propagation;
+import org.springframework.transaction.annotation.Transactional;
+
+@Component("gatherThreadNew")
+@Scope("prototype")
+public class GatherThreadNew implements Runnable {
+ private static Logger logger = Logger.getLogger(GatherThread.class);
+ private Configure conf;
+
+ //注解的方式生成bean并获取汇总操作对象
+ @Resource
+ private GatherDao gatherDao;
+ @Resource
+ private PointerDao pointerDao;
+ //控制帖子的操作
+ @Resource
+ private PKControlPostsDao pkControlPostsDao;
+ //汇总表操作对象
+ @Resource
+ private TargetDao targetDao;
+ //项目控制表操作
+ @Resource
+ private PKControlProjectsDao pkControlProjectsDao;
+
+ private int idsBegin; // 转移开始Id值
+ private int idsEnd; // 转移结束Id值
+ private int idsIncrement;// 每次转移的Id量
+
+ private int beginId;
+ private int endId;
+
+ private String sourceTableName;
+ /**pk_control_*这类表都是为了保证重新汇总时保持ID一致，比如一个
+ * 新的项目来了之后，他的Url_MD5为***,然后程序会去pk_control_projects里
+ * 找该url_md5是否存在；若存在就取出对应的ID，取得的ID就是新项目存到汇总表的ID，
+ * 若不存在就在控制表里插入一条url_md5做为其记录
+ */
+ private String pkControlPostsTableName = "pk_control_posts";
+ private String pkControlProjectsTableName = "pk_control_projects";
+ //标签
+ private String taggingsTableName = "taggings";
+ private String tagsTableName = "tags";
+
+ private String gatherPostsTableName = "relative_memos";
+ private int maxId;
+
+ public void setParameters(Configure conf, String sourceTableName) {
+ this.conf = conf;
+ this.sourceTableName = sourceTableName;
+ }
+
+ // 读指针
+ public int readPointer(String table, String source, String target) {
+ int pointer = 1;
+ try {
+ //调用pointerDao对象读取数据库的pointer表获取上次汇总的位置
+ pointer = pointerDao.getPointer(table, source, target);
+ } catch (Exception e) {
+ // 表示表中没有数据
+ logger.info("No such pointer! Create one");
+ //表不存在的话就调用pointerDao中的函数创建pointer的表并插入1
+ pointerDao.insertPointer(table, source, target, 1);
+ }
+ return pointer;
+ }
+
+ @Override
+ public void run(){
+ //获取系统当年时间
+ long start = System.currentTimeMillis();
+
+ //设置当前进程数据源的表名
+ Thread.currentThread().setName(sourceTableName);
+ //idscrement为一个批处理的处理的数量即id一次增加的数量
+ idsIncrement = conf.getIdsIncrement();
+ //读取上次汇总结束的地址做为本次汇总开始的地址
+ idsBegin = readPointer(conf.getPointerTableName(), sourceTableName,
+ conf.getTargetTableName());
+ //读取数据来源表的数量
+ idsEnd = maxId = gatherDao.getMaxId(sourceTableName);
+ //idsEnd = maxId = 1000;
+ //如果开始的ID小于结束的id就从idsBegin开始汇总
+ while (idsBegin < idsEnd) {
+ beginId = idsBegin;
+ endId = beginId + idsIncrement - 1; // 取数据时两边都取等号
+ if (endId <= maxId) {
+ //如果上一批量结尾的id小于最大id说明汇总还没完成，就继续进行
+ handleBatchData(beginId, endId, conf);
+ idsBegin = idsBegin + idsIncrement;
+ } else {
+ endId = maxId; // endId应小于maxId
+ handleBatchData(beginId, endId, conf);
+ break;
+ }
+ }
+ GatherProcess.gatherState.put(sourceTableName, false);
+ long end = System.currentTimeMillis();
+ logger.info((end - start) / 6000);
+
+ }
+
+ @Transactional(propagation=Propagation.REQUIRED)
+ public void handleBatchData(int beginId, int endId, Configure conf) {
+ logger.info("BeginId#" + sourceTableName + ":" + beginId);
+ // 表示任务没有完成
+ int maxId = gatherDao.getMaxId(sourceTableName);
+ // 防止转移超过当前最大值的Id数据
+ if (beginId >= 0 && endId > 0 && maxId >= endId) {
+ // 更新执行开始时间
+ logger.info("begin gathering...");
+
+ // 插入Id段数据，忽略重复值
+ try {
+ String selectItems = getSelectItems(conf);
+ //这里的目标表是relative_memos
+ if (conf.getTargetTableName().equals("relative_memos")) {
+ List dataGet = gatherDao.getPostGatherData(sourceTableName, selectItems, beginId, endId,
+ conf.getAndWhere());
+ gatherPosts(dataGet);
+
+ } else
+ //这是对项目进行汇总的操作
+ if (conf.getTargetTableName().equals("gather_projects")) {
+ List dataGet = gatherDao.getPrjGatherData(sourceTableName, selectItems, beginId, endId,
+ conf.getAndWhere());
+ gatherProjects(dataGet);
+
+ } else {
+ List dataGet = gatherDao.getJobGatherData(
+ sourceTableName, selectItems, beginId, endId,
+ conf.getAndWhere());
+ gatehrRequirement(dataGet);
+ }
+
+ } catch (Exception ex) {
+ // 数据迁移过程可能发生异常情况
+ logger.error(ex);
+ System.exit(0);
+ }
+
+ // 更新游标到本次 EndId+1;
+ pointerDao.updatePointer(conf.getPointerTableName(),
+ sourceTableName, conf.getTargetTableName(), endId + 1);// sourceIdBegin
+ // +
+ // idsIncrement
+ logger.info("current--" + sourceTableName + ": " + endId);
+ }
+ }
+
+ // 处理URL不存在的帖子 插入relative_memos表
+ public void handleInsertGatherPosts(RelativeMemo model, Configure conf) {
+ try {
+ targetDao.insertRelativeMemo(conf.getTargetTableName(),
+ conf.getTargetFields(), model);
+ } catch (Exception e) {
+ logger.error(e);
+ }
+ }
+
+ // 处理URL相同的帖子更新 id表示更新的帖子固定id//更新url_md5相同的帖子
+ public void handleUpdateGatherPosts(int id, RelativeMemo model_new) {
+ targetDao.updateRelativeMemo(gatherPostsTableName, model_new, id);// 更新数据relative_memos表
+ }
+
+ // 处理URL不存在的项目 插入gather_projects表
+ public void handleInsertGatherProjects(GatherProject model, Configure conf) {
+ try {
+ targetDao.insertOpenSourceProject(conf.getTargetTableName(),
+ conf.getTargetFields(), model);
+ } catch (Exception e) {
+ logger.error(e);
+ }
+
+ }
+
+ // 处理URL相同的项目更新 id表示更新的项目固定id
+ public void handleUpdateGatherProjects(int id, GatherProject model_new) {
+ targetDao.updateOpenSourceProject(conf.getTargetTableName(), model_new,
+ id);// 更新数据gather_projects表
+ }
+
+ // 处理URL不存在的项目 插入job_requirements表
+ public void handleInsertGatherJobs(JobRequirement model, Configure conf) {
+ try {
+ targetDao.insertJobRequirement(conf.getTargetTableName(),
+ conf.getTargetFields(), model);
+ } catch (Exception e) {
+ logger.error(e);
+ }
+
+ }
+
+ // 处理URL相同的项目更新 id表示更新的项目固定id
+ public void handleUpdateGatherJobs(int id, JobRequirement model_new) {
+ targetDao.updateJobRequirement(conf.getTargetTableName(), model_new, id);// 更新数据job_requirements表
+ }
+
+ // 将tag和项目的关系存入表item_tag_relation 并分离tag
+ public void handleTags(RelativeMemo model,int postId)
+ {
+ // 将tag和项目的关系存入表item_tag_relation 并分离tag
+ String tags = model.getTags();
+ if(tags != null)
+ {
+ List tagList = DataHandler.tagsSegmentation(tags);
+ for (String tag : tagList) {
+ targetDao.insertTag(tagsTableName, tag);// ignore方式插入该项目的标签
+ int tag_id = targetDao.selectTagIdByName(
+ tagsTableName, tag);
+ Taggings taggings = new Taggings();
+ taggings.setTag_id(tag_id);
+ taggings.setTaggable_id(postId);
+ taggings.setTaggable_type("RelativeMemo");
+ taggings.setContext("tags");
+ taggings.setCreated_at(DataHandler.getNow());
+ // 将Taggings对象存入数据库中
+ try {
+ targetDao.insertTaggings(taggingsTableName,
+ taggings);
+ } catch (Exception e) {
+ // 在插入记录之前 relative_memos表中的记录已经被删除掉了
+ logger.error(e);
+ System.exit(0);
+ }
+ }
+ }
+ }
+
+ @Transactional(propagation=Propagation.REQUIRED)
+ public void gatherPosts(List dataGet){
+ for (int i = 0; i < dataGet.size(); i++) {
+ RelativeMemo model = dataGet.get(i);
+ String urlMD5 = model.getUrl_md5();// 通过urlMD5判断是不是已经存在该帖子、是否更新
+
+ int postId = 0;
+
+ //查找pk_control_posts表以判定此帖子是否已经存在
+ PKControlPosts pkControlModel = pkControlPostsDao.selectItemByUrlMD5(
+ pkControlPostsTableName, urlMD5);
+
+ //帖子已经存在则进行更新操作
+ if(pkControlModel != null){
+ model.setId(pkControlModel.getId());
+ try{
+ handleUpdateGatherPosts(pkControlModel.getId(), model);
+ }catch(Exception e){
+ logger.error("更新帖子时出错:" + e);
+ }
+ }
+ else{//帖子不存在就插入pk_control_posts表并为帖子生成唯一固定ID
+ pkControlPostsDao.insertOneItem(pkControlPostsTableName, urlMD5);
+ PKControlPosts controlItem = pkControlPostsDao.selectItemByUrlMD5(
+ pkControlPostsTableName, urlMD5);
+
+ model.setId(controlItem.getId());
+
+ try{
+ handleInsertGatherPosts(model, conf);
+ }catch(Exception e){
+ logger.error("插入帖子时出错:" + e);
+ }
+ }
+
+ postId = model.getId();
+ // 将tag和项目的关系存入表item_tag_relation 并分离tag
+ handleTags(model,postId);
+
+ }
+ }
+ @Transactional(propagation=Propagation.REQUIRED)
+ public void gatherProjects(List dataGet){
+ for (int i = 0; i < dataGet.size(); i++) {
+ GatherProject model = dataGet.get(i);
+ String urlMD5 = model.getUrl_md5();// 通过urlMD5判断是不是已经存在该项目、是否更新
+
+ int prjId = 0;
+
+ PKControlProjects pkControlProjects = pkControlProjectsDao.selectItemByUrlMD5(
+ pkControlProjectsTableName, urlMD5);//查看有没有固定的id
+
+ if(pkControlProjects != null){
+ model.setId(pkControlProjects.getId());
+ model.setUpdate_mark(1);
+
+ try{
+ handleUpdateGatherProjects(pkControlProjects.getId(), model);
+ }catch(Exception e){
+ logger.error("更新项目时出错:" + e);
+ }
+
+ }else{
+ // 在pk_control_posts表中生成当前项目对应的id
+ pkControlProjectsDao.insertOneItem(
+ pkControlProjectsTableName, urlMD5);
+ // 查看刚刚插入信息的id
+ PKControlProjects controlItem = pkControlProjectsDao.selectItemByUrlMD5(
+ pkControlProjectsTableName,urlMD5);
+ // 用id构造model对应的固定不变的id
+ model.setId(controlItem.getId());
+
+ model.setUpdate_mark(0);
+ try{
+ handleInsertGatherProjects(model, conf);
+ }catch(Exception e){
+ logger.error("插入项目时出错:" + e);
+ }
+ }
+
+ prjId = model.getId();
+ }
+ }
+
+ @Transactional(propagation=Propagation.REQUIRED)
+ public void gatehrRequirement(List dataGet){
+ for (int i = 0; i < dataGet.size(); i++) {
+ JobRequirement model = dataGet.get(i);
+ String urlMD5 = model.getUrl_md5();// 通过urlMD5判断是不是已经存在该帖子
+ // 是否更新
+ int postId = 0;
+
+ PKControlPosts pkControlPosts = pkControlPostsDao
+ .selectItemByUrlMD5(
+ pkControlPostsTableName, urlMD5);// 查看有没有固定的id
+
+ if (pkControlPosts != null){
+ model.setId(pkControlPosts.getId());
+
+ try{
+ handleUpdateGatherJobs(pkControlPosts.getId(), model);
+ }catch(Exception e){
+ logger.error("更新jobRequirement时出错:" + e);
+ }
+ }
+ else {
+ // 在pk_control_posts表中生成当前项目对应的id
+ pkControlPostsDao.insertOneItem(
+ pkControlPostsTableName, urlMD5);
+ // 查看刚刚插入信息的id
+ PKControlPosts controlItem = pkControlPostsDao
+ .selectItemByUrlMD5(
+ pkControlPostsTableName, urlMD5);
+ // 用id构造model对应的固定不变的id
+ model.setId(controlItem.getId());
+
+ try{
+ handleInsertGatherJobs(model, conf);
+ }catch(Exception e){
+ logger.error("插入jobRequirement时出错:"+ e);
+ }
+ }
+ postId = model.getId();
+ }
+ }
+
+ public String getSelectItems(Configure conf){
+ String[] sourceFields = conf.getSourceFields().split(",");
+ String[] targetFields = conf.getTargetFields().split(",");
+ String selectItems = "";
+ for (int i = 0; i < sourceFields.length; i++) {
+ String str_source = sourceFields[i];
+ String str_target = targetFields[i];
+ selectItems += str_source + " as " + str_target + ",";
+ }
+ selectItems = selectItems
+ .substring(0, selectItems.length() - 1) + " ";
+ return selectItems;
+ }
+
+}
diff --git a/gather_program/src/main/resource/applicationContext-myBatis.xml b/gather_program/src/main/resource/applicationContext-myBatis.xml
index 1eb00058d..abb8e8115 100644
--- a/gather_program/src/main/resource/applicationContext-myBatis.xml
+++ b/gather_program/src/main/resource/applicationContext-myBatis.xml
@@ -19,7 +19,7 @@
destroy-method="close">
+ value="jdbc:mysql://localhost:3306/ossean_gather?characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull&autoReconnect=true" />
@@ -34,12 +34,12 @@
-
+
+ value="jdbc:mysql://localhost:3306/ossean_gather?characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull&autoReconnect=true" />
@@ -53,7 +53,7 @@
-
+
diff --git a/gather_program/src/main/resource/gather_projects.xml b/gather_program/src/main/resource/gather_projects.xml
index 5bdbd1a06..77ff53b05 100644
--- a/gather_program/src/main/resource/gather_projects.xml
+++ b/gather_program/src/main/resource/gather_projects.xml
@@ -3,7 +3,7 @@
TableFlow
pointers
- oschina_project,openhub_project
+ sourceforge_project,openhub_project
gather_projects
id,name,tags,url,url_md5,description,language,source,license,homepage,now(),extracted_time,created_time