Skip to content

M
messageflow
Commit 22a3d9e4 by shentao
Options

2018/10/30 消息流事件采集系统1.0版

parent 0044078e
Showing with 171 additions and 54 deletions
src/main/java/com/zhiwei/messageflow/ES4RedisRunner.java
......@@ -45,7 +45,7 @@ public class ES4RedisRunner implements ApplicationRunner {
// 手动注入bean ES4RedisStart
ES4RedisStart start = ApplicationContextProvider.getBean("ES4RedisStart", ES4RedisStart.class);
DES4RedisStart directstart = ApplicationContextProvider.getBean("DES4RedisStart", DES4RedisStart.class);
EVENT4RedisStart eventStart = ApplicationContextProvider.getBean("EVENT4RedisStart", EVENT4RedisStart.class);
EVENTAutoMarkStart eventStart = ApplicationContextProvider.getBean("EVENTAutoMarkStart", EVENTAutoMarkStart.class);
// 定时器
Timer timer = new Timer();
......
src/main/java/com/zhiwei/messageflow/ES4RedisStart.java
......@@ -44,7 +44,7 @@ public class ES4RedisStart {
// 遍历项目
for (Project project : projects) {
// if(!project.getProjectName().equals("证监会")) {
// if(!project.getProjectName().equals("今日头条")) {
// continue;
// }
......
src/main/java/com/zhiwei/messageflow/EVENT4RedisStart.java → src/main/java/com/zhiwei/messageflow/EVENTAutoMarkStart.java
......@@ -7,16 +7,17 @@ import java.util.List;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import com.fasterxml.jackson.core.JsonParseException;
import com.fasterxml.jackson.databind.JsonMappingException;
import com.zhiwei.messageflow.bean.Event;
import com.zhiwei.messageflow.mongo.bean.Project;
import com.zhiwei.messageflow.redis.service.RedisService;
import com.zhiwei.messageflow.service.EventService;
public class EVENT4RedisStart {
private final static Logger log = LogManager.getLogger(EVENT4RedisStart.class);
@Component
public class EVENTAutoMarkStart {
private final static Logger log = LogManager.getLogger(EVENTAutoMarkStart.class);
@Autowired
protected RedisService redisService;
......@@ -24,6 +25,7 @@ public class EVENT4RedisStart {
@Autowired
private EventService eventService;
@Autowired
private EVENTAutoMarkTask eventAutoMark;
public void startThread() throws JsonParseException, JsonMappingException, IOException, InterruptedException {
......
src/main/java/com/zhiwei/messageflow/EVENTAutoMarkTask.java
package com.zhiwei.messageflow;
import java.io.IOException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import com.fasterxml.jackson.core.JsonParseException;
import com.fasterxml.jackson.databind.JsonMappingException;
import com.zhiwei.messageflow.bean.Event;
import com.zhiwei.messageflow.service.EventService;
@Component
public class EVENTAutoMarkTask {
@Autowired
private EventService eventService;
public boolean eventCollectionAutoMark(Event eventCollection) throws JsonParseException, JsonMappingException, IOException {
public boolean eventCollectionAutoMark(Event eventCollection) throws Exception {
return eventService.autoMark(eventCollection);
}
......
src/main/java/com/zhiwei/messageflow/EVENTAutoMarkThread.java
......@@ -22,7 +22,7 @@ public class EVENTAutoMarkThread extends Thread{
private EVENTAutoMarkTask eventAutoMark;
public EVENTAutoMarkThread(Event eventCollection, EVENTAutoMarkTask eventAutoMark2) {
public EVENTAutoMarkThread(Event eventCollection, EVENTAutoMarkTask eventAutoMark) {
this.threadName = eventCollection.getId();
this.eventCollection = eventCollection;
this.eventAutoMark = eventAutoMark;
......@@ -53,27 +53,21 @@ public class EVENTAutoMarkThread extends Thread{
Thread.sleep(10L);
// 程序运行
log.info("Running 事件采集进入自动标注 {}", threadName);
log.info("Running 事件采集自动标注中 {}", threadName);
// 该项目执行事件采集存储
boolean flag = eventAutoMark.eventCollectionAutoMark(eventCollection);
if (!flag) {
Thread.currentThread().interrupted();
log.error("{}事件采集进入自动标注出现异常，线程状态:{}",threadName,Thread.currentThread().isInterrupted());
log.error("{}事件采集自动标注出现异常，线程状态:{}",threadName,Thread.currentThread().isInterrupted());
}
}catch (JsonParseException e) {
e.printStackTrace();
} catch (JsonMappingException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}catch (Exception e) {
log.error("Thread {}事件采集有误:{} \n 位置{}.", threadName,e,e.getStackTrace());
} finally {
// 线程退出
log.info("Thread 事件采集进入自动标注 {} exiting.", threadName);
log.info("Thread 事件采集自动标注结束 {} exiting.", threadName);
}
}
......
src/main/java/com/zhiwei/messageflow/MessageflowApplication.java
......@@ -4,8 +4,10 @@ import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.data.mongo.MongoDataAutoConfiguration;
import org.springframework.boot.autoconfigure.mongo.MongoAutoConfiguration;
import org.springframework.kafka.annotation.EnableKafka;
@SpringBootApplication(exclude = {MongoAutoConfiguration.class,MongoDataAutoConfiguration.class})
@EnableKafka
public class MessageflowApplication {
public static void main(String[] args) {
......
src/main/java/com/zhiwei/messageflow/config/RedisConfig.java
......@@ -27,7 +27,10 @@ public class RedisConfig {
private String password;
public static final String DIRECTKEY = "Direct:";
public static final String EVENTKEY = "Event:";
/** 事件采集待处理列表 **/
public static final String EVENTLISTKEY = "Event:EventList";
/** 事件采集去重url集 **/
public static final String EVENTHASHKEY = "Event:Hash:";
private int keyMaxSize;
......
src/main/java/com/zhiwei/messageflow/kafka/KafkaConsumer.java
......@@ -30,12 +30,12 @@ public class KafkaConsumer {
// L.i("message =" + message);
// }
// }
@KafkaListener(topics = { "qbjc_event_topic" }, containerFactory = "kafkaListenerContainerFactory")
@KafkaListener(topics = {"qbjc_event_topic"}, containerFactory = "kafkaListenerContainerFactory")
public void consumerMsg(List<ConsumerRecord> records, Acknowledgment ack) {
try {
// eventService.saveRecords(records);
for (ConsumerRecord record : records) {
log.info("record = {}" , record);
// log.info("record = {}" , record);
Optional<?> kafkaMessage = Optional.ofNullable(record.value());
if (kafkaMessage.isPresent()) {
Object message = kafkaMessage.get();
......
src/main/java/com/zhiwei/messageflow/redis/RedisPoolAndTools.java
......@@ -320,5 +320,82 @@ public class RedisPoolAndTools {
jedis.lrem(key, 0, id);
returnResource(jedis);
}
/**
* 判断key是否存在
* @Title: exists
* @Description: 判断key是否存在
* @param @param key
* @param @return 设定文件
* @return boolean 返回类型
*/
public boolean exists(String key) {
Jedis jedis = getJedis();
while (true) {
if (null != jedis) {
break;
} else {
jedis = getJedis();
}
}
boolean res=jedis.exists(key);
returnResource(jedis);
return res;
}
/**
* 判断hash中key是否存在
* @Title: hexists
* @Description: 判断hash中key是否存在
* @param @param redisKey
* @param @param url
* @param @return 设定文件
* @return boolean 返回类型
*/
public boolean hexists(String key, String url) {
Jedis jedis = getJedis();
while (true) {
if (null != jedis) {
break;
} else {
jedis = getJedis();
}
}
boolean res=jedis.hexists(key, url);
returnResource(jedis);
return res;
}
/**
* 如果哈希表不存在，一个新的哈希表被创建并进行 HSET 操作。如果字段已经存在于哈希表中，旧值将被覆盖。
* @Title: hset
* @Description: 如果哈希表不存在，一个新的哈希表被创建并进行 HSET 操作。如果字段已经存在于哈希表中，旧值将被覆盖。
* @param @param redisKey
* @param @param url 设定文件
* @return void 返回类型
*/
public void hset(String key, String url) {
Jedis jedis = getJedis();
while (true) {
if (null != jedis) {
break;
} else {
jedis = getJedis();
}
}
jedis.hset(key, url, url);
returnResource(jedis);
}
public boolean del(String key) {
Jedis jedis = getJedis();
while (true) {
if (null != jedis) {
break;
} else {
jedis = getJedis();
}
}
boolean res= 1==jedis.del(key)?true:false;
returnResource(jedis);
return res;
}
}
\ No newline at end of file
src/main/java/com/zhiwei/messageflow/redis/service/RedisService.java
......@@ -170,5 +170,24 @@ public interface RedisService {
* @return boolean 返回类型
*/
boolean removeEventAutoMarkList(String id);
/**
* 判断url是否存在
* @param id
* @Title: existsCollectionHashByUrlkey
* @Description: 判断url是否存在
* @param @param url
* @param @return 设定文件
* @return boolean 返回类型
*/
boolean existsCollectionHashByUrlkey(String url, String id);
/**
* 按事件采集id删除事件采集urlHash
* @Title: dropCollectionHash
* @Description: 按事件采集id删除事件采集urlHash
* @param @param id
* @param @return 设定文件
* @return boolean 返回类型
*/
boolean dropCollectionHash(String id);
}
src/main/java/com/zhiwei/messageflow/redis/service/impl/RedisServiceImpl.java
......@@ -213,4 +213,28 @@ public class RedisServiceImpl implements RedisService {
return true;
}
@Override
public boolean existsCollectionHashByUrlkey(String url, String id) {
String redisKey = RedisConfig.EVENTHASHKEY+id;
if(redisPoolAndTools.exists(redisKey)) {
//判断是否有hash urlkey
if(redisPoolAndTools.hexists(redisKey,url)) {
return true;
}else {
redisPoolAndTools.hset(redisKey,url);
return false;
}
}else {
//新建hash
redisPoolAndTools.hset(redisKey,url);
return false;
}
}
@Override
public boolean dropCollectionHash(String id) {
String redisKey = RedisConfig.EVENTHASHKEY+id;
return redisPoolAndTools.del(redisKey);
}
}
src/main/java/com/zhiwei/messageflow/service/EventService.java
......@@ -62,8 +62,9 @@ public interface EventService {
* @throws IOException
* @throws JsonMappingException
* @throws JsonParseException
* @throws Exception
*/
boolean autoMark(Event eventCollection) throws JsonParseException, JsonMappingException, IOException;
boolean autoMark(Event eventCollection);
/**
* 更新事件等待采集列表中的前十个任务状态为采集完毕
* @Title: updateTopTenCollection
......
src/main/resources/application.properties
......@@ -44,9 +44,6 @@ spring.data.mongodb.thirdary.password=stno1q2w3e4r
spring.data.mongodb.thirdary.authenticationDatabase=admin
#kafka
spring.kafka.bootstrap-servers = kafka1.irybd.com:9092
spring.kafka.producer.key-serializer = org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer = org.apache.kafka.common.serialization.StringSerializer
#=============== consumer =======================
# group id
#spring.kafka.consumer.group-id=group1
......@@ -59,11 +56,11 @@ spring.kafka.producer.value-serializer = org.apache.kafka.common.serialization.S
#spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
#spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
kafka.consumer.servers=kafka1.irybd.com:9092
kafka.consumer.servers=192.168.0.203:9092,192.168.0.203:9093,192.168.0.203:9094
kafka.consumer.enable.auto.commit=false
kafka.consumer.session.timeout=15000
kafka.consumer.auto.commit.interval=100
kafka.consumer.auto.offset.reset=earliest
kafka.consumer.group.id=group
kafka.consumer.concurrency=10
kafka.consumer.group.id=stno
kafka.consumer.concurrency=1
kafka.consumer.maxPollRecordsConfig=100
\ No newline at end of file
src/main/resources/application.properties.local1
##服务端uri
##\u670D\u52A1\u7AEFuri
#spring.data.mongodb.uri=115.236.59.91:27017
#服务端数据库
#\u670D\u52A1\u7AEF\u6570\u636E\u5E93
spring.data.mongodb.primary.database=qbjcPhoenix
#服务ip
#\u670D\u52A1ip
spring.data.mongodb.primary.host=127.0.0.1
#服务port
#\u670D\u52A1port
spring.data.mongodb.primary.port=27017
#spring.data.mongodb.primary.username=stno
......@@ -13,11 +13,11 @@ spring.data.mongodb.primary.port=27017
#spring.data.mongodb.primary.authenticationDatabase=admin
#服务端数据库
#\u670D\u52A1\u7AEF\u6570\u636E\u5E93
spring.data.mongodb.secondary.database=eventMuseum
#服务ip
#\u670D\u52A1ip
spring.data.mongodb.secondary.host=202.107.192.94
#服务port
#\u670D\u52A1port
spring.data.mongodb.secondary.port=30000
spring.data.mongodb.secondary.username=stno
......@@ -26,11 +26,11 @@ spring.data.mongodb.secondary.password=stno1q2w3e4r
spring.data.mongodb.secondary.authenticationDatabase=admin
#服务端数据库
#\u670D\u52A1\u7AEF\u6570\u636E\u5E93
spring.data.mongodb.thirdary.database=WechatPublic
#服务ip
#\u670D\u52A1ip
spring.data.mongodb.thirdary.host=202.107.192.94
#服务port
#\u670D\u52A1port
spring.data.mongodb.thirdary.port=30000
spring.data.mongodb.thirdary.username=stno
......@@ -40,9 +40,6 @@ spring.data.mongodb.thirdary.password=stno1q2w3e4r
spring.data.mongodb.thirdary.authenticationDatabase=admin
#kafka
spring.kafka.bootstrap-servers = kafka1.irybd.com:9092
spring.kafka.producer.key-serializer = org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer = org.apache.kafka.common.serialization.StringSerializer
#=============== consumer =======================
# group id
#spring.kafka.consumer.group-id=group1
......@@ -55,11 +52,12 @@ spring.kafka.producer.value-serializer = org.apache.kafka.common.serialization.S
#spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
#spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
#kafka1.irybd.com:9092,kafka1.irybd.com:9093,kafka1.irybd.com:9094
kafka.consumer.servers=kafka1.irybd.com:9092
kafka.consumer.enable.auto.commit=false
kafka.consumer.session.timeout=15000
kafka.consumer.auto.commit.interval=100
kafka.consumer.auto.offset.reset=earliest
kafka.consumer.group.id=group
kafka.consumer.concurrency=10
kafka.consumer.group.id=group-test
kafka.consumer.concurrency=1
kafka.consumer.maxPollRecordsConfig=100
\ No newline at end of file
src/main/resources/application.properties.local2
......@@ -59,7 +59,7 @@ kafka.consumer.servers=kafka1.irybd.com:9092
kafka.consumer.enable.auto.commit=false
kafka.consumer.session.timeout=15000
kafka.consumer.auto.commit.interval=100
kafka.consumer.auto.offset.reset=earliest
kafka.consumer.auto.offset.reset=latest
kafka.consumer.group.id=group
kafka.consumer.concurrency=10
kafka.consumer.concurrency=1
kafka.consumer.maxPollRecordsConfig=100
\ No newline at end of file
src/main/resources/application.properties.test10001
......@@ -44,9 +44,6 @@ spring.data.mongodb.thirdary.password=stno1q2w3e4r
spring.data.mongodb.thirdary.authenticationDatabase=admin
#kafka
spring.kafka.bootstrap-servers = kafka1.irybd.com:9092
spring.kafka.producer.key-serializer = org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer = org.apache.kafka.common.serialization.StringSerializer
#=============== consumer =======================
# group id
#spring.kafka.consumer.group-id=group1
......@@ -59,11 +56,11 @@ spring.kafka.producer.value-serializer = org.apache.kafka.common.serialization.S
#spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
#spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
kafka.consumer.servers=kafka1.irybd.com:9092
kafka.consumer.servers=192.168.0.203:9092,192.168.0.203:9093,192.168.0.203:9094
kafka.consumer.enable.auto.commit=false
kafka.consumer.session.timeout=15000
kafka.consumer.auto.commit.interval=100
kafka.consumer.auto.offset.reset=earliest
kafka.consumer.group.id=group
kafka.consumer.concurrency=10
kafka.consumer.group.id=stno
kafka.consumer.concurrency=1
kafka.consumer.maxPollRecordsConfig=100
\ No newline at end of file
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