TaskProducerColumnFocus.java
2.93 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
/**@author Sherlock_yb
* Created time:2014年12月2日
*/
package nlp.whu.utils;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.TimeUnit;
import org.apache.log4j.Logger;
import nlp.whu.focus.FocusPool;
import nlp.whu.model.SubjectDetail;
import nlp.whu.model.TaskProducer;
/**@description：
*/
public class TaskProducerColumnFocus implements TaskProducer{
private static final Logger log = Log4jUtil.getInstance().getLogger(TaskProducerSubjectFocus.class);
private static final long ONE_MINUTE_MS = TimeUnit.MINUTES.toMillis(1L);
private PriorityBlockingQueue<TaskWrapperColumnFocus> taskQueue;
private Date startDate;
private int subjectAnalysisInteval = LoadFocusPath.subjectAnalysisInteval;
private boolean initialStart = true;
public TaskProducerColumnFocus(PriorityBlockingQueue<TaskWrapperColumnFocus> taskQueue
, Date startDate){
this.taskQueue = taskQueue;
this.startDate = startDate;
}
public void wrapperTask(){
if(initialStart){
log.info("wrapperTask initial start, construct initial task list.");
List<SubjectDetail> sdList = FocusPool.subjectDetailDao.getSubjectDetailList();
Map<Integer, Set<Integer>> userSubjectTypesMap = FocusPool.subjectDetailDao.getUserSubjectTypesMap(sdList);
for(Integer userId : userSubjectTypesMap.keySet()){
Set<Integer> subjectTypes = userSubjectTypesMap.get(userId);
for(Integer subjectTypeId : subjectTypes){
TaskWrapperColumnFocus tw = new TaskWrapperColumnFocus();
tw.setUserId(userId);
tw.setSubjectTypeId(subjectTypeId);
addTask(tw);
}
}
initialStart = false;
}else{
long now = System.currentTimeMillis();
long startTime = startDate.getTime();
long gap = (now - startTime) / (ONE_MINUTE_MS);
if(gap > 0 && gap % subjectAnalysisInteval == 0){
List<SubjectDetail> sdList = FocusPool.subjectDetailDao.getSubjectDetailList();
Map<Integer, Set<Integer>> userSubjectTypesMap = FocusPool.subjectDetailDao.getUserSubjectTypesMap(sdList);
for(Integer userId : userSubjectTypesMap.keySet()){
Set<Integer> subjectTypes = userSubjectTypesMap.get(userId);
for(Integer subjectTypeId : subjectTypes){
TaskWrapperColumnFocus tw = new TaskWrapperColumnFocus();
tw.setUserId(userId);
tw.setSubjectTypeId(subjectTypeId);
addTask(tw);
}
}
}
}
}
public void addTask(TaskWrapperColumnFocus taskWrapper){
taskQueue.put(taskWrapper);
log.info("add one task, info: " + taskWrapper);
}
@Override
public void run() {
while(true){
wrapperTask();
try {
//gap的单位为min，故只能每1min检测一次当前系统时间
log.info("try to sleep 1 min, check the system time every 1 min.");
TimeUnit.MINUTES.sleep(1);
} catch (InterruptedException e) {
log.info("stop sleep", e);
}
}
}
}