Skip to content

M
messageflow
Commit ed96a533 by shentao
Options

2018/11/19 稳定版：追踪规则+消息流灵活上限1.0.0版

parent 22a3d9e4
Showing with 279 additions and 318 deletions
src/main/java/com/zhiwei/messageflow/ES4RedisStart.java
......@@ -29,12 +29,16 @@ public class ES4RedisStart {
@Autowired
private ES4RedisTask es4RedisTask;
@Autowired
private ESGetCommonId esGetCommonId;
/**
* 启动线程
*/
public void startThread() throws JsonParseException, JsonMappingException, IOException, InterruptedException {
//更新commonid
esGetCommonId.getCommonId();
// 项目列表
List<Project> projects = projectDao.getAllProjects();
......@@ -44,7 +48,7 @@ public class ES4RedisStart {
// 遍历项目
for (Project project : projects) {
// if(!project.getProjectName().equals("今日头条")) {
// if(!project.getProjectName().equals("测试")) {
// continue;
// }
......
src/main/java/com/zhiwei/messageflow/ES4RedisTask.java
......@@ -97,29 +97,6 @@ public class ES4RedisTask {
// 项目预警规则列表
List<TrackRule> trackRules = trackRuleDao.getTrackRuleByProject(project.getProjectName());
/*
* //测试预警规则 List<TrackRule> trackRules=new ArrayList<>(); TrackRule tr1=new
* TrackRule(); TrackRule tr2=new TrackRule(); TrackRule tr3=new TrackRule();
* Date date = new Date(); Long createAt=date.getTime();
* tr1.setCreateAt(createAt); tr1.setAndOr(""); tr1.setWarn(false);
* tr1.setRuleType("keyWords"); tr1.setSubmitter("虞诚毅");
* tr1.setKeyWordsInputOne("知乎"); tr1.setRuleName("测试规则1");
* tr1.setRuleExplain("测试说明1"); tr1.setEarlyWarningTime("三天");
* tr1.setProject("腾讯"); tr1.setEarlyWarning("email");
* tr2.setCreateAt(createAt); tr2.setAndOr("或"); tr2.setWarn(false);
* tr2.setRuleType("keyWords"); tr2.setSubmitter("虞诚毅");
* tr2.setKeyWordsInputOne("知乎"); tr2.setKeyWordsInputTwo("微博");
* tr2.setRuleName("测试规则2"); tr2.setRuleExplain("测试说明3");
* tr2.setEarlyWarningTime("三天"); tr2.setProject("腾讯");
* tr2.setEarlyWarning("email"); tr3.setCreateAt(createAt); tr3.setAndOr("且");
* tr3.setWarn(false); tr3.setRuleType("keyWords"); tr3.setSubmitter("虞诚毅");
* tr3.setKeyWordsInputOne("知乎"); tr3.setKeyWordsInputTwo("问题");
* tr3.setRuleName("测试规则2"); tr3.setRuleExplain("测试说明3");
* tr3.setEarlyWarningTime("三天"); tr3.setProject("腾讯");
* tr3.setEarlyWarning("wechat"); trackRules.add(tr1); trackRules.add(tr2);
* trackRules.add(tr3);
*/
// 遍历平台
for (PlatformNew platform : platformNames) {
......@@ -242,7 +219,7 @@ public class ES4RedisTask {
} // 遍历关键词组
// 预警
earlyWarningService.earlyWarningNew(trackRules, count, allstartrsid, allrsid, platform,
earlyWarningService.earlyWarningNew(messages,trackRules, count, allstartrsid, allrsid, platform,
project.getProjectName());
// 向redis库中存储新的rsid Map,覆盖原有数据
......
src/main/java/com/zhiwei/messageflow/ESGetCommonId.java
......@@ -2,6 +2,7 @@ package com.zhiwei.messageflow;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import org.springframework.beans.factory.annotation.Autowired;
......@@ -15,30 +16,31 @@ public class ESGetCommonId {
@Autowired
private ESDao esDao;
@Autowired
private EventService eventService;
public static int START_COMMONID;
public static long START_COMMONID;
public static long TIME;
public void getCommonId() {
long nowtime = System.currentTimeMillis();
if (TIME < nowtime - 60 * 1000L) {
if (TIME < nowtime - 60 * 60 * 1000L) {
List<Integer> cidList = new ArrayList<>();
int weibo = esDao.get24hoursFirstCommonid("weibo");
int weibo = esDao.get48hoursFirstCommonid("weibo");
cidList.add(weibo);
int media = esDao.get24hoursFirstCommonid("media");
int media = esDao.get48hoursFirstCommonid("media");
cidList.add(media);
int video = esDao.get24hoursFirstCommonid("video");
int video = esDao.get48hoursFirstCommonid("video");
cidList.add(video);
int zhihu = esDao.get24hoursFirstCommonid("zhihu");
int zhihu = esDao.get48hoursFirstCommonid("zhihu");
cidList.add(zhihu);
Collections.sort(cidList);
System.err.println(weibo + " " + media + " " + video + " " + zhihu + "\n" + cidList);
START_COMMONID = cidList.get(0);
TIME = nowtime;
System.err.println("START_COMMONID:" + START_COMMONID + "\tNOW:" + new Date(TIME));
}
}
......@@ -55,7 +57,8 @@ public class ESGetCommonId {
int zhihu = esDao.getCommonidByTime("zhihu", endtime);
cidList.add(zhihu);
Collections.sort(cidList);
// System.err.println(weibo + " " + media + " " + video + " " + zhihu + "\n" + cidList);
// System.err.println(weibo + " " + media + " " + video + " " + zhihu + "\n" +
// cidList);
int commonid = cidList.get(0);
return commonid;
}
......
src/main/java/com/zhiwei/messageflow/config/RedisConfig.java
......@@ -31,6 +31,8 @@ public class RedisConfig {
public static final String EVENTLISTKEY = "Event:EventList";
/** 事件采集去重url集 **/
public static final String EVENTHASHKEY = "Event:Hash:";
/** 追踪规则已追踪集 **/
public static final String TRACKKEY = "TrackRule:";
private int keyMaxSize;
......
src/main/java/com/zhiwei/messageflow/es/dao/ESDao.java
......@@ -101,7 +101,7 @@ public interface ESDao {
public SearchHits getDataFromEs(List<String> keywords, int count, long startid, long endid, PlatformNew platform,
String project, String matchFields);
public int get24hoursFirstCommonid(String tableName);
public int get48hoursFirstCommonid(String tableName);
public int getCommonidByTime(String tableName, long endtime);
/**
......
src/main/java/com/zhiwei/messageflow/es/dao/impl/ESDaoImpl.java
......@@ -725,14 +725,17 @@ public class ESDaoImpl implements ESDao {
}
@Override
public int get24hoursFirstCommonid(String tableName) {
public int get48hoursFirstCommonid(String tableName) {
// 获取当天最早commonid
int commonid = 0;
long date = System.currentTimeMillis() - 1 * 24 * 60 * 60 * 1000L;
String weibotime = TimeUtil.formatDateToMinute(new Date(date));
String zhihutime = TimeUtil.formatEsDate(new Date(date - 8 * 3600 * 1000L)) + "Z";
String vmtime = TimeUtil.formatEsDate(new Date(date - 8 * 3600 * 1000L));
long now =System.currentTimeMillis();
long date = now - 1 * 48 * 60 * 60 * 1000L;
// String wEtime = TimeUtil.formatDateToMinute(new Date(now));
// String weibotime = TimeUtil.formatDateToMinute(new Date(date));
// String zEtime = TimeUtil.formatEsDate(new Date(now - 8 * 3600 * 1000L)) + "Z";
// String zhihutime = TimeUtil.formatEsDate(new Date(date - 8 * 3600 * 1000L)) + "Z";
// String vmEtime = TimeUtil.formatEsDate(new Date(now - 8 * 3600 * 1000L));
// String vmtime = TimeUtil.formatEsDate(new Date(date - 8 * 3600 * 1000L));
// weibo
if (tableName.equals("weibo")) {
List<String> list = indexes.getLastIndexes("network", 2);
......@@ -741,7 +744,7 @@ public class ESDaoImpl implements ESDao {
.setQuery(QueryBuilders.matchAllQuery()).setFetchSource(new String[] { "commonid" }, null)
.addSort("commonid", SortOrder.ASC).setFrom(0).setSize(1);
searchRequestBuilder.setPostFilter(QueryBuilders.rangeQuery("time").from(weibotime));
searchRequestBuilder.setPostFilter(QueryBuilders.rangeQuery("savetime").from(date).to(now));
SearchResponse response = searchRequestBuilder.execute().actionGet();
SearchHits searchHits = response.getHits();
......@@ -754,7 +757,7 @@ public class ESDaoImpl implements ESDao {
.setQuery(QueryBuilders.matchAllQuery()).setFetchSource(new String[] { "commonid" }, null)
.addSort("commonid", SortOrder.ASC).setFrom(0).setSize(1);
searchRequestBuilder.setPostFilter(QueryBuilders.rangeQuery("time").from(vmtime));
searchRequestBuilder.setPostFilter(QueryBuilders.rangeQuery("savetime").from(date).to(now));
SearchResponse response = searchRequestBuilder.execute().actionGet();
SearchHits searchHits = response.getHits();
......@@ -767,7 +770,7 @@ public class ESDaoImpl implements ESDao {
.setQuery(QueryBuilders.matchAllQuery()).setFetchSource(new String[] { "commonid" }, null)
.addSort("commonid", SortOrder.ASC).setFrom(0).setSize(1);
searchRequestBuilder.setPostFilter(QueryBuilders.rangeQuery("time").from(vmtime));
searchRequestBuilder.setPostFilter(QueryBuilders.rangeQuery("savetime").from(date).to(now));
SearchResponse response = searchRequestBuilder.execute().actionGet();
SearchHits searchHits = response.getHits();
......@@ -778,7 +781,7 @@ public class ESDaoImpl implements ESDao {
.setQuery(QueryBuilders.matchAllQuery()).setFetchSource(new String[] { "commonid" }, null)
.addSort("commonid", SortOrder.ASC).setFrom(0).setSize(1);
searchRequestBuilder.setPostFilter(QueryBuilders.rangeQuery("insert_at").from(zhihutime));
searchRequestBuilder.setPostFilter(QueryBuilders.rangeQuery("savetime").from(date).to(now));
SearchResponse response = searchRequestBuilder.execute().actionGet();
SearchHits searchHits = response.getHits();
......
src/main/java/com/zhiwei/messageflow/es/service/EarlyWarningService.java
......@@ -2,6 +2,7 @@ package com.zhiwei.messageflow.es.service;
import java.util.List;
import com.alibaba.fastjson.JSONObject;
import com.zhiwei.messageflow.mongo.bean.PlatformNew;
import com.zhiwei.messageflow.mongo.bean.TrackRule;
......@@ -9,6 +10,7 @@ public interface EarlyWarningService {
/**
* 预警
* @param messages
*
* @param trackRules
* 预警规则
......@@ -23,9 +25,6 @@ public interface EarlyWarningService {
* @param projectName
* 项目名
*/
void earlyWarning(List<TrackRule> trackRules, int count, long startid, long endid, String platformName,
String projectName);
void earlyWarningNew(List<TrackRule> trackRules, int count, long startid, Long endid, PlatformNew platform,
void earlyWarningNew(List<JSONObject> messages, List<TrackRule> trackRules, int count, long startid, Long endid, PlatformNew platform,
String projectName);
}
src/main/java/com/zhiwei/messageflow/es/service/TrackHitAndWarnService.java
package com.zhiwei.messageflow.es.service;
import org.elasticsearch.search.SearchHits;
import com.zhiwei.messageflow.mongo.bean.PlatformNew;
import com.alibaba.fastjson.JSONObject;
import com.zhiwei.messageflow.mongo.bean.TrackRule;
public interface TrackHitAndWarnService {
/**
* 关键词预警数据获取
*
* @param truckRule
* @param count
* @param startid
* @param endid
* @param platformName
* @return
*/
SearchHits keyWordsTrackHit(TrackRule truckRule, int count, long startid, long endid, String platformName);
/**
* 渠道预警数据获取
*
* @param truckRule
* @param count
* @param startid
* @param endid
* @param platformName
* @return
* 判断该条追踪规则是否处于预警状态
* @Title: isWarnTrackrule
* @Description: 判断该条追踪规则是否处于预警状态
* @param @param tr
* @param @return 设定文件
* @return boolean 返回类型
*/
SearchHits channelTrackHit(TrackRule truckRule, int count, long startid, long endid, String platformName);
boolean isWarnTrackrule(TrackRule trackRule);
/**
* 相似新闻数预警数据获取
*
* @param truckRule
* @param count
* @param startid
* @param endid
* @param platformName
* @return
* 使用单条追踪规则
* @Title: useTrackrule
* @Description: 使用单条追踪规则
* @param @param trackrule
* @param @return 设定文件
* @return boolean 返回类型
*/
SearchHits articleTrackHit(TrackRule truckRule, int count, long startid, long endid, String platformName);
boolean endTrackrule(TrackRule trackrule);
/**
* 邮件预警
*
* @param trackHits
* @param platformName
* @param trackRule
* 是否命中预警规则
* @param platformName
* @Title: ishitWarnMsg
* @Description:是否命中预警规则
* @param @param msg
* @param @param tr 设定文件
* @return void 返回类型
*/
void WarnEmail(SearchHits trackHit, String platformName, TrackRule trackRule, String projectName);
boolean ishitWarnMsg(JSONObject msg, String platformName, TrackRule tr);
/**
* 微信预警
*
* @param trackHits
* @param platformName
* @param trackRule
* 按不同规则预警
* @Title: warnMsg
* @Description: 按不同规则预警
* @param @param msg
* @param @param tr 设定文件
* @return void 返回类型
*/
void WarnWechat(SearchHits trackHit, String platformName, TrackRule trackRule, String projectName);
SearchHits keyWordsTrackHitNew(TrackRule trackRule, int count, long startid, Long endid, PlatformNew platform);
SearchHits channelTrackHitNew(TrackRule trackRule, int count, long startid, Long endid, PlatformNew platform);
SearchHits articleTrackHitNew(TrackRule trackRule, int count, long startid, Long endid, PlatformNew platform);
void WarnWechatNew(SearchHits trackHit, String platformName, TrackRule trackRule, String projectName);
void WarnEmailNew(SearchHits trackHit, String platformName, TrackRule trackRule, String projectName);
void warnMsg(JSONObject msg, TrackRule tr);
}
src/main/java/com/zhiwei/messageflow/es/service/impl/EarlyWarningServiceImpl.java
......@@ -2,176 +2,37 @@ package com.zhiwei.messageflow.es.service.impl;
import java.util.List;
import org.elasticsearch.search.SearchHits;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import com.alibaba.fastjson.JSONObject;
import com.zhiwei.messageflow.es.service.EarlyWarningService;
import com.zhiwei.messageflow.es.service.TrackHitAndWarnService;
import com.zhiwei.messageflow.mongo.bean.PlatformNew;
import com.zhiwei.messageflow.mongo.bean.TrackRule;
import com.zhiwei.messageflow.mongo.dao.TrackRuleDao;
@Component
public class EarlyWarningServiceImpl implements EarlyWarningService {
@Autowired
private TrackRuleDao trackRuleDao;
@Autowired
private TrackHitAndWarnService trackHitAndWarnService;
@Override
public void earlyWarning(List<TrackRule> trackRules, int count, long startid, long endid, String platformName,
String projectName) {
for (TrackRule trackRule : trackRules) {
// 判断是否预警
if (trackRule.isWarn()) {
continue;
}
/**
* 判断是否过期
*/
long createAt = trackRule.getCreateAt();
long addDay = 0;
// if (trackRule.getEarlyWarningTime().equals("一天")) {
// addDay = 24 * 60 * 60 * 1000L;
// } else if (trackRule.getEarlyWarningTime().equals("二天")) {
// addDay = 2 * 24 * 60 * 60 * 1000L;
// } else if (trackRule.getEarlyWarningTime().equals("三天")) {
// addDay = 3 * 24 * 60 * 60 * 1000L;
// }
long confirm = System.currentTimeMillis()+1L;//createAt + addDay;
if (confirm <= System.currentTimeMillis()) {
// 过期
trackRule.setWarn(true);
trackRuleDao.updateTrackrule(trackRule.get_id());
} else {
// 判断规则
// 首先判断预警方式
if (trackRule.getEarlyWarning().equals("no")) {
} else if (trackRule.getEarlyWarning().equals("wechat")) {
// 微信预警
// 处理规则
SearchHits TrackHit = null;
if (trackRule.getRuleType().equals("keyWords")) {
// 关键词追踪
TrackHit = trackHitAndWarnService.keyWordsTrackHit(trackRule, count, startid, endid,
platformName);
} else if (trackRule.getRuleType().equals("channel")) {
// 渠道追踪
TrackHit = trackHitAndWarnService.channelTrackHit(trackRule, count, startid, endid,
platformName);
} else if (trackRule.getRuleType().equals("article")) {
// 相似新闻数追踪
TrackHit = trackHitAndWarnService.articleTrackHit(trackRule, count, startid, endid,
platformName);
}
trackHitAndWarnService.WarnWechat(TrackHit, platformName, trackRule, projectName);
} else if (trackRule.getEarlyWarning().equals("email")) {
// 邮箱预警
// 处理规则
SearchHits TrackHit = null;
if (trackRule.getRuleType().equals("keyWords")) {
// 关键词追踪
TrackHit = trackHitAndWarnService.keyWordsTrackHit(trackRule, count, startid, endid,
platformName);
} else if (trackRule.getRuleType().equals("channel")) {
// 渠道追踪
TrackHit = trackHitAndWarnService.channelTrackHit(trackRule, count, startid, endid,
platformName);
} else if (trackRule.getRuleType().equals("article")) {
// 相似新闻数追踪
TrackHit = trackHitAndWarnService.articleTrackHit(trackRule, count, startid, endid,
platformName);
}
trackHitAndWarnService.WarnEmail(TrackHit, platformName, trackRule, projectName);
}
}
}
}
@Override
public void earlyWarningNew(List<TrackRule> trackRules, int count, long startid, Long endid, PlatformNew platform,
String projectName) {
public void earlyWarningNew(List<JSONObject> messages, List<TrackRule> trackRules, int count, long startid,
Long endid, PlatformNew platform, String projectName) {
String platformName = platform.getPlatformName();
for (TrackRule trackRule : trackRules) {
// 判断是否预警
if (trackRule.isWarn()) {
continue;
}
/**
* 判断是否过期
*/
long createAt = trackRule.getCreateAt();
long addDay = 0;
// if (trackRule.getEarlyWarningTime().equals("一天")) {
// addDay = 24 * 60 * 60 * 1000L;
// } else if (trackRule.getEarlyWarningTime().equals("二天")) {
// addDay = 2 * 24 * 60 * 60 * 1000L;
// } else if (trackRule.getEarlyWarningTime().equals("三天")) {
// addDay = 3 * 24 * 60 * 60 * 1000L;
// } else if (trackRule.getEarlyWarningTime().equals("一月")) {
// addDay = 30 * 24 * 60 * 60 * 1000L;
// }
long confirm = System.currentTimeMillis()+1L;//createAt + addDay;
if (confirm <= System.currentTimeMillis()) {
// 过期
trackRule.setWarn(true);
trackRuleDao.updateTrackrule(trackRule.get_id());
} else {
// 判断规则
// 首先判断预警方式
if (trackRule.getEarlyWarning().equals("no")) {
} else if (trackRule.getEarlyWarning().equals("wechat")) {
// 微信预警
// 处理规则
SearchHits TrackHit = null;
if (trackRule.getRuleType().equals("keyWords")) {
// 关键词追踪
TrackHit = trackHitAndWarnService.keyWordsTrackHitNew(trackRule, count, startid, endid,
platform);
} else if (trackRule.getRuleType().equals("channel")) {
// 渠道追踪
TrackHit = trackHitAndWarnService.channelTrackHitNew(trackRule, count, startid, endid,
platform);
} else if (trackRule.getRuleType().equals("article")) {
// 相似新闻数追踪
TrackHit = trackHitAndWarnService.articleTrackHitNew(trackRule, count, startid, endid,
platform);
}
trackHitAndWarnService.WarnWechatNew(TrackHit,platformName, trackRule, projectName);
} else if (trackRule.getEarlyWarning().equals("email")) {
// 邮箱预警
// 处理规则
SearchHits TrackHit = null;
if (trackRule.getRuleType().equals("keyWords")) {
// 关键词追踪
TrackHit = trackHitAndWarnService.keyWordsTrackHitNew(trackRule, count, startid, endid,
platform);
} else if (trackRule.getRuleType().equals("channel")) {
// 渠道追踪
TrackHit = trackHitAndWarnService.channelTrackHitNew(trackRule, count, startid, endid,
platform);
} else if (trackRule.getRuleType().equals("article")) {
// 相似新闻数追踪
TrackHit = trackHitAndWarnService.articleTrackHitNew(trackRule, count, startid, endid,
platform);
}
trackHitAndWarnService.WarnEmailNew(TrackHit,platformName, trackRule, projectName);
}
for (int i = 0; i < messages.size(); i++) {
JSONObject msg = messages.get(i);
for (TrackRule tr : trackRules) {
// 判断是否预警
if (!trackHitAndWarnService.isWarnTrackrule(tr))
continue;
// 是否命中预警规则
boolean ishit = trackHitAndWarnService.ishitWarnMsg(msg, platformName, tr);
// 按不同预警规则预警
if (ishit)
trackHitAndWarnService.warnMsg(msg, tr);
}
}
......
src/main/java/com/zhiwei/messageflow/es/service/impl/HighLightFillingServiceImpl.java
......@@ -420,7 +420,6 @@ public class HighLightFillingServiceImpl implements HighLightFillingService {
message = mapper.writeValueAsString(sourceHitMap);
res = JSONObject.parseObject(message);
} catch (JsonProcessingException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
String type = searchHit.getType();
......
src/main/java/com/zhiwei/messageflow/es/service/impl/NoiseProcessingServiceImpl.java
......@@ -3,13 +3,9 @@ package com.zhiwei.messageflow.es.service.impl;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.TimeZone;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.elasticsearch.search.SearchHit;
......@@ -18,7 +14,6 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import com.alibaba.fastjson.JSONObject;
import com.zhiwei.messageflow.DirectES4RedisThread;
import com.zhiwei.messageflow.bean.MediaMessage;
import com.zhiwei.messageflow.bean.VideoMessage;
import com.zhiwei.messageflow.bean.WeiboMessage;
......@@ -28,7 +23,6 @@ import com.zhiwei.messageflow.es.service.NoiseProcessingService;
import com.zhiwei.messageflow.mongo.bean.NoiseRule;
import com.zhiwei.messageflow.mongo.bean.PlatformNew;
import com.zhiwei.messageflow.mongo.dao.NoiseRuleDao;
import com.zhiwei.messageflow.util.ESQueryUtil;
@Component
public class NoiseProcessingServiceImpl implements NoiseProcessingService {
......@@ -651,7 +645,7 @@ public class NoiseProcessingServiceImpl implements NoiseProcessingService {
try {
date = dateFormat.parse(time);
if ((date.getTime() + (8 * 3600 * 1000L)) < zero) {
log.info(date.getTime() + (8 * 3600 * 1000L)+"|"+zero);
// log.info(date.getTime() + (8 * 3600 * 1000L)+"|"+zero);
return true;
}
} catch (ParseException e) {
......@@ -681,7 +675,7 @@ public class NoiseProcessingServiceImpl implements NoiseProcessingService {
String time = map.get("time") != null ? map.get("time").toString() : null;
if (isTimeOver(time)) {
log.info("weibo:time:" + time +"\t"+ map.get("id").toString());
// log.info("weibo:time:" + time +"\t"+ map.get("id").toString());
return true;
}
String text = map.get("text") != null ? map.get("text").toString() : null;
......@@ -786,7 +780,7 @@ public class NoiseProcessingServiceImpl implements NoiseProcessingService {
String platform = platformNew.getPlatformName();
String time = map.get("created_at") != null ? map.get("created_at").toString() : null;
if (isTimeOver(time)) {
log.info("zhihu:time:" + time +"\t"+ map.get("id").toString());
// log.info("zhihu:time:" + time +"\t"+ map.get("id").toString());
return true;
}
String text = map.get("question_title") != null ? map.get("question_title").toString() : null;
......@@ -917,7 +911,7 @@ public class NoiseProcessingServiceImpl implements NoiseProcessingService {
String platform = platformNew.getPlatformName();
String time = map.get("time") != null ? map.get("time").toString() : null;
if (isTimeOver(time)) {
log.info("video:time:" + time +"\t"+ map.get("id").toString());
// log.info("video:time:" + time +"\t"+ map.get("id").toString());
return true;
}
String text = map.get("title") != null ? map.get("title").toString() : null;
......@@ -1047,7 +1041,7 @@ public class NoiseProcessingServiceImpl implements NoiseProcessingService {
String platform = platformNew.getPlatformName();
String time = map.get("time") != null ? map.get("time").toString() : null;
if (isTimeOver(time)) {
log.info("media:time:" + time +"\t"+ map.get("id").toString());
// log.info("media:time:" + time +"\t"+ map.get("id").toString());
return true;
}
String text = map.get("title") != null ? map.get("title").toString() : null;
......
src/main/java/com/zhiwei/messageflow/mongo/bean/TrackRule.java
package com.zhiwei.messageflow.mongo.bean;
import java.util.List;
import java.util.Map;
import org.springframework.data.annotation.Id;
import org.springframework.data.mongodb.core.mapping.Document;
......@@ -10,25 +13,74 @@ import lombok.ToString;
@ToString
@Document(collection = "qbjc_trackrule")
public class TrackRule {
/**
* id用户id
*/
@Id
private long _id;
private String _id;
/**
* ruleName 规则名
*/
private String ruleName;
private String ruleExplain;
/**
* ruleType 规则类型
*/
private String ruleType;
private String KeyWordsInputOne;
private String andOr;
private String KeyWordsInputTwo;
private String earlyWarning;
/**
* keyword 关键词：xxx xx，xx xxx
*/
private String keyword;
/**
* channel List<Map> 规则内容 渠道：平台 渠道名 主键
*/
private List<Map<String, Object>> channel;
/**
* warnRule 预警规则 （normal wechat）
*/
private String warnRule;
/**
* startTime 预警起始时间
*/
private Long startTime;
/**
* endTime long 预警结束时间
*/
private Long endTime;
/**
* highlighted String 高亮颜色
*/
private String highlighted;
private String earlyWarningTime;
private boolean isWarn;
private long createAt;
/**
* createAt long 创建时间
*/
private Long createAt;
/**
* updateAt long 更新时间
*/
private Long updateAt;
/**
* submitter String 提交人
*/
private String submitter;
/**
* isUsed boolean 是否使用
*/
private Boolean isUsed;
/**
* project String 所属项目
*/
private String project;
private String channelPt;
private String channelQd;
private String articleTitle;
private String articleNum;
private String timeRange;
/**
* warnLimit int 预警上限
*/
private Integer warnLimit;
/**
* WarnNum int 本次预警条数
*/
private Integer warnNum;
/**
* hasTrackNum int 已追踪总条数
*/
private Integer hasTrack;
}
src/main/java/com/zhiwei/messageflow/mongo/bean/WechatCode.java
......@@ -10,7 +10,7 @@ import lombok.ToString;
@Data
@ToString
@Document(collection = "wechatCode")
@Document(collection = "wechatCodeNew")
public class WechatCode {
@Id
private String _id;
......
src/main/java/com/zhiwei/messageflow/mongo/dao/TrackRuleDao.java
......@@ -2,6 +2,9 @@ package com.zhiwei.messageflow.mongo.dao;
import java.util.List;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import com.zhiwei.messageflow.mongo.bean.TrackRule;
public interface TrackRuleDao {
......@@ -15,9 +18,22 @@ public interface TrackRuleDao {
List<TrackRule> getTrackRuleByProject(String projectName);
/**
* 更新项目预警情况
*
* @param trackRule_id
* 更新单条追踪规则按条件和更新条件
* @Title: updateTrackrule
* @Description: 更新单条追踪规则
* @param @param query
* @param @param update
* @param @return 设定文件
* @return boolean 返回类型
*/
boolean updateTrackrule(Query query, Update update);
/**
* 按id去追踪规则
* @Title: getTrackRuleById
* @Description: 按id去追踪规则
* @param @param get_id
* @param @return 设定文件
* @return TrackRule 返回类型
*/
void updateTrackrule(Long trackRule_id);
TrackRule getTrackRuleById(String id);
}
src/main/java/com/zhiwei/messageflow/mongo/dao/impl/TrackRuleDaoImpl.java
......@@ -22,18 +22,25 @@ public class TrackRuleDaoImpl implements TrackRuleDao {
@Override
public List<TrackRule> getTrackRuleByProject(String projectName) {
Query query = new Query();
query.addCriteria(Criteria.where("project").is(projectName));
primaryMongoTemplate.getCollectionName(TrackRule.class);
List<TrackRule> trackRules = primaryMongoTemplate.find(query, TrackRule.class);
return trackRules;
Query query = new Query(Criteria.where("project").is(projectName));
return primaryMongoTemplate.find(query, TrackRule.class);
}
@Override
public void updateTrackrule(Long trackRule_id) {
// System.out.println("预警状态更新");
primaryMongoTemplate.updateFirst(Query.query(Criteria.where("_id").is(trackRule_id)),
new Update().set("isWarn", true), TrackRule.class);
public boolean updateTrackrule(Query query, Update update) {
TrackRule t = primaryMongoTemplate.findOne(query, TrackRule.class);
if (t != null) {
primaryMongoTemplate.updateFirst(query, update, TrackRule.class);
} else {
return false;
}
return true;
}
@Override
public TrackRule getTrackRuleById(String id) {
Query query = new Query(Criteria.where("_id").is(id));
return primaryMongoTemplate.findOne(query, TrackRule.class);
}
}
src/main/java/com/zhiwei/messageflow/redis/RedisPoolAndTools.java
......@@ -398,4 +398,31 @@ public class RedisPoolAndTools {
return res;
}
public Set<String> zrevrange(String redisKey, int start, int end) {
Jedis jedis = getJedis();
while (true) {
if (null != jedis) {
break;
} else {
jedis = getJedis();
}
}
Set<String> res = jedis.zrevrange(redisKey, start, end);
returnResource(jedis);
return res;
}
public void zremrangebyscore(String key, long start, long end) {
Jedis jedis = getJedis();
while (true) {
if (null != jedis) {
break;
} else {
jedis = getJedis();
}
}
jedis.zremrangeByScore(key, start, end);
returnResource(jedis);
}
}
\ No newline at end of file
src/main/java/com/zhiwei/messageflow/redis/service/RedisService.java
......@@ -189,5 +189,14 @@ public interface RedisService {
* @return boolean 返回类型
*/
boolean dropCollectionHash(String id);
/**
* 按追踪id添加已追踪消息
* @Title: addWarnMsg2Set
* @Description: 按追踪id添加已追踪消息
* @param @param id
* @param @param msg 设定文件
* @return void 返回类型
*/
boolean addWarnMsg2Set(String id, JSONObject msg, int maxSize);
}
src/main/java/com/zhiwei/messageflow/redis/service/impl/RedisServiceImpl.java
package com.zhiwei.messageflow.redis.service.impl;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
......@@ -11,6 +12,7 @@ import com.alibaba.fastjson.JSONObject;
import com.fasterxml.jackson.annotation.JsonInclude.Include;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.zhiwei.messageflow.ESGetCommonId;
import com.zhiwei.messageflow.bean.MediaMessage;
import com.zhiwei.messageflow.bean.VideoMessage;
import com.zhiwei.messageflow.bean.WeiboMessage;
......@@ -151,27 +153,45 @@ public class RedisServiceImpl implements RedisService {
/**
* 删除超出存储上限的数据
*/
removeOverLimit(redisKey, maxSize);
}
private void removeOverLimit(String redisKey, int maxSize) {
long nowCount = redisPoolAndTools.getNowCount(redisKey);// 当前数据量
int removeIndex = (int) (nowCount - maxSize);// 需移除数据数量
if (maxSize > 0 && nowCount > maxSize) {
redisPoolAndTools.removeDataByName(redisKey, removeIndex);
// 判断zrevrange取maxSize条的commonid是否大于limitCommonid（小于或等于为以maxSize作为limit，大于以limitCommonid作为limit）
try {
Set<String> maxSizeSet = redisPoolAndTools.zrevrange(redisKey, maxSize - 1, maxSize - 1);
JSONObject object = JSONObject.parseObject((new ArrayList<>(maxSizeSet)).get(0));
long commonid = object.getLongValue("commonid");
if (commonid > ESGetCommonId.START_COMMONID) {
redisPoolAndTools.zremrangebyscore(redisKey, 0L,ESGetCommonId.START_COMMONID);
} else {
redisPoolAndTools.removeDataByName(redisKey, removeIndex);
}
} catch (Exception e) {
e.printStackTrace();
redisPoolAndTools.removeDataByName(redisKey, removeIndex);
}
}
}
@Override
public String getDirectRedisKey(String projectName, String groupName, String pt) {
return RedisConfig.DIRECTKEY+projectName+"-"+groupName+"-"+pt;
return RedisConfig.DIRECTKEY + projectName + "-" + groupName + "-" + pt;
}
@Override
public String getDirectRsidMapKey(String projectName) {
return RedisConfig.DIRECTKEY+projectName;
return RedisConfig.DIRECTKEY + projectName;
}
@Override
public boolean insertEvent(JSONObject ob) {
String redisKey = RedisConfig.EVENTKEY+ob.getString("eventId").replace(":", "：");
String redisKey = RedisConfig.EVENTKEY + ob.getString("eventId").replace(":", "：");
try {
redisPoolAndTools.sortedSetZadd(redisKey, (double) ob.getLong("offset"), mapper.writeValueAsString(ob));
} catch (JsonProcessingException e) {
......@@ -191,18 +211,18 @@ public class RedisServiceImpl implements RedisService {
@Override
public List<String> getNeedAutoMark() {
String redisKey = RedisConfig.EVENTLISTKEY;
return redisPoolAndTools.Lrange(redisKey,0,9);
return redisPoolAndTools.Lrange(redisKey, 0, 9);
}
@Override
public int countCollectionData(String id) {
String redisKey = RedisConfig.EVENTKEY+id.replace(":", "：");
String redisKey = RedisConfig.EVENTKEY + id.replace(":", "：");
return Integer.parseInt(String.valueOf(redisPoolAndTools.getNowCount(redisKey)));
}
@Override
public Set<String> getCollectionData(String id, int start, int end) {
String redisKey = RedisConfig.EVENTKEY+id.replace(":", "：");
String redisKey = RedisConfig.EVENTKEY + id.replace(":", "：");
return redisPoolAndTools.zrange(redisKey, start, end);
}
......@@ -215,26 +235,41 @@ public class RedisServiceImpl implements RedisService {
@Override
public boolean existsCollectionHashByUrlkey(String url, String id) {
String redisKey = RedisConfig.EVENTHASHKEY+id;
if(redisPoolAndTools.exists(redisKey)) {
//判断是否有hash urlkey
if(redisPoolAndTools.hexists(redisKey,url)) {
String redisKey = RedisConfig.EVENTHASHKEY + id;
if (redisPoolAndTools.exists(redisKey)) {
// 判断是否有hash urlkey
if (redisPoolAndTools.hexists(redisKey, url)) {
return true;
}else {
redisPoolAndTools.hset(redisKey,url);
} else {
redisPoolAndTools.hset(redisKey, url);
return false;
}
}else {
//新建hash
redisPoolAndTools.hset(redisKey,url);
} else {
// 新建hash
redisPoolAndTools.hset(redisKey, url);
return false;
}
}
@Override
public boolean dropCollectionHash(String id) {
String redisKey = RedisConfig.EVENTHASHKEY+id;
String redisKey = RedisConfig.EVENTHASHKEY + id;
return redisPoolAndTools.del(redisKey);
}
@Override
public boolean addWarnMsg2Set(String id, JSONObject jo, int maxSize) {
String redisKey = RedisConfig.TRACKKEY + id;
redisPoolAndTools.sortedSetZadd(redisKey, (double) jo.getLongValue("commonid"), jo.toJSONString());
/**
* 删除超出存储上限的数据
*/
long nowCount = redisPoolAndTools.getNowCount(redisKey);// 当前数据量
int removeIndex = (int) (nowCount - maxSize);// 需移除数据数量
if (maxSize > 0 && nowCount > maxSize) {
redisPoolAndTools.removeDataByName(redisKey, removeIndex);
}
return true;
}
}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment