kevinli
/
educoder
0
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

add gatherthreadnew

This commit is contained in:
lqk 2016-12-08 11:17:29 +08:00
parent c028943de6
commit 2da3f2ae83
6 changed files with 412 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
gather_program/bin/resources/applicationContext-myBatis.xml
View File

@ -39,9 +39,9 @@
destroy-method="close">
<property name="driverClassName" value="com.mysql.jdbc.Driver" />
<property name="url"
value="jdbc:mysql://172.16.128.36:3306/ossean_production?characterEncoding=UTF-8&amp;zeroDateTimeBehavior=convertToNull&amp;autoReconnect=true" />
value="jdbc:mysql://172.16.128.30:3306/ossean_production?characterEncoding=UTF-8&amp;zeroDateTimeBehavior=convertToNull&amp;autoReconnect=true" />
<property name="username" value="gather" />
<property name="password" value="Influx@1234" />
<property name="password" value="influx1234" />
<property name="validationQuery" value="SELECT 1" />
<property name="testOnBorrow" value="true"/>
</bean>

2
gather_program/sql/pk_control_projects.sql
View File

@ -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;

10
gather_program/src/main/java/org/ossean/gather/process/GatherProcess.java
View File

@ -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");

398
gather_program/src/main/java/org/ossean/gather/process/GatherThreadNew.java Normal file
View File

@ -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<RelativeMemo> dataGet = gatherDao.getPostGatherData(sourceTableName, selectItems, beginId, endId,
conf.getAndWhere());
gatherPosts(dataGet);
} else
//这是对项目进行汇总的操作
if (conf.getTargetTableName().equals("gather_projects")) {
List<GatherProject> dataGet = gatherDao.getPrjGatherData(sourceTableName, selectItems, beginId, endId,
conf.getAndWhere());
gatherProjects(dataGet);
} else {
List<JobRequirement> 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<String> 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<RelativeMemo> 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<GatherProject> 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<JobRequirement> 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;
}
}

8
gather_program/src/main/resource/applicationContext-myBatis.xml
View File

@ -19,7 +19,7 @@
destroy-method="close">
<property name="driverClassName" value="com.mysql.jdbc.Driver" />
<property name="url"
value="jdbc:mysql://localhost:3306/ossean_production?characterEncoding=UTF-8&amp;zeroDateTimeBehavior=convertToNull&amp;autoReconnect=true" />
value="jdbc:mysql://localhost:3306/ossean_gather?characterEncoding=UTF-8&amp;zeroDateTimeBehavior=convertToNull&amp;autoReconnect=true" />
<property name="username" value="root" />
<property name="password" value="123456" />
<property name="validationQuery" value="SELECT 1" />
@ -34,12 +34,12 @@
</bean>
<!--DestData -->
<!--DestData -->
<bean id="dataSourceTwo" class="org.apache.commons.dbcp.BasicDataSource"
destroy-method="close">
<property name="driverClassName" value="com.mysql.jdbc.Driver" />
<property name="url"
value="jdbc:mysql://localhost:3306/ossean_production?characterEncoding=UTF-8&amp;zeroDateTimeBehavior=convertToNull&amp;autoReconnect=true" />
value="jdbc:mysql://localhost:3306/ossean_gather?characterEncoding=UTF-8&amp;zeroDateTimeBehavior=convertToNull&amp;autoReconnect=true" />
<property name="username" value="root" />
<property name="password" value="123456" />
<property name="validationQuery" value="SELECT 1" />
@ -53,7 +53,7 @@
<property name="sqlSessionFactory" ref="sqlSessionFactoryTwo"></property>
</bean>
<tx:annotation-driven transaction-manager="transactionManager"/>
<tx:annotation-driven transaction-manager="transactionManager" proxy-target-class="true"/>
<bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
<property name="dataSource" ref="dataSourceOne" />
</bean>

2
gather_program/src/main/resource/gather_projects.xml
View File

@ -3,7 +3,7 @@
<properties>
<comment>TableFlow</comment>
<entry key="pointerTableName">pointers</entry>
<entry key="sourceTableName">oschina_project,openhub_project</entry>
<entry key="sourceTableName">sourceforge_project,openhub_project</entry>
<entry key="targetTableName">gather_projects</entry>
<entry key="sourceFields">id,name,tags,url,url_md5,description,language,source,license,homepage,now(),extracted_time,created_time</entry>
<!-- <entry key="targetFields">url,crawled_time,tags,license,name,description,language,platform,source,registered_time,urlMD5</entry> -->