Skip to content

V
VideoAIBatchTaskConsumerService
Commit 50cef139 authored by wangjinjing's avatar wangjinjing
Browse files
Options

发布版本

parent b0e0ee4f
Hide whitespace changes
Inline Side-by-side
Showing with 549 additions and 317 deletions
pom.xml
View file @ 50cef139
......@@ -107,26 +107,12 @@
<artifactId>dom4j</artifactId>
<version>1.6</version>
</dependency>
<!--<dependency>-->
<!--<groupId>org.jdom</groupId>-->
<!--<artifactId>jdom</artifactId>-->
<!--<version>1.1</version>-->
<!--</dependency>-->
<!--<dependency>-->
<!--<groupId>commons-io</groupId>-->
<!--<artifactId>commons-io</artifactId>-->
<!--</dependency>-->
<dependency>
<groupId>io.springfox</groupId>
<artifactId>springfox-swagger2</artifactId>
<version>2.7.0</version>
</dependency>
<dependency>
<groupId>io.springfox</groupId>
<artifactId>springfox-swagger-ui</artifactId>
<version>2.7.0</version>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-api</artifactId>
<version>2.17.0</version>
</dependency>
</dependencies>
......
src/main/java/com/cx/cn/cxquartz/CXQuartzApplication.java
View file @ 50cef139
......@@ -4,9 +4,7 @@ import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.boot.web.servlet.support.SpringBootServletInitializer;
import springfox.documentation.swagger2.annotations.EnableSwagger2;
@EnableSwagger2
@SpringBootApplication
public class CXQuartzApplication extends SpringBootServletInitializer {
......
src/main/java/com/cx/cn/cxquartz/bean/TaskResult.java
View file @ 50cef139
......@@ -11,6 +11,7 @@ public class TaskResult {
private String url;
private String threshold;
private String sendtype;
private String callbackurl;
public String getSendtype() {
return sendtype;
......@@ -20,7 +21,15 @@ public class TaskResult {
this.sendtype = sendtype;
}
public TaskResult(String devicecode, String recordtype, String timestamp, Long x, Long y, Long w, Long h, String url, String threshold, String sendtype) {
public String getCallbackurl() {
return callbackurl;
}
public void setCallbackurl(String callbackurl) {
this.callbackurl = callbackurl;
}
public TaskResult(String devicecode, String recordtype, String timestamp, Long x, Long y, Long w, Long h, String url, String threshold, String sendtype, String callbackurl) {
this.devicecode = devicecode;
this.recordtype = recordtype;
this.timestamp = timestamp;
......@@ -31,6 +40,7 @@ public class TaskResult {
this.url = url;
this.threshold = threshold;
this.sendtype = sendtype;
this.callbackurl=callbackurl;
}
public String getUrl() {
......
src/main/java/com/cx/cn/cxquartz/config/PerformedTaskCornChange.java
View file @ 50cef139
......@@ -44,12 +44,10 @@ public class PerformedTaskCornChange implements SchedulingConfigurer {
@Override
public void run() {
//判断最小执行时间, 如果小于2分钟，立即执行所有设备的获取rtsp与hls 的服务，下次执行时间为间隔2分钟
// 如果大于2分钟，下次执行时间为间隔1秒
Long time= sbtdspsrService.getPeriodicseconds();
if(time<1200){
setTimer(120000L);
List<Sbtdspsr> list= sbtdspsrService.getPerformedTasks();
if(null!=list && list.size()>0) {
// 如果大于2分钟，下次执行时间为间隔time-2分钟秒
Long time= sbtdspsrService.getPeriodicseconds();
if(null!=time && time<=120){
List<Sbtdspsr> list= sbtdspsrService.getPerformedTasks();
for(Sbtdspsr sbtdspsr:list) {
//丢到rabbitMq中
String msgId = UUID.randomUUID().toString();
......@@ -59,10 +57,10 @@ public class PerformedTaskCornChange implements SchedulingConfigurer {
MessageHelper.objToMsg(sbtdspsr),
correlationData);
}
setTimer(1000L);//间隔一秒再去判断
}
}
else {
setTimer(1000L);
setTimer(time*1000-120000);
}
}
}, new Trigger() {
......
src/main/java/com/cx/cn/cxquartz/config/RabbitMqAdminConf.java 0 → 100644
View file @ 50cef139
package com.cx.cn.cxquartz.config;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;
@Component
public class RabbitMqAdminConf {
@Bean
public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
// 只有设置为 true，spring 才会加载 RabbitAdmin 这个类
rabbitAdmin.setAutoStartup(true);
return rabbitAdmin;
}
}
src/main/java/com/cx/cn/cxquartz/config/RabbitMqService.java 0 → 100644
View file @ 50cef139
package com.cx.cn.cxquartz.config;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.rabbit.core.ChannelCallback;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
@Component
public class RabbitMqService {
@Resource
RabbitAdmin rabbitAdmin;
/**
* 获取对应队列的数量;
*
* @param queue
* @return
*/
public int getMessageCount(String queue) {
AMQP.Queue.DeclareOk declareOk = rabbitAdmin.getRabbitTemplate().execute(new ChannelCallback<AMQP.Queue.DeclareOk>() {
public AMQP.Queue.DeclareOk doInRabbit(Channel channel) throws Exception {
return channel.queueDeclarePassive(queue);
}
});
return declareOk.getMessageCount();
}
}
src/main/java/com/cx/cn/cxquartz/config/RedisPCQueueConfig.java
View file @ 50cef139
......@@ -11,7 +11,7 @@ import org.springframework.context.annotation.Configuration;
/**
* redis 消费者初始化
*/
@Configuration
//@Configuration
public class RedisPCQueueConfig {
@Autowired
......
src/main/java/com/cx/cn/cxquartz/config/ScheduleTaskConfig.java
View file @ 50cef139
package com.cx.cn.cxquartz.config;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
@Configuration
@EnableScheduling
public class ScheduleTaskConfig {
@Value("${file.getrtspbyurl}")
private String getrtspbyurl;
private static final Logger logger = LoggerFactory.getLogger(ScheduleTaskConfig.class);
@Autowired
private RabbitMqService rRabbitMqService;
@Scheduled(cron = "*/45 * * * * ?")//每隔45s查询所有queue的消费数目，超过警戒值的
private void statistoday() {
try {
int count= rRabbitMqService.getMessageCount("RabbitMQ.DirectQueue.SendToVoiceConsumer");
logger.info("count:{}", count);
}
catch (Exception ex){
logger.error(ex.toString());
}
}
}
src/main/java/com/cx/cn/cxquartz/config/Swagger2Config.java deleted 100644 → 0
View file @ b0e0ee4f
package com.cx.cn.cxquartz.config;
import com.google.common.base.Predicates;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import springfox.documentation.builders.ApiInfoBuilder;
import springfox.documentation.builders.PathSelectors;
import springfox.documentation.builders.RequestHandlerSelectors;
import springfox.documentation.service.ApiInfo;
import springfox.documentation.service.Contact;
import springfox.documentation.spi.DocumentationType;
import springfox.documentation.spring.web.plugins.Docket;
import springfox.documentation.swagger2.annotations.EnableSwagger2;
@Configuration
@EnableSwagger2
public class Swagger2Config {
@Bean
public Docket createRestApi() {
return new Docket(DocumentationType.SWAGGER_2).apiInfo(apiInfo()).select().apis(RequestHandlerSelectors.any())
.paths(Predicates.not(PathSelectors.regex("/error.*"))).paths(Predicates.not(PathSelectors.regex("/info.*")))
.paths(Predicates.not(PathSelectors.regex("/autoconfig.*")))
.paths(Predicates.not(PathSelectors.regex("/health.*")))
.paths(Predicates.not(PathSelectors.regex("/metrics.*")))
.paths(Predicates.not(PathSelectors.regex("/mappings.*")))
.paths(Predicates.not(PathSelectors.regex("/trace.*")))
.paths(Predicates.not(PathSelectors.regex("/configprops.*")))
.paths(Predicates.not(PathSelectors.regex("/beans.*"))).paths(Predicates.not(PathSelectors.regex("/env.*")))
.paths(Predicates.not(PathSelectors.regex("/dump.*")))
.paths(Predicates.not(PathSelectors.regex("/auditevents.*")))
.paths(Predicates.not(PathSelectors.regex("/docs.*"))).paths(Predicates.not(PathSelectors.regex("/archaius.*")))
.paths(Predicates.not(PathSelectors.regex("/features.*")))
.paths(Predicates.not(PathSelectors.regex("/pause.*"))).paths(Predicates.not(PathSelectors.regex("/refresh.*")))
.paths(Predicates.not(PathSelectors.regex("/resume.*")))
.paths(Predicates.not(PathSelectors.regex("/actuator.*")))
.paths(Predicates.not(PathSelectors.regex("/jolokia.*")))
.paths(Predicates.not(PathSelectors.regex("/loggers.*")))
.paths(Predicates.not(PathSelectors.regex("/restart.*")))
.paths(Predicates.not(PathSelectors.regex("/service-registry/instance-status.*")))
.paths(Predicates.not(PathSelectors.regex("/logfile.*")))
.paths(Predicates.not(PathSelectors.regex("/channels.*"))).paths(Predicates.not(PathSelectors.regex("/bus.*")))
.paths(Predicates.not(PathSelectors.regex("/heapdump.*"))).build();
}
/**
* Swagger2主界面信息,描述api的基本信息用于展示
*
* @return
*/
private ApiInfo apiInfo() {
return new ApiInfoBuilder().title("video服务接口定义和规范").description("knotvideo服务接口定义和规范")
.termsOfServiceUrl("http://****.com/").contact(new Contact("@****.com", "", "")).version("1.0")
.license("版权所有 ?***").build();
}
}
src/main/java/com/cx/cn/cxquartz/controller/AutoSnapController.java
View file @ 50cef139
......@@ -5,9 +5,6 @@ import com.cx.cn.cxquartz.helper.MessageHelper;
import com.cx.cn.cxquartz.rabbitmq.QueueConstants;
import com.cx.cn.cxquartz.util.*;
import com.cx.cn.cxquartz.vo.*;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.dom4j.Attribute;
import org.dom4j.Document;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
......@@ -19,30 +16,26 @@ import org.springframework.web.bind.annotation.*;
import org.springframework.web.multipart.MultipartFile;
import java.io.File;
import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.*;
@RestController
@Api(description ="自动抓拍摄像头结果推送服务接口",tags = "自动抓拍摄像头结果推送服务")
public class AutoSnapController {
private static final Logger logger = LoggerFactory.getLogger(ExtController.class);
@Autowired
RabbitTemplate rabbitTemplate;
@Value("${file.rootpath}")
String rootpath;
@Value("${snapnote.note1}")
String note1;
@RequestMapping(value = "/autoSnap", method = RequestMethod.POST)
@ResponseBody
public String AutoSnap(@RequestParam("imageinfo.xml_0") MultipartFile filexml,@RequestParam("vcaimage_0.jpg_1") MultipartFile filejpg ) {
@RequestMapping(value = "/AutoSnap", method = RequestMethod.POST)
@ApiOperation(value = "自动抓拍摄像头结果推送服务",notes = "上传xml和jpg文件")
public void AutoSnap(@RequestParam("file") MultipartFile[] file) {
/*如果文件不为空，保存文件*/
if (file != null && file.length>0 && file.length==2) {
//获得xml文件 和jpg文件
MultipartFile filexml=file[0].getName().endsWith(".xml")==true?file[0]:file[1];
MultipartFile filejpg=file[0].getName().endsWith(".jpg")||file[0].getName().endsWith(".jpeg") ==true?file[0]:file[1];
logger.info("autosnap xml:{}",filexml.getName());
logger.info("autosnap filejpg:{}",filejpg.getName());
// /*如果文件不为空，保存文件*/
if (filexml != null && filejpg!= null) {
//获得xml文件里信息，获得图片地址，将信息与地址整合成rabbitmq 中的格式进行消费
try {
Document doc= XmlUtils.readDocument(filexml);
......@@ -50,28 +43,25 @@ public class AutoSnapController {
XmlUtils.parseDocument(vCACapture,doc);
vCACapture.setDateTime(vCACapture.getDateTime()==null?null:DateUtils.dealDateFormat(vCACapture.getDateTime()));
//获得jpg 文件，将文件存放到某个路径下面
Calendar rightNow = Calendar.getInstance();
String filepath=rootpath+File.separator+rightNow.get(Calendar.YEAR)+File.separator+rightNow.get(Calendar.MONTH)+File.separator+vCACapture.getDeviceId()+File.separator+vCACapture.getDeviceId()+"_"+DateUtils.formatDateToNoSign(vCACapture.getDateTime())+".jpg";
//将文件copy到该路径下
File desfile=new File(filepath);
if (!desfile.getParentFile().exists()) {
desfile.getParentFile().mkdirs();
}
if (!desfile.exists()) {
desfile.createNewFile();
}
filejpg.transferTo(desfile);
//
Map map=new HashMap<>();
map.put("resourcePath",filepath);
map.put("timestamp", vCACapture.getDateTime() );
map.put("devicecode", vCACapture.getDeviceId());
CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
rabbitTemplate.convertAndSend(QueueConstants.QueueAutoSnapEnum.QUEUE_AUTOSNAP_ENUM.getExchange(),
QueueConstants.QueueAutoSnapEnum.QUEUE_AUTOSNAP_ENUM.getRouteKey(),
MessageHelper.objToMsg(map),
correlationData);
//拼接数据为如下格式
String filepath = DateUtils.formatCurrDatefileYMD() + File.separator + vCACapture.getDeviceId() + File.separator + vCACapture.getDeviceId() + "_" + DateUtils.formatDateToNoSign(vCACapture.getDateTime()) + ".jpg";
logger.info(filepath);
//将文件copy到该路径下
File desfile = new File(note1 +File.separator + filepath);
if (!desfile.getParentFile().exists()) {
desfile.getParentFile().mkdirs();
}
filejpg.transferTo(desfile);
Map map=new HashMap<>();
map.put("resourcePath",filepath);//不加节点的resourcePath，方便哪个节点的消费者抢到消息直接添加节点信息
map.put("timestamp", vCACapture.getDateTime());
map.put("devicecode", vCACapture.getDeviceId());
CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
rabbitTemplate.convertAndSend(QueueConstants.QueueAutoSnapEnum.QUEUE_AUTOSNAP_ENUM.getExchange(),
QueueConstants.QueueAutoSnapEnum.QUEUE_AUTOSNAP_ENUM.getRouteKey(),
MessageHelper.objToMsg(map),
correlationData);
return "success";
//拼接数据为如下格式
// String result="{\n" +
// " \"ret\": 0,\n" +
// " \"desc\": \"succ!\",\n" +
......@@ -83,11 +73,11 @@ public class AutoSnapController {
} catch (IOException e) {
e.printStackTrace();
logger.error("IOException",e);
return "IO error";
}
}else{
logger.error("文件数量不是2个");
logger.error("参数为null");
return "null";
}
}
}
src/main/java/com/cx/cn/cxquartz/controller/ExtController.java
View file @ 50cef139
......@@ -5,11 +5,12 @@ import com.cx.cn.cxquartz.job.WebSocket;
import com.cx.cn.cxquartz.service.quartz.*;
import com.cx.cn.cxquartz.service.quartz.impl.EventWriteService;
import com.cx.cn.cxquartz.util.DateUtils;
import com.cx.cn.cxquartz.util.JsonUtil;
import com.cx.cn.cxquartz.util.RestUtil;
import com.cx.cn.cxquartz.util.ResultUtil;
import com.cx.cn.cxquartz.vo.JobTjParam;
import com.cx.cn.cxquartz.vo.ResultObj;
import com.cx.cn.cxquartz.vo.TraffAlarmRecord;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
......@@ -25,13 +26,10 @@ import java.util.List;
@RestController
@RequestMapping("/ext")
@PropertySource("classpath:file.properties")
@Api(description ="QST物体识别算法接口",tags = "QST物体识别算法接口")
public class ExtController {
private static final Logger logger = LoggerFactory.getLogger(AutoSnapController.class);
@Value("${file.rtspurl}")
private String rtspurl;
@Value("${file.recogurl}")
private String recogurl;
@Value("${file.getrtspbyurl}")
private String getrtspbyurl;
@Value("${file.publicpictureurl}")
......@@ -45,28 +43,14 @@ public class ExtController {
@Value("${voice.eventId}")
private Integer eventId;
@Value("${file.rootpath}")
private String filerootpath;
@Value("${file.outpath}")
private String outpath;
@Value("${voice.unionId}")
private String unionId;
@Value("${countryside.callbackurl}")
private String callbackurl;
@Value("${web.url}")
private String weburl;
@Autowired
private RestTemplate restTemplate;
@Autowired
private SbtdspsrService sbtdspsrService;
@Autowired
TraffAlarmRecordService traffAlarmRecordService;
@Autowired
......@@ -100,8 +84,7 @@ public class ExtController {
}
@RequestMapping(value = "/getRTSP/{photonum}", method = RequestMethod.POST)
@ApiOperation(value = "简单抽帧服务接口",notes = "")
public String getrtsp(@RequestBody String videoid,
public String getrtsp(@RequestBody String videoid,
@PathVariable("photonum") Integer photonum) {
//尝试抽取第一张图片
List<String> imgUrls = new ArrayList<>();
......@@ -482,6 +465,12 @@ public class ExtController {
//
// }
// }
@RequestMapping(value = "/callback", method = RequestMethod.POST)
@ResponseBody
public ResultObj sendcallback(JobTjParam jobTjParam){
logger.info(JsonUtil.objToStr(jobTjParam));
return ResultObj.ok();
}
}
src/main/java/com/cx/cn/cxquartz/rabbitmq/QueueConstants.java
View file @ 50cef139
package com.cx.cn.cxquartz.rabbitmq;
/**
* 消息队列常量
*/
......@@ -410,4 +411,74 @@ public class QueueConstants {
*/
String ROUTEKEY = "RabbitMQ.RouteKey.AutoSnapConsumer";
}
/**
* 派发任务队列常量
*/
public interface QueueSnapTaskConsumer{
/**
* 交换机名称
*/
String EXCHANGE = "RabbitMQ.DirectExchange.SnapTaskConsumer";
/**
* 队列名称
*/
String QUEUE = "RabbitMQ.DirectQueue.SnapTaskConsumer";
/**
* 路由键
*/
String ROUTEKEY = "RabbitMQ.RouteKey.SnapTaskConsumer";
}
/**
* 消息通知队列
*/
public enum QueueSnapTaskEnum {
QUEUE_SNAP_TASK_ENUM(QueueConstants.QueueSnapTaskConsumer.EXCHANGE,
QueueConstants.QueueSnapTaskConsumer.QUEUE,
QueueConstants.QueueSnapTaskConsumer.ROUTEKEY);
/**
* 交换机名称
*/
private String exchange;
/**
* 队列名称
*/
private String queue;
/**
* 路由键
*/
private String routeKey;
QueueSnapTaskEnum(String exchange, String queue, String routeKey) {
this.exchange = exchange;
this.queue = queue;
this.routeKey = routeKey;
}
public String getExchange() {
return exchange;
}
public void setExchange(String exchange) {
this.exchange = exchange;
}
public String getQueue() {
return queue;
}
public void setQueue(String queue) {
this.queue = queue;
}
public String getRouteKey() {
return routeKey;
}
public void setRouteKey(String routeKey) {
this.routeKey = routeKey;
}
}
}
src/main/java/com/cx/cn/cxquartz/rabbitmq/comsumer/SendToDXConsumer.java
View file @ 50cef139
......@@ -40,8 +40,8 @@ public class SendToDXConsumer implements BaseConsumer {
if (null != result.get("id") && null!=result.get("traff") && null!=result.get("callback")) {
JobTjParam jobTjParam=JsonUtil.strToObj(result.get("traff").toString(),JobTjParam.class);
if(null!=jobTjParam) {
// eventWriteService.sendEventByCallUrl(Long.parseLong(result.get("id").toString())
// , jobTjParam, result.get("callback").toString());
eventWriteService.sendEventByCallUrl(Long.parseLong(result.get("id").toString())
, jobTjParam, result.get("callback").toString());
}
}
}
......
src/main/java/com/cx/cn/cxquartz/rabbitmq/comsumer/SendToVoiceConsumer.java
View file @ 50cef139
......@@ -65,7 +65,7 @@ public class SendToVoiceConsumer implements BaseConsumer {
voicedata.setRequestData(new Voice(eventId,unionId));
// logger.info(" send to voice: {}", new ObjectMapper().writeValueAsString(voicedata));
//同步推送
// eventWriteService.sendVoice(voicedata);
eventWriteService.sendVoice(voicedata,result.get("callback").toString());
}
//推送告警到前端
// webSocket.GroupSending(JsonUtil.objToStr(traffpictureParamresult));
......
src/main/java/com/cx/cn/cxquartz/rabbitmq/comsumer/SnapShotConsumer.java
View file @ 50cef139
......@@ -21,7 +21,6 @@ public class SnapShotConsumer implements BaseConsumer{
logger.info("SnapShotConsumer 收到消息: {}", message.toString());
Sbtdspsr result = MessageHelper.msgToObj(message, Sbtdspsr.class);
if (result.getTdlx()==1) {
//调用rtsp 的服务
if(null!=result.getSbbh()&&(!result.getSbbh().startsWith("33") ||
(result.getSbbh().startsWith("33") &&result.getSbbh().length()==18))) {
String token = videoRTSPorURLService.getRTSPAccessToekenByDeviceCode(result.getSbbh());
......
src/main/java/com/cx/cn/cxquartz/rabbitmq/comsumer/SnapTaskConsumer.java 0 → 100644
View file @ 50cef139
package com.cx.cn.cxquartz.rabbitmq.comsumer;
import com.cx.cn.cxquartz.bean.QuartzTaskInformations;
import com.cx.cn.cxquartz.helper.MessageHelper;
import com.cx.cn.cxquartz.rabbitmq.QueueConstants;
import com.cx.cn.cxquartz.redis.OrderConsumer;
import com.cx.cn.cxquartz.service.quartz.SbtdspsrService;
import com.cx.cn.cxquartz.service.quartz.TraffAlarmRecordService;
import com.cx.cn.cxquartz.service.quartz.impl.ResultService;
import com.cx.cn.cxquartz.service.quartz.impl.VideoRTSPorURLService;
import com.cx.cn.cxquartz.util.DateUtils;
import com.cx.cn.cxquartz.util.JsonUtil;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.BatchingRabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
/**
* 消息处理并推送第三方
*/
@Component
public class SnapTaskConsumer implements BaseConsumer {
private static final Logger logger = LoggerFactory.getLogger(SnapTaskConsumer.class);
@Autowired
ResultService resultService;
@Value("${local.czurl}")
private String czurl;
@Value("${local.fxurl}")
private String fxurl;
@Value("${file.rootpath}")
private String czrooturl;
@Autowired
private VideoRTSPorURLService videoRTSPorURLService;
@Autowired
private SbtdspsrService sbtdspsrService;
@Autowired
BatchingRabbitTemplate batchQueueRabbitTemplate;
@Autowired
private RestTemplate restTemplate;
/**
* 消息消费入口
*
* @param message
* @param channel
* @throws IOException
*/
@Override
public void consume(Message message, Channel channel) throws IOException {
logger.info("SnapTaskConsumer 收到消息: {}", message.toString());
QuartzTaskInformations msg = MessageHelper.msgToObj(message, QuartzTaskInformations.class);
if (null != msg) {
try {
//调用抽帧服务
String devicecode = msg.getExecuteparamter();
String rtsporhls = "";
logger.info("开始消费消息{}", msg.getId());
//如果设备编号是用一次废一次的，此刻需要现场取得rtsp
if (null != devicecode && devicecode.startsWith("33") && devicecode.length() != 18) {
//调用抽帧服务
String token = videoRTSPorURLService.getRTSPAccessToekenByDeviceCode(devicecode);
rtsporhls = videoRTSPorURLService.getRTSPByDeviceCode(token, devicecode);
} else {
//取表里最新的rtsp 或者hls 的值
rtsporhls = sbtdspsrService.getRtspOrHLSByDeviceCode(devicecode);
}
if (rtsporhls.equals("")) {
//尝试重新抽帧一遍
String token = videoRTSPorURLService.getRTSPAccessToekenByDeviceCode(devicecode);
rtsporhls = videoRTSPorURLService.getRTSPByDeviceCode(token, devicecode);
if (rtsporhls.equals("")) {
logger.error(devicecode + "rtsp 、hls 地址为空");
return;
}
}
//将rtsp 作为参数调用抽帧服务
// String result="{\n" +
// " \"ret\": 0,\n" +
// " \"desc\": \"succ!\",\n" +
// " \"resourcePath\": \"http://zjh189.ncpoi.cc:7080/download/202109/08/33050300001327599605/33050300001327599605_20210908_134131031.jpg\",\n" +
// " \"localuri\": \"/home/ubuntu/pictures/slice/202109/08/33050300001327599605/33050300001327599605_20210908_134131031.jpg\",\n" +
// " \"timestamp\": \"2021-09-08 13:41:31.031\",\n" +
// " \"devicecode\": \"33050300001327599605\"\n" +
// "}";
HttpHeaders headers = new HttpHeaders();
headers.setContentType(MediaType.APPLICATION_JSON);
Map map = new HashMap();
map.put("resourcePath", czrooturl + "/" + DateUtils.formatCurrDatefileYMD() + "/" + devicecode + "/" + devicecode + "_" + DateUtils.formatCurrDayNoSign() + "_" + DateUtils.formatCurrDateHHmmss() + ".jpg");
map.put("deviceID", devicecode);
map.put("resourceParam", rtsporhls);
HttpEntity<Map> requestEntity = new HttpEntity<>(map, headers);
logger.info("抽帧参数:{}", JsonUtil.objToStr(map));
Map resultmap = restTemplate.postForObject(czurl, requestEntity, Map.class);
if (null != resultmap.get("ret")) {
if (resultmap.get("ret").toString().equals("0")
&& null != resultmap.get("resourcePath")
&& !resultmap.get("resourcePath").toString().equals("")) {
//抽帧结果放到rabbttmq 中，根据msg 的检测metatype ,分别派发到不同的queue中，方便以后10条10条的去皮皮昂分析
Map m = new HashMap();
m.put("task", JsonUtil.objToStr(msg));
m.put("result", JsonUtil.objToStr(resultmap));
CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
batchQueueRabbitTemplate.send(QueueConstants.QueueTaskEnum.QUEUE_TASK_ENUM.getExchange(),
QueueConstants.QueueTaskEnum.QUEUE_TASK_ENUM.getRouteKey() + "_QST",
MessageHelper.objToMsg(m),
correlationData);
} else {
logger.error("抽帧失败:{}", JsonUtil.objToStr(resultmap));
}
} else {
logger.error("返回状态码为null");
}
} catch (Exception ex) {
logger.error("消费消息{}error:{}", msg.getId(), ex.toString());
}
}
}
}
src/main/java/com/cx/cn/cxquartz/rabbitmq/comsumer/TaskQSTConsumer.java
View file @ 50cef139
package com.cx.cn.cxquartz.rabbitmq.comsumer;
import com.cx.cn.cxquartz.common.Constants;
import com.cx.cn.cxquartz.helper.MessageHelper;
import com.cx.cn.cxquartz.service.quartz.TraffAlarmRecordService;
import com.cx.cn.cxquartz.service.quartz.impl.EventWriteService;
import com.cx.cn.cxquartz.service.quartz.impl.TaskRecog;
import com.cx.cn.cxquartz.util.CommonUtil;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
......@@ -22,7 +19,7 @@ import java.util.List;
import java.util.Map;
/**
* 消息处理并推送第三方
* 消息处理并推送第三方
*/
@Component
public class TaskQSTConsumer{
......@@ -50,7 +47,7 @@ public class TaskQSTConsumer{
@Value("${voice.eventId}")
private Integer eventId;
/**
* 消息消费入口
* 消息消费入口
*
* @param messageList
* @param channel
......
src/main/java/com/cx/cn/cxquartz/rabbitmq/comsumer/listener/EventProcessingReceiver.java
View file @ 50cef139
package com.cx.cn.cxquartz.rabbitmq.comsumer.listener;
import com.cx.cn.cxquartz.rabbitmq.QueueConstants;
import com.cx.cn.cxquartz.rabbitmq.comsumer.BaseConsumer;
import com.cx.cn.cxquartz.rabbitmq.comsumer.BaseConsumerProxy;
import com.cx.cn.cxquartz.rabbitmq.comsumer.EventProcessingConsumer;
......@@ -9,7 +8,6 @@ import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
......@@ -25,7 +23,7 @@ public class EventProcessingReceiver {
@Autowired
private TraffPictureService traffPictureService;
@RabbitListener(queues = QueueConstants.QueueEventProcessingConsumer.QUEUE)
// @RabbitListener(queues = QueueConstants.QueueEventProcessingConsumer.QUEUE)
public void process(Message message, Channel channel) {
try {
BaseConsumerProxy baseConsumerProxy = new BaseConsumerProxy(eventProcessingConsumer, traffPictureService);
......
src/main/java/com/cx/cn/cxquartz/rabbitmq/comsumer/listener/SendtoVoiceAlarmReceiver.java
View file @ 50cef139
package com.cx.cn.cxquartz.rabbitmq.comsumer.listener;
import com.cx.cn.cxquartz.rabbitmq.QueueConstants;
import com.cx.cn.cxquartz.rabbitmq.comsumer.BaseConsumer;
import com.cx.cn.cxquartz.rabbitmq.comsumer.BaseConsumerProxy;
import com.cx.cn.cxquartz.rabbitmq.comsumer.SendToVoiceConsumer;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
......@@ -25,7 +27,7 @@ public class SendtoVoiceAlarmReceiver {
* @param channel
* @throws IOException
*/
// @RabbitListener(queues = QueueConstants.QueueSendToVoiceConsumer.QUEUE)
@RabbitListener(queues = QueueConstants.QueueSendToVoiceConsumer.QUEUE)
public void consume(Message message, Channel channel) throws IOException {
BaseConsumerProxy baseConsumerProxy = new BaseConsumerProxy(sendToVoiceConsumer);
BaseConsumer proxy = (BaseConsumer) baseConsumerProxy.getProxy();
......
src/main/java/com/cx/cn/cxquartz/rabbitmq/comsumer/listener/SnapTaskReceiver.java 0 → 100644
View file @ 50cef139
package com.cx.cn.cxquartz.rabbitmq.comsumer.listener;
import com.cx.cn.cxquartz.rabbitmq.QueueConstants;
import com.cx.cn.cxquartz.rabbitmq.comsumer.BaseConsumer;
import com.cx.cn.cxquartz.rabbitmq.comsumer.BaseConsumerProxy;
import com.cx.cn.cxquartz.rabbitmq.comsumer.SnapTaskConsumer;
import com.cx.cn.cxquartz.service.quartz.TraffPictureService;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
/**
* 告警信息结果分析
*/
@Component
public class SnapTaskReceiver {
private static final Logger logger = LoggerFactory.getLogger(SnapTaskReceiver.class);
@Autowired
private SnapTaskConsumer snapTaskConsumer;
@Autowired
private TraffPictureService traffPictureService;
@RabbitListener(queues = QueueConstants.QueueSnapTaskConsumer.QUEUE)
public void process(Message message, Channel channel) {
try {
BaseConsumerProxy baseConsumerProxy = new BaseConsumerProxy(snapTaskConsumer, traffPictureService);
BaseConsumer proxy = (BaseConsumer) baseConsumerProxy.getProxy();
if (null != proxy) {
proxy.consume(message, channel);
}
} catch (Exception e) {
logger.error("告警信息结果分析 error:{}",e);
}
}
}
src/main/java/com/cx/cn/cxquartz/rabbitmq/comsumer/listener/TaskQSTReceiver.java
View file @ 50cef139
......@@ -8,7 +8,6 @@ import org.springframework.amqp.core.BatchMessageListener;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import java.util.List;
......@@ -31,7 +30,7 @@ public class TaskQSTReceiver implements BatchMessageListener {
taskQSTConsumer.consume(messages,null);
}
} catch (Exception e) {
logger.error("批量分析图片 error:{}",e);
logger.error("批量分析图片 error:{}",e.toString());
}
}
}
src/main/java/com/cx/cn/cxquartz/rabbitmq/producer/callback/ConsumerConfirmAndReturnCallback.java
View file @ 50cef139
......@@ -13,7 +13,7 @@ import javax.annotation.PostConstruct;
import java.nio.charset.StandardCharsets;
/**
* 消息发送确认的回调
* 消息发送确认的回调
*/
@Component
public class ConsumerConfirmAndReturnCallback implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {
......@@ -23,18 +23,18 @@ public class ConsumerConfirmAndReturnCallback implements RabbitTemplate.ConfirmC
private RabbitTemplate rabbitTemplate;
/**
* PostConstruct: 用于在依赖关系注入完成之后需要执行的方法上，以执行任何初始化.
* PostConstruct: 用于在依赖关系注入完成之后需要执行的方法上，以执行任何初始化.
*/
@PostConstruct
public void init() {
//指定 ConfirmCallback
//指定 ConfirmCallback
rabbitTemplate.setConfirmCallback(this);
//指定 ReturnCallback
//指定 ReturnCallback
rabbitTemplate.setReturnCallback(this);
}
/**
* 消息从交换机成功到达队列，spring.rabbitmq.publisher-returns=true
* 消息从交换机成功到达队列，spring.rabbitmq.publisher-returns=true
*/
@Override
public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
......@@ -43,19 +43,19 @@ public class ConsumerConfirmAndReturnCallback implements RabbitTemplate.ConfirmC
}
/**
* 消息找不到对应的Exchange会先触发此方法
* 如果消息没有到达交换机,则该方法中isSendSuccess = false,error为错误信息;
* 如果消息正确到达交换机,则该方法中isSendSuccess = true;
* 需要开启 confirm 确认机制
* 消息找不到对应的Exchange会先触发此方法
* 如果消息没有到达交换机,则该方法中isSendSuccess = false,error为错误信息;
* 如果消息正确到达交换机,则该方法中isSendSuccess = true;
* 需要开启 confirm 确认机制
* spring.rabbitmq.publisher-confirms=true
*/
@Override
public void confirm(CorrelationData correlationData, boolean isSendSuccess, String error) {
if (correlationData != null) {
if (isSendSuccess) {
logger.info("confirm回调方法->消息成功发送到交换机！");
logger.info("confirm回调方法->消息成功发送到交换机！");
} else {
logger.info("confirm回调方法->消息[{}]发送到交换机失败！，原因 : [{}]", correlationData, error);
logger.info("confirm回调方法->消息[{}]发送到交换机失败！，原因 : [{}]", correlationData, error);
}
}
}
......
src/main/java/com/cx/cn/cxquartz/redis/OrderConsumer.java
View file @ 50cef139
package com.cx.cn.cxquartz.redis;
import com.alibaba.druid.util.HttpClientUtils;
import com.cx.cn.cxquartz.bean.QuartzTaskInformations;
import com.cx.cn.cxquartz.helper.MessageHelper;
import com.cx.cn.cxquartz.rabbitmq.QueueConstants;
......@@ -8,8 +7,6 @@ import com.cx.cn.cxquartz.service.quartz.SbtdspsrService;
import com.cx.cn.cxquartz.service.quartz.impl.VideoRTSPorURLService;
import com.cx.cn.cxquartz.util.DateUtils;
import com.cx.cn.cxquartz.util.JsonUtil;
import com.cx.cn.cxquartz.vo.ResultObj;
import com.rabbitmq.tools.json.JSONUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
......@@ -19,13 +16,10 @@ import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Component;
import org.springframework.util.DigestUtils;
import org.springframework.web.client.RestTemplate;
import javax.annotation.PostConstruct;
import java.io.File;
import java.util.*;
@Component
......@@ -35,12 +29,8 @@ public class OrderConsumer implements Consumer {
private String czurl;
@Value("${local.fxurl}")
private String fxurl;
@Value("${file.recogurl}")
private String recogurl;
@Value("${file.model}")
private String model;
@Value("${local.czrooturl}")
@Value("${file.rootpath}")
private String czrooturl;
......@@ -62,8 +52,6 @@ public class OrderConsumer implements Consumer {
orderConsumer.batchQueueRabbitTemplate = this.batchQueueRabbitTemplate;
orderConsumer.czurl=this.czurl;
orderConsumer.fxurl=this.fxurl;
orderConsumer.recogurl=this.recogurl;
orderConsumer.model=this.model;
orderConsumer.videoRTSPorURLService=this.videoRTSPorURLService;
orderConsumer.sbtdspsrService=this.sbtdspsrService;
orderConsumer.restTemplate=this.restTemplate;
......@@ -76,31 +64,31 @@ public class OrderConsumer implements Consumer {
if(message instanceof QuartzTaskInformations){
QuartzTaskInformations msg =(QuartzTaskInformations) message;
try {
//调用抽帧服务
//调用抽帧服务
String devicecode=msg.getExecuteparamter();
String rtsporhls="";
log.info("开始消费消息{}", msg.getId());
//如果设备编号是用一次废一次的，此刻需要现场取得rtsp
log.info("开始消费消息{}", msg.getId());
//如果设备编号是用一次废一次的，此刻需要现场取得rtsp
if(null!=devicecode&&devicecode.startsWith("33") && devicecode.length()!=18){
//调用抽帧服务
//调用抽帧服务
String token= orderConsumer.videoRTSPorURLService.getRTSPAccessToekenByDeviceCode(devicecode);
rtsporhls=orderConsumer.videoRTSPorURLService.getRTSPByDeviceCode(token,devicecode);
}
else{
//取表里最新的rtsp 或者hls 的值
//取表里最新的rtsp 或者hls 的值
rtsporhls=orderConsumer.sbtdspsrService.getRtspOrHLSByDeviceCode(devicecode);
}
if(rtsporhls.equals("")){
//尝试重新抽帧一遍
//尝试重新抽帧一遍
String token= orderConsumer.videoRTSPorURLService.getRTSPAccessToekenByDeviceCode(devicecode);
rtsporhls=orderConsumer.videoRTSPorURLService.getRTSPByDeviceCode(token,devicecode);
if(rtsporhls.equals("")) {
log.error(devicecode + "rtsp 、hls 地址为空");
log.error(devicecode + "rtsp 、hls 地址为空");
return;
}
}
//将rtsp 作为参数调用抽帧服务
//将rtsp 作为参数调用抽帧服务
// String result="{\n" +
......@@ -114,20 +102,20 @@ public class OrderConsumer implements Consumer {
HttpHeaders headers = new HttpHeaders();
headers.setContentType(MediaType.APPLICATION_JSON);
Map map=new HashMap();
map.put("resourcePath",orderConsumer.czrooturl+"/"+ DateUtils.formatCurrDateYYMM()+"/"+devicecode+"/"+devicecode+"_"+DateUtils.formatCurrDayNoSign()+"_"+DateUtils.formatCurrDateHHmmss()+".jpg");
map.put("resourcePath",orderConsumer.czrooturl+"/"+ DateUtils.formatCurrDatefileYMD()+"/"+devicecode+"/"+devicecode+"_"+DateUtils.formatCurrDayNoSign()+"_"+DateUtils.formatCurrDateHHmmss()+".jpg");
map.put("deviceID",devicecode);
map.put("resourceParam",rtsporhls);
HttpEntity<Map> requestEntity = new HttpEntity<>(map, headers);
String response= orderConsumer.restTemplate.postForObject(orderConsumer.czurl,requestEntity, String.class);
Map resultmap =JsonUtil.strToObj(response,Map.class);
log.info("抽帧参数:{}",JsonUtil.objToStr(map));
Map resultmap= orderConsumer.restTemplate.postForObject(orderConsumer.czurl,requestEntity, Map.class);
if(null!=resultmap.get("ret")) {
if (resultmap.get("ret").toString().equals("0")
&& null!=resultmap.get("resourcePath")
&& !resultmap.get("resourcePath").toString().equals("")) {
//抽帧结果放到rabbttmq 中，根据msg 的检测metatype ,分别派发到不同的queue中，方便以后10条10条的去皮皮昂分析
//抽帧结果放到rabbttmq 中，根据msg 的检测metatype ,分别派发到不同的queue中，方便以后10条10条的去皮皮昂分析
Map m = new HashMap();
m.put("task", JsonUtil.objToStr(msg));
m.put("result", response);
m.put("result", JsonUtil.objToStr(resultmap));
CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
orderConsumer.batchQueueRabbitTemplate.send(QueueConstants.QueueTaskEnum.QUEUE_TASK_ENUM.getExchange(),
QueueConstants.QueueTaskEnum.QUEUE_TASK_ENUM.getRouteKey() + "_QST",
......@@ -135,12 +123,12 @@ public class OrderConsumer implements Consumer {
correlationData);
}
else{
log.error("抽帧失败",JsonUtil.objToStr(resultmap));
log.error("抽帧失败:{}",JsonUtil.objToStr(resultmap));
}
}
else{
log.error("返回状态码为null");
log.error("返回状态码为null");
}
// String roistr= URLEncoder.encode("[")+msg.getObjectx()+","+msg.getObjecty()+","+msg.getObjectw()+","+msg.getObjecth()+URLEncoder.encode("]");
......@@ -152,10 +140,10 @@ public class OrderConsumer implements Consumer {
// log.info(" deviceCode:{} ",msg.getExecuteparamter());
// String result = HttpClientUtil.doGet(orderConsumer.recogurl + "?deviceCode=" + msg.getExecuteparamter() + "&model=" + orderConsumer.model + "&roi=" + roistr);
// log.info(" picture result:{} ",result);
// if (null != result && result.contains("ret"))//放入rabbitmq.数据由消费者去处理
// if (null != result && result.contains("ret"))//放入rabbitmq.数据由消费者去处理
// {
// Map objresult = JsonUtil.strToObj(result, Map.class);
// //处理消息
// //处理消息
// if (null != objresult.get("ret") && objresult.get("ret").toString().equals("0")
// && null != objresult.get("desc")
// && objresult.get("desc").toString().contains("succ")) {
......@@ -164,7 +152,7 @@ public class OrderConsumer implements Consumer {
// m.put("result", result);
// String msgId = UUID.randomUUID().toString();
// CorrelationData correlationData = new CorrelationData(msgId);
// //根绝不同的检测事件对象metatype分发到不同的通道
// //根绝不同的检测事件对象metatype分发到不同的通道
//
// orderConsumer.rabbitTemplate.convertAndSend(QueueConstants.QueueTaskEnum.QUEUE_TASK_ENUM.getExchange(),
// QueueConstants.QueueTaskEnum.QUEUE_TASK_ENUM.getRouteKey(),
......@@ -172,10 +160,9 @@ public class OrderConsumer implements Consumer {
// correlationData);
// }
// }
//添加具体的消费逻辑，修改数据库什么的
log.info("消费消息{}完成", msg.getId());
//添加具体的消费逻辑，修改数据库什么的
}catch (Exception ex){
log.error("消费消息{}error:{}", msg.getId(), ex.toString());
log.error("消费消息{}error:{}", msg.getId(), ex.toString());
}
}
}
......
src/main/java/com/cx/cn/cxquartz/redis/container/QueueListener.java
View file @ 50cef139
......@@ -7,6 +7,7 @@ import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.RedisTemplate;
import java.util.concurrent.TimeUnit;
public class QueueListener implements Runnable {
public static final Logger log = LoggerFactory.getLogger(QueueListener.class);
private RedisTemplate redisTemplate;
......@@ -21,26 +22,27 @@ public class QueueListener implements Runnable {
this.consumer = consumer;
}
public QueueListener() {
}
/**
* 使用队列右出获取消息
* 没获取到消息则线程 sleep 一秒，减少资源浪费
*/
@Override
public void run() {
while (RedisMQConsumerContainer.RUNNING){
Object message = redisTemplate.opsForList().rightPop(queue,500,TimeUnit.MICROSECONDS);
if( message instanceof QuartzTaskInformations){
consumer.consume(message);
}else{
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
log.error("RedisMQConsumer:{}",e.toString());
try {
while (true) {
Object message = redisTemplate.opsForList().rightPop(queue);
if (message instanceof QuartzTaskInformations) {
consumer.consume(message);
} else {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
log.error("RedisMQConsumer error:{}", e.toString());
}
}
}
} catch (Exception e) {
log.error("RedisMQConsumer error:{}", e.toString());
}
}
}
\ No newline at end of file
src/main/java/com/cx/cn/cxquartz/redis/container/RedisMQConsumerContainer.java
View file @ 50cef139
......@@ -13,7 +13,7 @@ import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
@Component
//@Component
public class RedisMQConsumerContainer {
public static final Logger log = LoggerFactory.getLogger(RedisMQConsumerContainer.class);
public static boolean RUNNING;
......@@ -44,7 +44,7 @@ public class RedisMQConsumerContainer {
this.executor.shutdown();
log.info("QueueListener exiting.");
while (!this.executor.isTerminated()) {
log.error("this.executor.isTerminated()");
}
log.info("QueueListener exited.");
}
......
src/main/java/com/cx/cn/cxquartz/service/quartz/impl/EventWriteService.java
View file @ 50cef139
......@@ -70,9 +70,6 @@ public class EventWriteService {
@Value("${file.outpath}")
private String outpath;
@Value("${file.uploadurl}")
private String uploadurl;
public void sendEvent(TraffpictureParam traffpictureParamresult, TraffrecordData sendtozhiui) {
if (traffpictureParamresult.getTargetnum().contains("/")) {
sendtozhiui.setAlarmnum(Integer.parseInt(traffpictureParamresult.getTargetnum().split("/")[1]));
......@@ -107,10 +104,10 @@ public class EventWriteService {
traffPictureMapper.updateTraffpicturePushStatus(traffpictureParamresult);
}
public void sendVoice(VoiceData voice) {
public void sendVoice(VoiceData voice,String url ) {
VoiceResultObj result;
try {
result = sendVoioceMessage(voice);
result = sendVoioceMessage(voice, url);
if (result.getCode() == 1) {
log.info(" push to voice success");
} else {
......@@ -190,10 +187,10 @@ public class EventWriteService {
traffPictureMapper.updateTraffpicturePushStatus(traffpictureParamresult);
}
private VoiceResultObj sendVoioceMessage(VoiceData voiceData) throws InterruptedException, ExecutionException, TimeoutException {
private VoiceResultObj sendVoioceMessage(VoiceData voiceData,String url) throws InterruptedException, ExecutionException, TimeoutException {
HttpHeaders headers = new HttpHeaders();
HttpEntity<VoiceData> requestEntity = new HttpEntity<>(voiceData, headers);
return restTemplate.postForObject(voiceurl, requestEntity, VoiceResultObj.class);
return restTemplate.postForObject(url, requestEntity, VoiceResultObj.class);
}
......
src/main/java/com/cx/cn/cxquartz/service/quartz/impl/ResultService.java
View file @ 50cef139
......@@ -86,7 +86,7 @@ public class ResultService {
getPoi(taskinfo, roiarray);
//图片划线并上传
String basepath = DateUtils.formatCurrDayYM() + File.separator + DateUtils.formatCurrDayDD() + File.separator + devicecode;
String filename = devicecode + "_" + DateUtils.parseDateToStrNoSign(timestamp) + "_result.jpg";
String filename = devicecode + "_" + DateUtils.parseDateToStrNoSign(timestamp) + "_"+ recordtype + "_result.jpg";
String filenameurl = File.separator + outpath + File.separator + basepath + File.separator + filename;
jobTjParam.setImageUrl(weburl + filenameurl);
traffpictureParamresult.setImagedata(filenameurl); //获得点位
......
src/main/java/com/cx/cn/cxquartz/service/quartz/impl/TaskRecog.java
View file @ 50cef139
......@@ -26,10 +26,6 @@ public class TaskRecog implements InitializingBean {
private static final Logger logger = LoggerFactory.getLogger(TaskRecog.class);
@Value("${file.recogurl}")
String recogurl;
@Value("${file.recogqsturl}")
String recogqsturl;
......@@ -39,7 +35,7 @@ public class TaskRecog implements InitializingBean {
private RabbitTemplate rabbitTemplate;
public void sender(List<Map> list) throws Exception {
public void sender(List<Map> list) {
String result="{\n" +
" \"ret\": 0,\n" +
......@@ -50,6 +46,7 @@ public class TaskRecog implements InitializingBean {
" \"devicecode\": \"33050300001327599605\"\n" +
"}";
//调用服务，10个一组批量调用分析服务
try{
int i=0;
GoalStructureParam goalSparam=new GoalStructureParam();
List<ImageList> imglist=new ArrayList<>();
......@@ -78,15 +75,23 @@ public class TaskRecog implements InitializingBean {
img.setData(picture.get("resourcePath").toString());
imglist.add(img);
recordtypeMap.put(i,new TaskResult(param.getVideoid(),
param.getRecordtype(),picture.get("timestamp").toString(),param.getObjectx(),param.getObjecty(), param.getObjectw() , param.getObjecth(),
picture.get("resourcePath").toString(), param.getMetatype(), param.getSendtype())
param.getRecordtype(),
picture.get("timestamp").toString(),
param.getObjectx(),param.getObjecty(),
param.getObjectw() , param.getObjecth(),
picture.get("resourcePath").toString(),
param.getMetatype(), param.getSendtype(),
param.getUrl()
)
);
i++;
}
}
goalSparam.setImageList(imglist);
goalSparam.setModel(model);
String resultstr = restTemplate.postForObject(recogqsturl, goalSparam, String.class);
logger.info("参数：{}",JsonUtil.objToStr(goalSparam));
String resultstr = restTemplate.postForObject(recogqsturl, goalSparam, String.class);
logger.info("分析结果：{}",resultstr);
// try {
Map resulMap = JsonUtil.strToObj(resultstr, Map.class);
if (null != resulMap.get("ret") && resulMap.get("ret").equals("200")) {
......@@ -95,7 +100,6 @@ public class TaskRecog implements InitializingBean {
logger.info(" objectresult is empty");
return ;
}
for (Integer key : recordtypeMap.keySet()) {
List<Map> senderresult = new ArrayList<>();
//遍历所有返回结果
......@@ -114,13 +118,11 @@ public class TaskRecog implements InitializingBean {
QueueConstants.QueueEventProcessingEnum.QUEUE_EVENT_PROCESSING_ENUM.getRouteKey(),
MessageHelper.objToMsg(map),
correlationData);
}
}
// } catch (Exception ex) {
// logger.error(ex.toString());
// }
} catch (Exception ex) {
logger.error(ex.toString());
}
}
......
src/main/java/com/cx/cn/cxquartz/service/quartz/impl/VideoRTSPorURLService.java
View file @ 50cef139
......@@ -170,6 +170,7 @@ public class VideoRTSPorURLService {
HttpEntity<String> httpEntity = new HttpEntity(headers);
ResponseEntity<String> exchange = restTemplate.exchange(rtspurl + "?deviceCode=" + deviceCode, HttpMethod.GET, httpEntity, String.class);
String body = exchange.getBody();
logger.info("get rtsp result:{}",body);
Sbtdspsr sbtdspsr = new Sbtdspsr();
try {
Map result = new ObjectMapper().readValue(body, Map.class);
......@@ -204,7 +205,6 @@ public class VideoRTSPorURLService {
//根据token获得rtsp 地址
} catch (Exception ex) {
logger.error("getRTSPByDeviceCode" + ex.toString());
}
return rtsp;
......@@ -216,3 +216,4 @@ public class VideoRTSPorURLService {
}
src/main/java/com/cx/cn/cxquartz/util/DateUtils.java
View file @ 50cef139
......@@ -21,6 +21,8 @@ public class DateUtils {
private static final DateTimeFormatter DD= DateTimeFormat.forPattern("dd");
private static final DateTimeFormatter YYMM= DateTimeFormat.forPattern("yyyy/MM");
private static final DateTimeFormatter HHmmss= DateTimeFormat.forPattern("HHmmssSSS");
private static final DateTimeFormatter fileYMD = DateTimeFormat.forPattern("yyyyMM/dd");
private static final DateTimeFormatter YYMMDD_HMS = DateTimeFormat.forPattern("yyyyMMdd_HHmmss");
public static String formatCurrDate(){
return formatCurrDateByType(YMD_HMS);
......@@ -40,7 +42,9 @@ public class DateUtils {
public static String formatCurrDateYYMM(){
return formatCurrDateByType(YYMM);
}
public static String formatCurrDatefileYMD(){
return formatCurrDateByType(fileYMD);
}
public static String formatCurrDayNoSign(){
return formatCurrDateByType(YMD);
}
......@@ -90,7 +94,7 @@ public class DateUtils {
}
public static String formatDateToNoSign(String date){
return DateTime.parse(date,YMD_HMS).toString(YMDHMS);
return DateTime.parse(date,YMD_HMS).toString(YYMMDD_HMS);
}
private static String formatCurrDateByType(DateTimeFormatter dateTimeFormatter){
......
src/main/java/com/cx/cn/cxquartz/util/XmlUtils.java
View file @ 50cef139
......@@ -110,6 +110,7 @@ public class XmlUtils {
String elementName = element.getName();
if(elementName.equalsIgnoreCase("deviceId")){
obj.setDeviceId(element.getStringValue());
logger.info("deviceId:{}",element.getStringValue());
}
for (Iterator<Element> iter = element.elementIterator(); iter.hasNext(); ) {
Element elementiter = iter.next();
......@@ -118,6 +119,7 @@ public class XmlUtils {
if(elementchilsthree.getName().equalsIgnoreCase("datetime"))
{
obj.setDateTime(elementchilsthree.getStringValue());
logger.info("datetime:{}",elementchilsthree.getStringValue());
}
}
}
......
src/main/java/com/cx/cn/cxquartz/vo/Autosnaptaskinfo.java
View file @ 50cef139
......@@ -8,7 +8,10 @@ public class Autosnaptaskinfo {
private String starthour;
private String endhour;
private String recordtype;
private String region;
private String regionx;
private String regiony;
private String regionw;
private String regionh;
private String spId;
private String status;
private String cjrq;
......@@ -27,21 +30,27 @@ public class Autosnaptaskinfo {
this.sendtype = sendtype;
}
public Autosnaptaskinfo(String taskid, String taskname, String devicenum, String starthour, String endhour, String recordtype, String region, String spId, String status, String objectType, String sendurl, String threshold, String algorithmfrom,String sendtype) {
public Autosnaptaskinfo() {
}
public Autosnaptaskinfo(String taskid, String taskname, String devicenum, String starthour, String endhour, String recordtype, String regionx, String regiony, String regionw, String regionh, String spId, String status, String objectType, String sendurl, String threshold, String algorithmfrom, String sendtype) {
this.taskid = taskid;
this.taskname = taskname;
this.devicenum = devicenum;
this.starthour = starthour;
this.endhour = endhour;
this.recordtype = recordtype;
this.region = region;
this.regionx = regionx;
this.regiony = regiony;
this.regionw = regionw;
this.regionh = regionh;
this.spId = spId;
this.status = status;
this.objectType = objectType;
this.sendurl = sendurl;
this.threshold = threshold;
this.algorithmfrom = algorithmfrom;
this.sendtype=sendtype;
this.sendtype = sendtype;
}
public String getTaskid() {
......@@ -92,12 +101,36 @@ public class Autosnaptaskinfo {
this.recordtype = recordtype;
}
public String getRegion() {
return region;
public String getRegionx() {
return regionx;
}
public void setRegionx(String regionx) {
this.regionx = regionx;
}
public String getRegiony() {
return regiony;
}
public void setRegiony(String regiony) {
this.regiony = regiony;
}
public String getRegionw() {
return regionw;
}
public void setRegionw(String regionw) {
this.regionw = regionw;
}
public String getRegionh() {
return regionh;
}
public void setRegion(String region) {
this.region = region;
public void setRegionh(String regionh) {
this.regionh = regionh;
}
public String getSpId() {
......
src/main/resources/application-dev.yml
View file @ 50cef139
......@@ -2,10 +2,10 @@ server:
port: 4082
spring:
datasource:
url: jdbc:mysql://172.16.24.29:3306/imagepro?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=UTC&allowPublicKeyRetrieval=true
url: jdbc:mysql://172.16.24.29:3306/imagepro?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=GMT%2B8
username: root
password: 123456
driver-class-name: com.mysql.jdbc.Driver
driver-class-name: com.mysql.cj.jdbc.Driver
type: com.alibaba.druid.pool.DruidDataSource
filters: stat
maxActive: 1000
......
src/main/resources/application-dev85.yml
View file @ 50cef139
......@@ -2,10 +2,10 @@ server:
port: 4085
spring:
datasource:
url: jdbc:mysql://172.16.24.29:3306/imagepro?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=UTC&allowPublicKeyRetrieval=true
url: jdbc:mysql://172.16.24.29:3306/imagepro?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=GMT%2B8
username: root
password: 123456
driver-class-name: com.mysql.jdbc.Driver
driver-class-name: com.mysql.cj.jdbc.Driver
type: com.alibaba.druid.pool.DruidDataSource
filters: stat
maxActive: 1000
......
src/main/resources/application-job.yml
View file @ 50cef139
......@@ -2,10 +2,10 @@ server:
port: 4084
spring:
datasource:
url: jdbc:mysql://172.16.24.29:3306/imagepro?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=UTC&allowPublicKeyRetrieval=true
url: jdbc:mysql://172.16.24.29:3306/imagepro?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=GMT%2B8
username: root
password: 123456
driver-class-name: com.mysql.jdbc.Driver
driver-class-name: com.mysql.cj.jdbc.Driver
type: com.alibaba.druid.pool.DruidDataSource
filters: stat
maxActive: 1000
......
src/main/resources/application-local.yml
View file @ 50cef139
server:
port: 8089
port: 8083
spring:
datasource:
url: jdbc:mysql://localhost:3306/imagepro?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=UTC&allowPublicKeyRetrieval=true
url: jdbc:mysql://localhost:3306/imagepro?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=GMT%2B8
username: root
password: root
driver-class-name: com.mysql.jdbc.Driver
driver-class-name: com.mysql.cj.jdbc.Driver
type: com.alibaba.druid.pool.DruidDataSource
filters: stat
maxActive: 1000
......@@ -65,7 +65,7 @@ local:
job: 0
czurl: http://zjh189.ncpoi.cc:7780/getDeviceSnapshot
czrooturl: /home/ubuntu/pictures/slice
fxurl: http://localhost:8089/ext/getDeviceSnapshotAndRecognize
fxurl: http://localhost:8083/ext/getDeviceSnapshotAndRecognize
file:
rtspurl: http://zjh189.ncpoi.cc:7080/getDeviceSnapshot
......
src/main/resources/application-test.yml deleted 100644 → 0
View file @ b0e0ee4f
server:
port: 8089
spring:
datasource:
url: jdbc:mysql://172.16.24.29:3306/hzdxtest?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=UTC&allowPublicKeyRetrieval=true
username: hzdxtest
password: 1qaz@wsx
driver-class-name: com.mysql.jdbc.Driver
type: com.alibaba.druid.pool.DruidDataSource
filters: stat
maxActive: 1000
initialSize: 100
maxWait: 60000
minIdle: 500
timeBetweenEvictionRunsMillis: 60000
minEvictableIdleTimeMillis: 300000
testWhileIdle: true
testOnBorrow: false
testOnReturn: false
poolPreparedStatements: true
maxOpenPreparedStatements: 20
local:
job: 0
czurl: http://localhost:8089/ext/getRTSP/1
fxurl: http://localhost:8089/ext/getDeviceSnapshotAndRecognize
countryside:
callbackurl: http://kvideo.51iwifi.com
file:
# rootpath: /home/ubuntu/pictures
recogurl: http://172.16.24.29:9098/images/recog
rtspurl: http://172.16.24.29:7180/getDeviceSnapshot
logging:
level:
com.cx.cn.cxquartz.dao:
debug
src/main/resources/application.yml
View file @ 50cef139
spring:
profiles:
active: local
active: devconsum
Servlet:
multipart:
max-file-size: 20MB
max-request-size: 100MB
mybatis:
type-aliases-package: com.cx.cn.cxquartz.bean
configuration:
......@@ -16,12 +20,15 @@ file:
uploadurl: http://172.16.24.29:7180/uploadResultFile
rootpath: /home/prod/pictures
outpath: result
redis:
database: 0
host: 172.16.24.29
port: 6379
countryside:
callbackurl: http://kvideo.51iwifi.com/hesc-mq/hesc/mq/receive/aiCallback
web:
url: http://zjh189.ncpoi.cc:20000
snapnote:
note1: /opt/data/public
note2: /opt/public
src/main/resources/log4j2.component.properties 0 → 100644
View file @ 50cef139
log4j2.formatMsgNoLookups=true
\ No newline at end of file
src/main/resources/logback-spring.xml
View file @ 50cef139
......@@ -2,7 +2,7 @@
<configuration debug="false" scan="false">
<springProperty scop="context" name="spring.application.name" source="spring.application.name" defaultValue=""/>
<!--<property name="log.path" value="/home/ubuntu/tar/zjdxtest/logs/${spring.application.name}"/>-->
<property name="log.path" value="/home/prod/deploy/HZDXService/logs/${spring.application.name}"/>
<property name="log.path" value="/home/ubuntu/tar/logs/taskconsumption/${spring.application.name}"/>
<!-- 彩色日志格式 -->
<property name="CONSOLE_LOG_PATTERN"
value="${CONSOLE_LOG_PATTERN:-%clr(%d{yyyy-MM-dd HH:mm:ss.SSS}){faint} %clr(${LOG_LEVEL_PATTERN:-%5p}) %clr(${PID:- }){magenta} %clr(---){faint} %clr([%15.15t]){faint} %clr(%-40.40logger{39}){cyan} %clr(:){faint} %m%n${LOG_EXCEPTION_CONVERSION_WORD:-%wEx}}"/>
......
src/main/resources/mapper/AutoSnapMapper.xml
View file @ 50cef139
......@@ -8,7 +8,10 @@
<result column="starthour" property="starthour"/>
<result column="endhour" property="endhour"/>
<result column="recordtype" property="recordtype"/>
<result column="region" property="region"/>
<result column="regionx" property="regionx"/>
<result column="regiony" property="regiony"/>
<result column="regionw" property="regionw"/>
<result column="regionh" property="regionh"/>
<result column="spId" property="spId"/>
<result column="status" property="status"/>
<result column="objectType" property="objectType"/>
......@@ -19,7 +22,8 @@
</resultMap>
<select id="query" parameterType="com.cx.cn.cxquartz.vo.Autosnaptaskinfo" resultMap="resultMap">
select taskid, taskname, devicenum, starthour, endhour, recordtype, region, spId, status, objectType, sendurl, threshold, algorithmfrom,sendtype from autosnaptaskinfo t where devicenum=#{devicenum} and status!=2
select taskid, taskname, devicenum, starthour, endhour, recordtype, regionx, regiony,regionw,regionh, spId, status, objectType, sendurl, threshold, algorithmfrom,sendtype from autosnaptaskinfo t where
devicenum=#{devicenum} and status!=2 and hour(now())>=starthour and hour(now()) <![CDATA[ <=]]> endhour
</select>
</mapper>
src/main/resources/mapper/SbtdspsrMapper.xml
View file @ 50cef139
......@@ -31,11 +31,11 @@
select * from sbtdspsr
</select>
<select id="getPeriodicseconds" resultType="java.lang.Long">
select str_to_date(min(urlnexttime),'%Y-%m-%d %H:%i:%s')-sysdate() from sbtdspsr;
select UNIX_TIMESTAMP(str_to_date(min(urlnexttime),'%Y-%m-%d %H:%i:%s'))-UNIX_TIMESTAMP(sysdate()) from sbtdspsr where left(sbbh,2)<![CDATA[ <>]]>'33' or ( left(sbbh,2)='33' and length(sbbh)=18)
</select>
<select id="getPerformedTasks" resultType="com.cx.cn.cxquartz.vo.Sbtdspsr">
select * from sbtdspsr where str_to_date(urlnexttime,'%Y-%m-%d %H:%i:%s')-sysdate()<![CDATA[ <=]]>120
select * from sbtdspsr where UNIX_TIMESTAMP(str_to_date(urlnexttime,'%Y-%m-%d %H:%i:%s'))-UNIX_TIMESTAMP(sysdate())<![CDATA[ <=]]>1200 and ( left(sbbh,2)<![CDATA[ <>]]>'33' or ( left(sbbh,2)='33' and length(sbbh)=18))
</select>
<select id="getRtspOrHLSByDeviceCode" parameterType="java.lang.String" resultType="java.lang.String">
......
src/test/java/com/quartz/cn/com/cx/cn/cxquartz/CXQuartzApplicationTests.java
View file @ 50cef139
package com.quartz.cn.com.cx.cn.cxquartz;
public class CXQuartzApplicationTests {
//
public class CXQuartzApplicationTests {
//
// @Test
// public void contextLoads() {
// String body = "{\"errorCode\":\"0\",\"errorMsg\":\"success\",\"data\":{\"rtspUri\":\"rtsp://60.188.134.18:30030/streamDelivery_JX04_136/1_0_52268/3d10d70b9e7012bd2f85f35fa3d5f8d3\",\"rtpUri\":\"rtp://60.188.134.18:11348/304\",\"rtmpUri\":\"rtmp://60.188.134.18:30031/streamDelivery_JX04_136/1_0_52268/3d10d70b9e7012bd2f85f35fa3d5f8d3\",\"hlsUri\":\"http://60.188.134.18:30120/streamDelivery_JX04_136/1_0_52268/3d10d70b9e7012bd2f85f35fa3d5f8d3.m3u8\",\"received\":true}}";
......@@ -31,6 +32,5 @@ package com.quartz.cn.com.cx.cn.cxquartz;
String ftpPath="ftp://"+ftppath+"/"+ftpfilepath;
System.out.println(ftpPath);
}
}
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