Skip to content

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

去掉不用的jar

parent 2fb71bd3
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 355 additions and 970 deletions
pom.xml
View file @ cb5db7cd
......@@ -8,8 +8,8 @@
<relativePath/> <!-- lookup parent from repository -->
</parent>
<modelVersion>4.0.0</modelVersion>
<groupId>com.quartz.cn</groupId>
<artifactId>VideoofCultural</artifactId>
<groupId>com.comsumer.cn</groupId>
<artifactId>TaskConsumption</artifactId>
<version>1.0</version>
<packaging>jar</packaging>
<description></description>
......@@ -27,12 +27,6 @@
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<!--mybatis-->
<dependency>
<groupId>org.mybatis.spring.boot</groupId>
......@@ -52,35 +46,6 @@
<artifactId>mysql-connector-java</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-thymeleaf</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-tomcat</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
<version>3.6</version>
</dependency>
<dependency>
<groupId>net.sf.json-lib</groupId>
<artifactId>json-lib</artifactId>
<version>2.4</version>
<classifier>jdk15</classifier>
</dependency>
<dependency>
<groupId>org.apache.poi</groupId>
<artifactId>poi</artifactId>
<version>3.15-beta2</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
......@@ -88,23 +53,11 @@
</dependency>
<dependency>
<groupId>commons-codec</groupId>
<artifactId>commons-codec</artifactId>
<version>1.9</version>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.google.code.gson</groupId>
<artifactId>gson</artifactId>
......@@ -115,47 +68,13 @@
<artifactId>jackson-datatype-joda</artifactId>
<version>2.9.6</version>
</dependency>
<!--quartz依赖-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-quartz</artifactId>
</dependency>
<dependency>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
<version>2.6</version>
</dependency>
<dependency>
<groupId>commons-net</groupId>
<artifactId>commons-net</artifactId>
<version>3.6</version>
</dependency>
<dependency>
<groupId>org.java-websocket</groupId>
<artifactId>Java-WebSocket</artifactId>
<version>1.3.8</version>
</dependency>
<!-- redis -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<!--kafka依赖-->
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
<!-- HttpClient依赖 -->
<dependency>
<groupId>commons-httpclient</groupId>
<artifactId>commons-httpclient</artifactId>
<version>3.1</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-mail</artifactId>
</dependency>
<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
......
src/main/java/com/cx/cn/cxquartz/config/EventProcessingConfig.java 0 → 100644
View file @ cb5db7cd
package com.cx.cn.cxquartz.config;
import com.cx.cn.cxquartz.rabbitmq.QueueConstants;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* 处理图片分析结果的队列
*/
@Configuration
public class EventProcessingConfig {
/**
* 创建交换机
*
* @return
*/
@Bean
public DirectExchange EventProcessingDirectExchange() {
return new DirectExchange(QueueConstants.QueueEventProcessingEnum.QUEUE_EVENT_PROCESSING_ENUM.getExchange());
}
/**
* 创建队列 true表示是否持久
*
* @return
*/
@Bean
public Queue EventProcessingDirectQueue() {
return new Queue(QueueConstants.QueueEventProcessingEnum.QUEUE_EVENT_PROCESSING_ENUM.getQueue(), true);
}
/**
* 将队列和交换机绑定,并设置用于匹配路由键
*
* @return
*/
@Bean
public Binding EventProcessingDirect() {
return BindingBuilder.bind(EventProcessingDirectQueue()).to(EventProcessingDirectExchange()).with(QueueConstants.QueueEventProcessingEnum.QUEUE_EVENT_PROCESSING_ENUM.getRouteKey());
}
}
\ No newline at end of file
src/main/java/com/cx/cn/cxquartz/config/RedisConfig.java
View file @ cb5db7cd
......@@ -12,7 +12,7 @@ import org.springframework.data.redis.core.*;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
// @Configuration
@Configuration
@EnableCaching //开启注解
public class RedisConfig extends CachingConfigurerSupport {
......
src/main/java/com/cx/cn/cxquartz/rabbitmq/QueueConstants.java
View file @ cb5db7cd
......@@ -279,8 +279,70 @@ public class QueueConstants {
this.routeKey = routeKey;
}
}
public enum QueueEventProcessingEnum {
QUEUE_EVENT_PROCESSING_ENUM(QueueConstants.QueueEventProcessingConsumer.EXCHANGE, QueueConstants.QueueEventProcessingConsumer.QUEUE,
QueueConstants.QueueEventProcessingConsumer.ROUTEKEY);
/**
* 交换机名称
*/
private String exchange;
/**
* 队列名称
*/
private String queue;
/**
* 路由键
*/
private String routeKey;
QueueEventProcessingEnum(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;
}
}
/**
* 获得rtsp 或者hls的队列
*/
public interface QueueEventProcessingConsumer{
/**
* 交换机名称
*/
String EXCHANGE = "RabbitMQ.DirectExchange.EventProcessingConsumer";
/**
* 队列名称
*/
String QUEUE = "RabbitMQ.DirectQueue.EventProcessingConsumer";
/**
* 路由键
*/
String ROUTEKEY = "RabbitMQ.RouteKey.EventProcessingConsumer";
}
......
src/main/java/com/cx/cn/cxquartz/rabbitmq/comsumer/TaskConsumConsumer.java → src/main/java/com/cx/cn/cxquartz/rabbitmq/comsumer/EventProcessingConsumer.java
View file @ cb5db7cd
......@@ -4,21 +4,21 @@ import com.cx.cn.cxquartz.helper.MessageHelper;
import com.cx.cn.cxquartz.service.quartz.TraffAlarmRecordService;
import com.cx.cn.cxquartz.service.quartz.impl.ResultService;
import com.cx.cn.cxquartz.util.JsonUtil;
import com.cx.cn.cxquartz.bean.QuartzTaskInformations;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.util.Map;
/**
* 消息处理并推送第三方
*/
//@Component
public class TaskConsumConsumer implements BaseConsumer {
private static final Logger logger = LoggerFactory.getLogger(TaskConsumConsumer.class);
@Component
public class EventProcessingConsumer implements BaseConsumer {
private static final Logger logger = LoggerFactory.getLogger(EventProcessingConsumer.class);
@Autowired
TraffAlarmRecordService traffAlarmRecordService;
......@@ -35,13 +35,14 @@ public class TaskConsumConsumer implements BaseConsumer {
public void consume(Message message, Channel channel) throws IOException {
logger.info("TaskConsumConsumer 收到消息: {}", message.toString());
Map result = MessageHelper.msgToObj(message, Map.class);
if (null != result.get("task")) {
QuartzTaskInformations taskinfo = JsonUtil.strToObj( result.get("task").toString(),QuartzTaskInformations.class);
if (null != result.get("result")) {
Map objresult = JsonUtil.strToObj(result.get("result").toString(), Map.class);
resultService.processResult(result);
//if (null !=result) {
// QuartzTaskInformations taskinfo = JsonUtil.strToObj( result.get("task").toString(),QuartzTaskInformations.class);
// if (null != result.get("result")) {
// Map objresult = JsonUtil.strToObj(result.get("result").toString(), Map.class);
//处理消息
resultService.processResult(taskinfo, objresult);
}
}
// resultService.processResult(taskinfo, objresult);
// }
//}
}
}
src/main/java/com/cx/cn/cxquartz/rabbitmq/comsumer/SnapShotConsumer.java
View file @ cb5db7cd
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.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.JsonUtil;
import com.cx.cn.cxquartz.vo.Sbtdspsr;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
......@@ -14,20 +10,18 @@ import org.springframework.amqp.core.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.util.Map;
@Component
public class SnapShotConsumer implements BaseConsumer{
private static final Logger logger = LoggerFactory.getLogger(TaskConsumConsumer.class);
private static final Logger logger = LoggerFactory.getLogger(SnapShotConsumer.class);
@Autowired
VideoRTSPorURLService videoRTSPorURLService;
@Override
public void consume(Message message, Channel channel) throws IOException {
public void consume(Message message, Channel channel) {
logger.info("SnapShotConsumer 收到消息: {}", message.toString());
Sbtdspsr result = MessageHelper.msgToObj(message, Sbtdspsr.class);
if (result.getTdlx().equals("1")) {
if (result.getTdlx()==1) {
//调用rtsp 的服务
String token=videoRTSPorURLService.getRTSPAccessToekenByDeviceCode(result.getSbbh());
videoRTSPorURLService.getRTSPByDeviceCode(token,result.getSbbh());
......
src/main/java/com/cx/cn/cxquartz/rabbitmq/comsumer/listener/OrderCancelReceiver.java → src/main/java/com/cx/cn/cxquartz/rabbitmq/comsumer/listener/EventProcessingReceiver.java
View file @ cb5db7cd
......@@ -3,7 +3,7 @@ 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.TaskConsumConsumer;
import com.cx.cn.cxquartz.rabbitmq.comsumer.EventProcessingConsumer;
import com.cx.cn.cxquartz.service.quartz.TraffPictureService;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
......@@ -16,20 +16,20 @@ import org.springframework.stereotype.Component;
/**
* 任务消息监听接受器
*/
//@Component
public class OrderCancelReceiver {
private static final Logger logger = LoggerFactory.getLogger(OrderCancelReceiver.class);
@Component
public class EventProcessingReceiver {
private static final Logger logger = LoggerFactory.getLogger(EventProcessingReceiver.class);
@Autowired
private TaskConsumConsumer taskConsumConsumer;
private EventProcessingConsumer eventProcessingConsumer;
@Autowired
private TraffPictureService traffPictureService;
@RabbitListener(queues = QueueConstants.QueueTaskConsumer.QUEUE,containerFactory="rabbitListenerContainerFactory")
@RabbitListener(queues = QueueConstants.QueueEventProcessingConsumer.QUEUE,containerFactory="rabbitListenerContainerFactory")
public void process(Message message, Channel channel) {
try {
logger.info("consumer->OrderCancelReceiver消费者收到消息 : " + message.toString());
BaseConsumerProxy baseConsumerProxy = new BaseConsumerProxy(taskConsumConsumer, traffPictureService);
BaseConsumerProxy baseConsumerProxy = new BaseConsumerProxy(eventProcessingConsumer, traffPictureService);
BaseConsumer proxy = (BaseConsumer) baseConsumerProxy.getProxy();
if (null != proxy) {
proxy.consume(message, channel);
......
src/main/java/com/cx/cn/cxquartz/rabbitmq/comsumer/listener/RTSPorHLSReceiver.java
View file @ cb5db7cd
......@@ -4,9 +4,6 @@ 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.SnapShotConsumer;
import com.cx.cn.cxquartz.rabbitmq.comsumer.TaskConsumConsumer;
import com.cx.cn.cxquartz.service.quartz.TraffAlarmRecordService;
import com.cx.cn.cxquartz.service.quartz.TraffPictureService;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
......
src/main/java/com/cx/cn/cxquartz/rabbitmq/comsumer/listener/SendtoDXReceiver.java
View file @ cb5db7cd
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.SendToDXConsumer;
import com.cx.cn.cxquartz.rabbitmq.comsumer.TaskConsumConsumer;
import com.cx.cn.cxquartz.service.quartz.impl.ResultService;
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;
......
src/main/java/com/cx/cn/cxquartz/redis/OrderConsumer.java
View file @ cb5db7cd
......@@ -5,7 +5,6 @@ import com.cx.cn.cxquartz.helper.MessageHelper;
import com.cx.cn.cxquartz.rabbitmq.QueueConstants;
import com.cx.cn.cxquartz.service.quartz.SbtdspsrService;
import com.cx.cn.cxquartz.service.quartz.impl.VideoRTSPorURLService;
import com.cx.cn.cxquartz.util.HttpClientUtil;
import com.cx.cn.cxquartz.util.JsonUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
......@@ -66,11 +65,11 @@ public class OrderConsumer implements Consumer {
if(null!=devicecode&&devicecode.startsWith("33") && devicecode.length()==18){
//调用抽帧服务
String token= orderConsumer.videoRTSPorURLService.getRTSPAccessToekenByDeviceCode(devicecode);
String rtsp=orderConsumer.videoRTSPorURLService.getRTSPByDeviceCode(token,devicecode);
orderConsumer.videoRTSPorURLService.getRTSPByDeviceCode(token,devicecode);
}
else{
//取表里最新的rtsp 或者hls 的值
String rtsp=orderConsumer.sbtdspsrService.getRtspOrHLSByDeviceCode(devicecode);
orderConsumer.sbtdspsrService.getRtspOrHLSByDeviceCode(devicecode);
}
//将rtsp 作为参数调用抽帧服务
String result="{\n" +
......
src/main/java/com/cx/cn/cxquartz/service/quartz/impl/EventWriteService.java
View file @ cb5db7cd
......@@ -5,23 +5,16 @@ import com.cx.cn.cxquartz.service.quartz.FtpService;
import com.cx.cn.cxquartz.service.quartz.TraffPictureService;
import com.cx.cn.cxquartz.util.*;
import com.cx.cn.cxquartz.vo.*;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.quartz.JobKey;
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.core.io.ByteArrayResource;
import org.springframework.core.io.FileSystemResource;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Service;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
import org.springframework.web.client.RestTemplate;
import sun.misc.BASE64Encoder;
import java.io.ByteArrayOutputStream;
import java.io.File;
......@@ -217,18 +210,6 @@ public class EventWriteService {
return restTemplate.postForObject(voiceurl, requestEntity, VoiceResultObj.class);
}
public String sendToService(JobKey jobKey, String jsondata) {
try {
HttpHeaders headers = new HttpHeaders();
headers.setContentType(MediaType.APPLICATION_JSON_UTF8);
HttpEntity<String> requestEntity = new HttpEntity<>(jsondata, headers);
return CompletableFuture.supplyAsync(() -> restTemplate.postForObject(jobKey.getName().contains("cz_") ? czurl : fxurl, requestEntity, String.class)).get(timeout, TimeUnit.SECONDS);
} catch (Exception ex) {
log.info("getRtspJob->executeInternal error:{}", ex.toString());
return null;
}
}
public TraffpictureParam getResult(TraffpictureParam traffpictureParamresult,int tarnum, Long[] roiarray, String imgurl,
List<Map> objectList,
......@@ -248,6 +229,23 @@ public class EventWriteService {
return null;
}
}
//人群总数和戴口罩统计
else if("60".equals(recordtype)){
int larmnum = getManNumber(objectList);
obj.put("alarmNum", tarnum);
if (tarnum < larmnum) {
obj.put("objectCount", larmnum);
traffpictureParamresult.setTargetnum(larmnum);
//判断戴口罩多少人
} else {
return null;
}
}
//获得所有对象的坐标
for (Map traffpictureParam : objectList) {
//根据imageid 获得 base64图片
......
src/main/java/com/cx/cn/cxquartz/service/quartz/impl/ResultService.java
View file @ cb5db7cd
......@@ -60,35 +60,39 @@ public class ResultService {
RabbitTemplate rabbitTemplate;
public static final Logger logger = LoggerFactory.getLogger(ResultService.class);
public void processResult(QuartzTaskInformations taskinfo, Map objectList) {
String devicecode=taskinfo.getExecuteparamter();
String recordtype = taskinfo.getRecordtype();
public void processResult( Map result) {
Map taskinfo= (Map) result.get("param");
String devicecode=(String) taskinfo.get("devicecode");
String recordtype =(String) taskinfo.get("recordtype");
String threshold=(String) taskinfo.get("threshold");
String timestamp=taskinfo.get("timestamp").toString().substring(0,taskinfo.get("timestamp").toString().indexOf("."));
JobTjParam jobTjParam = new JobTjParam();
jobTjParam.setDeviceId(devicecode);
jobTjParam.setDetectType(recordtype);
String imageurl = objectList.get("url").toString();
String imageurl = taskinfo.get("url").toString();
TraffpictureParam traffpictureParamresult=new TraffpictureParam();
try {
Map maprecogdata = JsonUtil.strToObj(objectList.get("recogdata").toString(), Map.class);
// Map maprecogdata = JsonUtil.strToObj(objectList.get("recogdata").toString(), Map.class);
List<Map> points = new ArrayList<>();
//分析结果数据
if (null != maprecogdata) {
List<Map> objectresult = (List<Map>) maprecogdata.get("ObjectList");
if (null != result) {
List<Map> objectresult = (List<Map>) result.get("ObjectList");
if (objectresult.size() < 1) {
logger.info(" objectresult is empty");
} else {
Long[] roiarray = new Long[4];
roiarray[0] = new Long(taskinfo.getObjectx());
roiarray[1] = new Long(taskinfo.getObjecty());
roiarray[2] = new Long(taskinfo.getObjectw());
roiarray[3] = new Long(taskinfo.getObjecth());
roiarray[0] = new Long(taskinfo.get("x").toString());
roiarray[1] = new Long(taskinfo.get("y").toString());
roiarray[2] = new Long(taskinfo.get("w").toString());
roiarray[3] = new Long(taskinfo.get("h").toString());
//图片划线并上传
String basepath = DateUtils.formatCurrDayYM() + File.separator + DateUtils.formatCurrDayDD() + File.separator + devicecode;
String filename = devicecode + "_" + DateUtils.parseDateToStrNoSign(objectList.get("timestamp").toString()) + "_result.jpg";
String filename = devicecode + "_" + DateUtils.parseDateToStrNoSign(timestamp) + "_result.jpg";
String filenameurl = File.separator + outpath + File.separator + basepath + File.separator + filename;
jobTjParam.setImageUrl(weburl + filenameurl);
traffpictureParamresult.setImagedata(filenameurl); //获得点位
traffpictureParamresult = eventWriteService.getResult(traffpictureParamresult, Integer.parseInt(taskinfo.getMetatype())
traffpictureParamresult = eventWriteService.getResult(traffpictureParamresult, Integer.parseInt(threshold)
, roiarray, imageurl, objectresult, jobTjParam, points);
if (null == traffpictureParamresult) {
logger.info("人群密度未超或目标未出现");
......@@ -97,22 +101,22 @@ public class ResultService {
eventWriteService.uploadPicture(traffpictureParamresult, imageurl, points, basepath, filename);
//新增
eventWriteService.setTraffpictureParam(recordtype, devicecode,
objectList.get("timestamp").toString(),
timestamp,
traffpictureParamresult);
Map sendtodxmap = new HashMap();
sendtodxmap.put("id", traffpictureParamresult.getId());
sendtodxmap.put("traff", JsonUtil.objToStr(jobTjParam));
sendtodxmap.put("callback", taskinfo.getUrl().equals("") ? callbackurl : taskinfo.getUrl());
sendtodxmap.put("callback", taskinfo.get("url").equals("") ? callbackurl : taskinfo.get("url").toString());
CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
rabbitTemplate.convertAndSend(QueueConstants.QueueSendToDXEnum.QUEUE_SEND_TO_DX_ENUM.getExchange(),
QueueConstants.QueueSendToDXEnum.QUEUE_SEND_TO_DX_ENUM.getRouteKey(),
MessageHelper.objToMsg(sendtodxmap),
correlationData);
rabbitTemplate.convertAndSend(QueueConstants.QueueSendToVoiceEnum.QUEUE_SEND_TO_VOICE_ENUM.getExchange(),
QueueConstants.QueueSendToVoiceEnum.QUEUE_SEND_TO_VOICE_ENUM.getRouteKey(),
MessageHelper.objToMsg(sendtodxmap),
correlationData);
// rabbitTemplate.convertAndSend(QueueConstants.QueueSendToDXEnum.QUEUE_SEND_TO_DX_ENUM.getExchange(),
// QueueConstants.QueueSendToDXEnum.QUEUE_SEND_TO_DX_ENUM.getRouteKey(),
// MessageHelper.objToMsg(sendtodxmap),
// correlationData);
//
// rabbitTemplate.convertAndSend(QueueConstants.QueueSendToVoiceEnum.QUEUE_SEND_TO_VOICE_ENUM.getExchange(),
// QueueConstants.QueueSendToVoiceEnum.QUEUE_SEND_TO_VOICE_ENUM.getRouteKey(),
// MessageHelper.objToMsg(sendtodxmap),
// correlationData);
// //回调第三方接口
//// logger.info("send to dianxin data:{}",JSONObject.toJSONString(jobTjParam));
// eventWriteService.sendEventByCallUrl(traffpictureParamresult, jobTjParam, taskinfo.getUrl().equals("") ? callbackurl : taskinfo.getUrl());
......
src/main/java/com/cx/cn/cxquartz/service/quartz/impl/VideoRTSPorURLService.java
View file @ cb5db7cd
......@@ -66,7 +66,8 @@ public class VideoRTSPorURLService {
{
//获得token 成功,更新表数据
String token= String.valueOf(result.get("obj"));
String serverHlsTokenNextTime =DateUtils.addMin(String.valueOf(result.get("serverTime")),5);
Long serverTime= Long.parseLong(result.get("serverTime").toString());
String serverHlsTokenNextTime =DateUtils.addMin(new Date(serverTime),5);
//根据设备编号更新token 及token 过期数据
Sbtdspsr sbtdspsr=new Sbtdspsr();
sbtdspsr.setSbbh(deviceCode);
......@@ -88,7 +89,7 @@ public class VideoRTSPorURLService {
public String getHLSByDeviceCode(String token,String devicecode){
//token 保活
HttpHeaders headers = new HttpHeaders();
headers.setContentType(MediaType.APPLICATION_JSON_UTF8);
headers.setContentType(MediaType.APPLICATION_JSON);
headers.add("token", token);
HttpEntity<String> httpEntity = new HttpEntity(headers);
ResponseEntity<String> exchange = restTemplate.exchange(hlsurl+"?cameraIndexCode="+devicecode, HttpMethod.GET ,httpEntity,String.class);
......@@ -105,7 +106,8 @@ public class VideoRTSPorURLService {
// "serverTime": "2021-07-21 12:40:06",
// "resultMsg": "成功"
//根据设备编号更新hls和下次执行时间 数据，hls 有效期是5分钟
String serverHlsNextTime =DateUtils.addMin(String.valueOf(result.get("serverTime")),5);
Long expire= Long.parseLong(result.get("serverTime").toString());
String serverHlsNextTime =DateUtils.addMin(new Date(expire),5);
Sbtdspsr sbtdspsr=new Sbtdspsr();
sbtdspsr.setSbbh(devicecode);
sbtdspsr.setUrlnexttime(serverHlsNextTime);
......@@ -130,7 +132,7 @@ public class VideoRTSPorURLService {
public String getRTSPAccessToekenByDeviceCode(String deviceCode){
//token 保活
HttpHeaders headers = new HttpHeaders();
headers.setContentType(MediaType.APPLICATION_JSON_UTF8);
headers.setContentType(MediaType.APPLICATION_JSON);
Map result=restTemplate.getForObject(rtspurltoken+"?appid="+appid+"&appsecret="+appsecret,Map.class);
if(null!=result.get("code") && "0".equals(result.get("code").toString())){
// {
......@@ -143,7 +145,8 @@ public class VideoRTSPorURLService {
//获得token 成功,更新表数据
Map obj=(Map) result.get("obj");
String token=String.valueOf(obj.get("accessToken"));
String serverRTSPTokenNextTime=DateUtils.formatDate(new Date(String.valueOf(obj.get("expire"))));
Long expire= Long.parseLong(obj.get("expire").toString());
String serverRTSPTokenNextTime=DateUtils.formatDate(new Date(expire));
//根据设备编号更新token 及token 过期数据
Sbtdspsr sbtdspsr=new Sbtdspsr();
sbtdspsr.setSbbh(deviceCode);
......@@ -156,20 +159,23 @@ public class VideoRTSPorURLService {
return null;
}
public String getRTSPByDeviceCode(String token,String deviceCode) {
public void getRTSPByDeviceCode(String token,String deviceCode) {
String timestamp = String.valueOf(new Date().getTime());
HttpHeaders headers = new HttpHeaders();
headers.setContentType(MediaType.APPLICATION_JSON_UTF8);
headers.setContentType(MediaType.APPLICATION_JSON);
headers.add("accessToken", token);
headers.add("timestamp", timestamp);
headers.add("sign", DigestUtils.md5DigestAsHex((params + deviceCode + "key" + timestamp).getBytes()));
HttpEntity<String> httpEntity = new HttpEntity(headers);
ResponseEntity<String> exchange = restTemplate.exchange(rtspurl + "?deviceCode=" + deviceCode, HttpMethod.GET, httpEntity, String.class);
String body = exchange.getBody();
Sbtdspsr sbtdspsr = new Sbtdspsr();
try {
Map result = new ObjectMapper().readValue(body, Map.class);
if (null != result.get("errorCode") && "0".equals(result.get("errorCode").toString())) {
sbtdspsr.setSbbh(deviceCode);
//获得token 成功,更新表数据
Map data = (Map) result.get("data");
// {
......@@ -185,9 +191,8 @@ public class VideoRTSPorURLService {
//省内设备 = 地址只能用一次 ,33开头的18位
// //集团设备 = 地址可以用20分钟，15位
Sbtdspsr sbtdspsr = new Sbtdspsr();
sbtdspsr.setSqurllj(rtsp);
sbtdspsr.setSbbh(deviceCode);
}
if (deviceCode.length() == 15) {
String serverHlsNextTime = DateUtils.addMin(new Date(), 20);
sbtdspsr.setUrlnexttime(serverHlsNextTime);
......@@ -196,16 +201,14 @@ public class VideoRTSPorURLService {
sbtdspsr.setUrlnexttime(null);
}
//更新表里的数据
if(null!=sbtdspsr.getSbbh() && !"".equals(sbtdspsr.getSbbh()))
sbtdspsrService.updateRTSPorHLSParam(sbtdspsr);
return rtsp;
//根据token获得rtsp 地址
}
} catch (Exception ex) {
logger.error("getRTSPByDeviceCode" + ex.toString());
}
return null;
}
/***
* 根据rtsp 或者hls 去抽帧
......
src/main/java/com/cx/cn/cxquartz/util/DateUtils.java
View file @ cb5db7cd
......@@ -41,6 +41,10 @@ public class DateUtils {
public static Date parseDate(String date){
return DateTime.parse(date,YMD_HMS).toDate();
}
public static Date parseMilisecondsToDate(Long time){
return new Date(time);
}
public static Date parseDateNoSign(String date){
return DateTime.parse(date,YMDHMS).toDate();
......
src/main/java/com/cx/cn/cxquartz/util/FTPUtil.java deleted 100644 → 0
View file @ 2fb71bd3
This diff is collapsed.
src/main/java/com/cx/cn/cxquartz/util/FileTransferManager.java deleted 100644 → 0
View file @ 2fb71bd3
package com.cx.cn.cxquartz.util;
import com.cx.cn.cxquartz.controller.ExtController;
import com.cx.cn.cxquartz.vo.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import sun.misc.BASE64Encoder;
import java.io.IOException;
import java.io.InputStream;
import java.net.HttpURLConnection;
import java.net.URL;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.TimeUnit;
public class FileTransferManager {
private static CompletionService<String> completionService = new ExecutorCompletionService<String>(ThreadPoolUtil.getPool());
private static CompletionService<String> threadService = new ExecutorCompletionService<String>(ThreadPoolUtil.getPool());
private static final Logger log = LoggerFactory.getLogger(ExtController.class);
/**
* @param ftp
* @param basePath
* @return
* @throws Exception
*/
public static void transferFile(final TraffAlarmRecord traffAlarmRecord, final Ftp ftp, final String basePath) throws Exception {
final Long recordid = traffAlarmRecord.getRecordid();
String fileName = DateUtils.formatCurrDayNoSign() + "_" + recordid + "_img1.jpg";
String ftputl = FTPUtil.getFtpUrl(ftp) + basePath + "/" + fileName;
traffAlarmRecord.setImg1path(ftputl);
completionService.submit(() -> {
try {
HttpURLConnection connection = (HttpURLConnection) new URL(traffAlarmRecord.getImg1urlfrom()).openConnection();
connection.setReadTimeout(2000); //延迟连接
connection.setConnectTimeout(3000);
connection.setRequestMethod("GET");
if (connection.getResponseCode() == HttpURLConnection.HTTP_OK || connection.getResponseCode() == HttpURLConnection.HTTP_NOT_MODIFIED) {
InputStream inputStream = connection.getInputStream();
boolean r = FTPUtil.uploadFile(ftp, basePath, fileName, inputStream);
if (r) {
log.info(" ftpupload success:{}", ftputl);
}
} else {
log.error("connection code: " + connection.getResponseCode());
}
} catch (IOException e) {
log.error("ftputl connection error:{}", e.toString());
return "0";
}
return "1";
}
);
}
public static void fetchUrlsFromRecord(TraffAlarmRecord record, Map<String, Object> transferRecord) {
if (record == null || record.getRecordid() == null) {
return;
}
transferRecord.put("recordid", record.getRecordid());
if (record.getImg1path() == null && record.getImg1urlfrom() != null) {
transferRecord.put("img1path", record.getImg1urlfrom());
}
}
public static TraffAlarmRecord traffAlarmRecordUrlUpdate(List<TransferResult> results) throws Exception {
TraffAlarmRecord record = new TraffAlarmRecord();
for (TransferResult result : results) {
record.setRecordid(result.getRecordid());
if (result.getResult()) {
switch (result.getPathvalue()) {
case "img1path":
record.setImg1path(result.getImgpath());
break;
case "videopath":
record.setVideopath(result.getImgpath());
break;
default:
break;
}
}
}
return record;
}
// public static GoalStructureParam getGoalStructureParam(Long[] roiarray, int model, BASE64Encoder base64Encoder, TraffAlarmRecord transferRecord) {
// //获得图片进行分析
// GoalStructureParam param = new GoalStructureParam();
// param.setOutput(new Output(1));
// param.setModel(model);
// param.setApiout("1");////打开1400标准输出，默认可以不填
// List<ImageList> list = new ArrayList<>();
// getImageList("1", roiarray, base64Encoder, list, transferRecord.getImg1path());
// param.setImageList(list);
// return param;
//
// }
public static GoalStructureParam getGoalStructureParam(Long[] roiarray, int model,int format, TraffAlarmRecord transferRecord) {
//获得图片进行分析
GoalStructureParam param = new GoalStructureParam();
Map outmap=new HashMap();
outmap.put("SubClass",1);
param.setOutput(outmap);
param.setModel(model);
param.setApiout("1");////打开1400标准输出，默认可以不填
List<Map> list = new ArrayList<>();
Map imgmap=new HashMap();
imgmap.put("ImageID","1");
imgmap.put("Roi",roiarray);
imgmap.put("apiout","1");
imgmap.put("Format",format);
imgmap.put("Data",transferRecord.getImg1path());
imgmap.put("RoiPadding",0);
imgmap.put("CropObjectImage",1);
imgmap.put("CropObjectImagePadding",20);
imgmap.put("CropObjectImageQuality",100);
list.add(imgmap);
// new ImageList("1", roiarray, format,10, 1, 20, 100, transferRecord.getImg1urlfrom())
// getImageList("1", roiarray,format, list, transferRecord.getImg1urlfrom());
param.setImageList(list);
return param;
}
// private static void getImageList(String id, Long[] roiarray, BASE64Encoder base64Encoder, List<ImageList> list, String img1path) {
// if (null != img1path && !"".equals(img1path)) {
//
// byte[] Img = FTPUtil.getFtpPicBytes(img1path);
// if (Img != null) {
// list.add(new ImageList(id, roiarray, 10, 1, 20, 100, base64Encoder.encode(Img).replaceAll("\r|\n", "")));
//
// }
// }
// }
public static String urlTransToFtp(String url, final Ftp ftp, final String basePath) {
threadService.submit(() -> {
try {
HttpURLConnection connection = (HttpURLConnection) new URL(url).openConnection();
connection.setReadTimeout(2000); //延迟连接
connection.setConnectTimeout(3000);
connection.setRequestMethod("GET");
if (connection.getResponseCode() == HttpURLConnection.HTTP_OK || connection.getResponseCode() == HttpURLConnection.HTTP_NOT_MODIFIED) {
InputStream inputStream = connection.getInputStream();
String pname = DateUtils.formatCurrDayNoSign() + ".jpg";
String ftputl = FTPUtil.getFtpUrl(ftp) + basePath + pname;
boolean r = FTPUtil.uploadFile(ftp, basePath, pname, inputStream);
if (r) {
return ftputl;
}
} else {
log.error("connection code: " + connection.getResponseCode());
}
} catch (IOException e) {
System.out.println(e.toString());
log.error(e.toString());
}
return url;
});
try {
return threadService.poll(10, TimeUnit.SECONDS).get();
} catch (Exception ex) {
return url;
}
}
}
src/main/java/com/cx/cn/cxquartz/util/HttpClientUtil.java deleted 100644 → 0
View file @ 2fb71bd3
package com.cx.cn.cxquartz.util;
import org.apache.commons.httpclient.*;
import org.apache.commons.httpclient.cookie.CookiePolicy;
import org.apache.commons.httpclient.methods.ByteArrayRequestEntity;
import org.apache.commons.httpclient.methods.GetMethod;
import org.apache.commons.httpclient.methods.PostMethod;
import org.apache.commons.httpclient.params.HttpConnectionManagerParams;
import org.apache.commons.httpclient.params.HttpMethodParams;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class HttpClientUtil {
private static Logger log = LoggerFactory.getLogger(HttpClientUtil.class);
private static final String ENCODING = "UTF-8";
private static final int CONNECTION_TIME_OUT = 3000;
private static final int SO_TIME_OUT = 100000;
private static final boolean STALE_CHECK_ENABLED = true;
private static final boolean TCP_NO_DELAY = true;
private static final int DEFAULT_MAX_CONNECTIONS_PER_HOST = 100;
private static final int MAX_TOTAL_CONNECTIONS = 1000;
private static final HttpConnectionManager connectionManager;
public static final HttpClient client;
static {
HttpConnectionManagerParams params = loadHttpConfFromFile();
connectionManager = new MultiThreadedHttpConnectionManager();
connectionManager.setParams(params);
client = new HttpClient(connectionManager);
}
private static HttpConnectionManagerParams loadHttpConfFromFile() {
HttpConnectionManagerParams params = new HttpConnectionManagerParams();
params.setConnectionTimeout(CONNECTION_TIME_OUT);
params.setStaleCheckingEnabled(STALE_CHECK_ENABLED);
params.setTcpNoDelay(TCP_NO_DELAY);
params.setSoTimeout(SO_TIME_OUT);
params.setDefaultMaxConnectionsPerHost(DEFAULT_MAX_CONNECTIONS_PER_HOST);
params.setMaxTotalConnections(MAX_TOTAL_CONNECTIONS);
params.setParameter(HttpMethodParams.RETRY_HANDLER, new DefaultHttpMethodRetryHandler(0, false));
return params;
}
/**
* get请求
*
* @param url
* @return
*/
public static String doGet(String url) {
String result = null;
try {
GetMethod getMethod = new GetMethod(url);
client.executeMethod(getMethod);
result = getMethod.getResponseBodyAsString();
} catch (Exception e) {
log.error("httpclient get request url=" + url + ",exception=" + e);
}
return result;
}
public static String doPost(String url, String contentType, String content) throws Exception {
PostMethod method = new PostMethod(url);
method.addRequestHeader("Connection", "Keep-Alive");
method.getParams().setCookiePolicy(CookiePolicy.IGNORE_COOKIES);
method.getParams().setParameter(HttpMethodParams.RETRY_HANDLER, new DefaultHttpMethodRetryHandler(0, false));
try {
method.setRequestEntity(new ByteArrayRequestEntity(content.getBytes(ENCODING)));
method.addRequestHeader("Content-Type", contentType);
int statusCode = client.executeMethod(method);
if (statusCode != HttpStatus.SC_OK) {
return null;
}
byte[] ret = method.getResponseBody();
if (ret == null)
return null;
return new String(ret, ENCODING);
} finally {
method.releaseConnection();
}
}
}
src/main/java/com/cx/cn/cxquartz/util/RestUtil.java
View file @ cb5db7cd
......@@ -2,7 +2,6 @@ package com.cx.cn.cxquartz.util;
import com.cx.cn.cxquartz.service.quartz.SbtdspsrService;
import com.cx.cn.cxquartz.vo.TraffAlarmRecord;
import net.sf.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
......
src/main/java/com/cx/cn/cxquartz/vo/Sbtdspsr.java
View file @ cb5db7cd
......@@ -8,7 +8,7 @@ public class Sbtdspsr {
private Integer tdbh;
private Long tdlx;
private int tdlx;
private String wbbh;
......@@ -206,11 +206,11 @@ public class Sbtdspsr {
this.tdbh = tdbh;
}
public Long getTdlx() {
public int getTdlx() {
return tdlx;
}
public void setTdlx(Long tdlx) {
public void setTdlx(int tdlx) {
this.tdlx = tdlx;
}
......
src/main/resources/application-local.yml
View file @ cb5db7cd
......@@ -20,11 +20,11 @@ spring:
poolPreparedStatements: true
maxOpenPreparedStatements: 20
rabbitmq:
host: 127.0.0.1
host: 192.168.168.110
port: 5672
username: guest
password: guest
virtual-host: /
username: admin
password: admin
virtual-host: my_vhost
#消息发送到交换机确认机制，是否确认回调
publisher-confirms: true
##消息发送到交换机确认机制，是否返回回调
......@@ -46,10 +46,16 @@ spring:
initial-interval: 3000
#最大重试次数
max-attempts: 3
redis:
database: 0
host: 127.0.0.1
redis:
database: 0
host: 192.168.168.110
port: 6379
pool:
max-active: 100 #连接池最大连接数（负值表示没有限制）
max-wait: 3000 #连接池最大阻塞等待时间（负值表示没有限制）
max-idle: 200 #连接池最大空闭连接数
min-idle: 50 #连接汉最小空闲连接数
timeout: 600 #连接超时时间（毫秒）
#logging:
# level:
......
src/main/resources/mapper/SbtdspsrMapper.xml
View file @ cb5db7cd
......@@ -41,5 +41,25 @@
<select id="getRtspOrHLSByDeviceCode" parameterType="java.lang.String" resultType="java.lang.String">
select squrllj from sbtdspsr where sbbh=#{deviceCode}
</select>
<update id="updateRTSPorHLSParam">
update sbtdspsr
<trim prefix="set" suffixOverrides=",">
<if test="squrllj != null and squrllj !=''">
squrllj =#{squrllj},
</if>
<if test="token != null and token !=''">
token =#{token},
</if>
<if test="tokennexttime != null and tokennexttime !=''">
tokennexttime =#{tokennexttime},
</if>
<if test="urlnexttime != null and urlnexttime !=''">
urlnexttime =#{urlnexttime},
</if>
</trim>
where sbbh=#{sbbh}
</update>
</mapper>
src/test/java/com/quartz/cn/com/cx/cn/cxquartz/CXQuartzApplicationTests.java
View file @ cb5db7cd
package com.quartz.cn.com.cx.cn.cxquartz;
import com.cx.cn.cxquartz.service.quartz.impl.SbtdspsrImpl;
import com.cx.cn.cxquartz.vo.Sbtdspsr;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.boot.autoconfigure.data.redis.RedisProperties;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
@SpringBootTest
public class CXQuartzApplicationTests {
//
// @Test
......@@ -30,8 +21,7 @@ import org.springframework.test.context.junit4.SpringRunner;
// }
// }
@Test
//
public void testfilepath(){
String ftpfilepath="ftp://jiuling:9ling.cn@172.16.24.29:21/gstraff/picturedemo/20210625.jpg";
String ftppath="jiuling:9ling.cn@172.16.24.29:21";
......
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