事件推送

src/main/java/com/hzjt/config/ScheduleTaskConfig.java

 package com.hzjt.config;
 
+import com.alibaba.fastjson.JSON;
 import com.hzjt.domain.ResultObj;
 import com.hzjt.domain.Traffalarmrecord;
+import com.hzjt.domain.VideoDeviceTraffic;
 import com.hzjt.handler.WebSocket;
 import com.hzjt.service.EventWriteService;
 import com.hzjt.service.TraffFlowService;
@@ -18,6 +20,7 @@ import org.springframework.stereotype.Component;
 import java.net.URI;
 import java.util.Calendar;
 import java.util.List;
+import java.util.Map;
 
 @Slf4j
 @Component
@@ -62,26 +65,34 @@ public class ScheduleTaskConfig {
     /***
      * 每2分钟执行一次查询事件推送给广达
      */
-    @Scheduled(cron = "0 */2 * * * ?")
+//    @Scheduled(cron = "0 */2 * * * ?")
     private void sendToguangdaEvent() {
         //查询所有未推送的数据
         List<Traffalarmrecord> resultlist = traffalarmrecordService.getTraffalarmrecordBystatus();
         for (Traffalarmrecord record : resultlist) {
             try {
                 ResultObj obj = eventWriteService.updateAndAutoSendEvent(record);
-                if (obj.getStatus() == 200) {
-                    log.error(record.getRecordid() + " send to guangda success");
-                } else {
-                    log.info(record.getRecordid() + " send to guangda fail"+obj.getMsg());
-                }
-            } catch (Exception ex) {
+                log.error( "2 min  send to guangda result"+obj.getMsg()+obj.getStatus());
+                } catch (Exception ex) {
                 log.error(record.getRecordid() + " send to guangda fail" + ex.toString());
             }
         }
 
     }
 
-
+    /***
+     * 每5分钟执行一次查询车流推送给广达
+     */
+    @Scheduled(cron = "0 */5 * * * ?")
+    private void sendToguangdaDevice() {
+        log.info("fiveflow  sendto guangda" );
+        //查询所有未推送的数据
+        List<VideoDeviceTraffic>  trafflist = traffFlowService.getfiveFlowByVideoid(null);
+        if(trafflist.size()>0) {
+            log.info("fiveflow" + JSON.toJSONString(trafflist));
+            traffFlowService.sendMessages(trafflist);
+        }
+    }
 
 
 //    @Scheduled(cron = "0 30/5 6-19 * * ?")//每隔5分钟统计当天目前为止的数据

src/main/java/com/hzjt/controller/RouterController.java

 package com.hzjt.controller;
 
 import com.alibaba.fastjson.JSONObject;
+import com.hzjt.domain.VideoDeviceTraffic;
 import com.hzjt.mapper.TraffFlowMapper;
 import com.hzjt.service.TraffFlowService;
 import lombok.extern.slf4j.Slf4j;
@@ -9,6 +10,7 @@ import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Controller;
 import org.springframework.web.bind.annotation.*;
 
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 
@@ -39,8 +41,7 @@ TraffFlowService traffFlowService;
     @GetMapping("/indexxx")
     public void indexxx() {
 
-        Map map = traffFlowService.getfiveFlowByVideoid("1");
-        System.out.println(map);
+        List<VideoDeviceTraffic> map = traffFlowService.getfiveFlowByVideoid("1");
     }
 
 }

src/main/java/com/hzjt/controller/TraffController.java

@@ -123,6 +123,12 @@ public class TraffController {
 //            log.info("video_id值异常 值为:" + videoId);
             return ResultObj.error(ResponseEnum.E_1002.getCode(), "video_id值异常 值为:" + videoId);
         }
+        /* 限流 */
+        if (!getRateLimiter(videoId).tryAcquire()) {
+            log.info("[事件推送]->设备" + videoId + "-推送已达到限流限制");
+            return ResultObj.error(ResponseEnum.E_9999.getCode(), "设备" + videoId + "请求过于频繁");
+        }
+
         String[] sbAndTd = videoId.split("_");
         String sbbh = sbAndTd[0];
         int tdbh = Integer.valueOf(sbAndTd[1]) + 1;
@@ -167,6 +173,7 @@ public class TraffController {
         traffAlarmRecord.setVideourlfrom(trffClientMessage.getVideo_record_url());
         traffAlarmRecord.setCreatetime(new Date());
         traffAlarmRecord.setObjlable(trffClientMessage.getObjLabel());
+        traffAlarmRecord.setPushstatus(9);//设置为未推送
 
 //        log.info("selectInworkRectifytimeTraffAlarm" + exists);
         if ((traffAlarmRecord.getRecordtype().equalsIgnoreCase("roadworks_out_of_line") || traffAlarmRecord.getRecordtype().equalsIgnoreCase("roadworks"))) {
@@ -262,17 +269,17 @@ public class TraffController {
                                         traffAlarmRecord.setPushstatus(9);//待推送及videopath
                                         traffAlarmRecordMapper.updateTraffAlarmRecordUrl(traffAlarmRecord);
                                         //推送给第三方
-//                                        if(!manualStatus.equalsIgnoreCase("1")) {
-//                                            ResultObj obj = eventWriteService.updateAndAutoSendEvent(traffAlarmRecord);
-////                                        log.info("send to guangda:HTTP_OK" + obj.toString());
-//                                        }
+                                        if(!manualStatus.equalsIgnoreCase("1")) {
+                                            ResultObj obj = eventWriteService.updateAndAutoSendEvent(traffAlarmRecord);
+//                                        log.info("send to guangda:HTTP_OK" + obj.toString());
+                                        }
                                     }
                                 } else {
                                         //只更新待推送
                                     traffAlarmRecord.setPushstatus(9);
                                     traffAlarmRecordMapper.updateTraffAlarmRecordUrl(traffAlarmRecord);
-                                    //ResultObj obj = eventWriteService.updateAndAutoSendEvent(traffAlarmRecord);
-//                                    log.info("send to guangda HttpURLConnection.HTTP_Fail--->response message:" + obj.toString());
+                                    ResultObj obj = eventWriteService.updateAndAutoSendEvent(traffAlarmRecord);
+                                    log.info("send to guangda HttpURLConnection.HTTP_Fail--->response message:" + obj.toString());
                                 }
 
 
@@ -294,8 +301,8 @@ public class TraffController {
                 if(!manualStatus.equalsIgnoreCase("1")) {
                     traffAlarmRecord.setPushstatus(9);
                     traffAlarmRecordMapper.updateTraffAlarmRecordUrl(traffAlarmRecord);
-//                    ResultObj obj = eventWriteService.updateAndAutoSendEvent(traffAlarmRecord);
-//                   log.info("send to guangda novideopath --->response message" + obj.toString());
+                   ResultObj obj = eventWriteService.updateAndAutoSendEvent(traffAlarmRecord);
+                   log.info("send to guangda novideopath --->response message" + obj.toString());
                 }
             }
 
@@ -383,6 +390,19 @@ public class TraffController {
         webSocket.AppointSending(rules.getVideo_id(), JsonUtil.beanToString(map));
         return ResultObj.ok();
     }
+    private RateLimiter getRateLimiter(String videoId) {
+        RateLimiter rateLimiter;
+        if (rateLimiterMap.containsKey(videoId)) {
+            rateLimiter = rateLimiterMap.get(videoId);
+        } else {
+            RateLimiter value = RateLimiter.create(rate);
+            rateLimiter = rateLimiterMap.putIfAbsent(videoId, value);
+            if (rateLimiter == null) {
+                rateLimiter = value;
+            }
+        }
+        return rateLimiter;
+    }
 
 
 
@@ -430,7 +450,7 @@ public class TraffController {
             List<AutoVehiclesdetail> vels = vehicles.getObjs();
             int upnum = 0, downnum = 0;
             Date date=new Date(Long.valueOf(vehicles.getTs()));
-            List<VideoDeviceTraffic> trafflist = new ArrayList<>();
+           // List<VideoDeviceTraffic> trafflist = new ArrayList<>();
 
             for (AutoVehiclesdetail detail : vels) {
                 Vehicle v = new Vehicle();
@@ -443,30 +463,30 @@ public class TraffController {
                 v.setVideo_id(vehicles.getVideo_id());
                 traffFlowService.saveTraffFlowDetail(v);
             }
-            Sbtdspsr  sbtdspsr =  sbtdspsrList.get(0);
-            Map  map = traffFlowService.getfiveFlowByVideoid(vehicles.getVideo_id());
-            //发送给广达
-            VideoDeviceTraffic traff = new VideoDeviceTraffic();
-            traff.setDeviceId(sbtdspsr.getWbbh());
-            traff.setDateTime(date);
-            traff.setSsjg(sbtdspsrList.get(0).getXzbh());
-            traff.setDeviceName(sbtdspsr.getTdmc());
-            traff.setSjly("zksy");
-            traff.setFlowRate(map.get("upfiveflow")==null?"0":String.valueOf(map.get("upfiveflow")));
-            traff.setFxmc("上行");
-            traff.setFx("1");
-            trafflist.add(traff);
-            VideoDeviceTraffic traffwodn = new VideoDeviceTraffic();
-            traffwodn.setDeviceId(sbtdspsr.getWbbh());
-            traffwodn.setDateTime(date);
-            traffwodn.setSsjg(sbtdspsrList.get(0).getXzbh());
-            traffwodn.setDeviceName(sbtdspsr.getTdmc());
-            traffwodn.setSjly("zksy");
-            traffwodn.setFlowRate(map.get("downfiveflow")==null?"0":String.valueOf(map.get("downfiveflow")));
-            traffwodn.setFxmc("下行");
-            traffwodn.setFx("2");
-            trafflist.add(traffwodn);
-            traffFlowService.sendMessages(trafflist);
+//            Sbtdspsr  sbtdspsr =  sbtdspsrList.get(0);
+//            Map  map = traffFlowService.getfiveFlowByVideoid(vehicles.getVideo_id());
+//            //发送给广达
+//            VideoDeviceTraffic traff = new VideoDeviceTraffic();
+//            traff.setDeviceId(sbtdspsr.getWbbh());
+//            traff.setDateTime(date);
+//            traff.setSsjg(sbtdspsrList.get(0).getXzbh());
+//            traff.setDeviceName(sbtdspsr.getTdmc());
+//            traff.setSjly("zksy");
+//            traff.setFlowRate(map.get("upfiveflow")==null?"0":String.valueOf(map.get("upfiveflow")));
+//            traff.setFxmc("上行");
+//            traff.setFx("1");
+//            trafflist.add(traff);
+//            VideoDeviceTraffic traffwodn = new VideoDeviceTraffic();
+//            traffwodn.setDeviceId(sbtdspsr.getWbbh());
+//            traffwodn.setDateTime(date);
+//            traffwodn.setSsjg(sbtdspsrList.get(0).getXzbh());
+//            traffwodn.setDeviceName(sbtdspsr.getTdmc());
+//            traffwodn.setSjly("zksy");
+//            traffwodn.setFlowRate(map.get("downfiveflow")==null?"0":String.valueOf(map.get("downfiveflow")));
+//            traffwodn.setFxmc("下行");
+//            traffwodn.setFx("2");
+//            trafflist.add(traffwodn);
+//            traffFlowService.sendMessages(trafflist);
             //查询近五分钟的车流量，当天车流量websocket  直接推送过去
             //根据连接的name ，群发根据videoid  查询的结果
             webSocket.GroupSendingByVideoid(vehicles.getVideo_id());

src/main/java/com/hzjt/handler/WebSocket.java

@@ -91,16 +91,18 @@ public class WebSocket {
      * @param message
      */
     public void GroupSending(String message) {
-        synchronized (session) {
+
             for (String name : webSocketSet.keySet()) {
                 try {
-                    if (null != webSocketSet.get(name) && null != webSocketSet.get(name).session && null != webSocketSet.get(name).session.getBasicRemote())
-                        webSocketSet.get(name).session.getBasicRemote().sendText(message);
-                } catch (Exception e) {
+
+                        if (null != webSocketSet.get(name) && null != webSocketSet.get(name).session && null != webSocketSet.get(name).session.getBasicRemote())
+                            synchronized (webSocketSet.get(name).session) {
+                            webSocketSet.get(name).session.getBasicRemote().sendText(message);
+                           }
+                    } catch (Exception e) {
                     e.printStackTrace();
                 }
             }
-        }
     }
 
     /**
@@ -141,7 +143,7 @@ public class WebSocket {
      */
     public void AppointSending(String name, String message) {
         if (null != webSocketSet.get(name) && null != webSocketSet.get(name).session && null != webSocketSet.get(name).session.getBasicRemote()) {
-            synchronized (session) {
+            synchronized (webSocketSet.get(name).session) {
                 try {
                     webSocketSet.get(name).session.getBasicRemote().sendText(message);
 

src/main/java/com/hzjt/mapper/TraffFlowMapper.java

@@ -18,7 +18,7 @@ public interface TraffFlowMapper {
     List<Map> selectFiveAndDayFlow(String videoid);
 
     List<Map> selectFiveAndTypeDayFlow(String videoid);
-    Map getfiveFlowByVideoid(String videoid);
+    List<VideoDeviceTraffic>  getfiveFlowByVideoid(String videoid);
 
     Integer  statisVehiclesByDay();
 

src/main/java/com/hzjt/redis/Receiver.java

@@ -151,7 +151,7 @@ public class Receiver {
 
                                             //推送给第三方
                                             ResultObj obj = eventWriteService.updateAndAutoSendEvent(traffAlarmRecord);
-//                                            log.info("调用接口推送给广达返回信息：" + obj.toString());
+                                          log.info("调用接口推送给广达返回信息：" + obj.toString());
 
                                         }
                                     }

src/main/java/com/hzjt/service/EventWriteService.java

@@ -154,15 +154,16 @@ public class EventWriteService {
             eventWriteParam.setEventProof5(Img != null ? base64Encoder.encode(Img) : null);
         }
 
-        if(traffalarmrecord.getVideopath() != null ) {
-            byte[] video = FTPUtil.getFtpPicBytes(traffalarmrecord.getVideopath());
-            eventWriteParam.setEventvideo1(video != null ? base64Encoder.encode(video) : null);
-        }
+//        if(traffalarmrecord.getVideopath() != null ) {
+//            byte[] video = FTPUtil.getFtpPicBytes(traffalarmrecord.getVideopath());
+//            eventWriteParam.setEventvideo1(video != null ? base64Encoder.encode(video) : null);
+//        }
      //String strBase64 = new BASE64Encoder().encode(FTPUtil.getFtpPicBytes(strings.get(i).get("PICPATH1").toString()));
 
         WriteResultObj writeResultObj;
         Long endtime = System.currentTimeMillis();
         try {
+            log.info("eventWriteParam========"+eventWriteParam.toString());
             writeResultObj = sendMessage(eventWriteParam);
         } catch (TimeoutException e) {
             log.error("事件推送eventwrite - sendEvent 请求超时:"+"事件="+traffalarmrecord.toString()+",错误原因=" + e.toString()+",请求总消耗时长"+(endtime-starttime)+"毫秒，"+writeInfoStr(eventWriteParam));
@@ -172,7 +173,7 @@ public class EventWriteService {
                     "nt 请求异常:"+"事件="+traffalarmrecord.toString()+",错误原因=" + e.toString()+",请求总消耗时长"+(endtime-starttime)+"毫秒，"+writeInfoStr(eventWriteParam));
             return ResultObj.error(ResponseEnum.E_9999.getCode(), e.toString());
         }
-//        log.info("事件推送eventwrite - sendEvent 请求成功:"+"事件="+traffalarmrecord.toString()+",请求总消耗时长"+(endtime-starttime)+"毫秒，"+writeInfoStr(eventWriteParam));
+        log.info("事件推送eventwrite - sendEvent success:"+"="+traffalarmrecord.toString()+",请求总消耗时长"+(endtime-starttime)+"毫秒，"+writeInfoStr(eventWriteParam));
         dealResult(writeResultObj, traffalarmrecord);
         return ResultObj.error(writeResultObj.getStatus(), writeResultObj.getMessage(), writeResultObj.getData());
     }
@@ -194,6 +195,7 @@ public class EventWriteService {
         headers.setContentType(MediaType.APPLICATION_JSON_UTF8);
         String token = stringRedisTemplate.opsForValue().get(qztoken);
         headers.add("token", token);
+        log.info("token========"+token);
         HttpEntity<List<EventWriteParam>> requestEntity = new HttpEntity<>(list, headers);
         return CompletableFuture.supplyAsync(() -> restTemplate.postForObject(url, requestEntity, WriteResultObj.class)).get(timeout, TimeUnit.MILLISECONDS);
     }

src/main/java/com/hzjt/service/TraffFlowService.java

 package com.hzjt.service;
 
+import com.alibaba.fastjson.JSON;
 import com.hzjt.domain.*;
 import com.hzjt.mapper.TraffFlowMapper;
 import lombok.extern.slf4j.Slf4j;
@@ -73,7 +74,7 @@ public class TraffFlowService {
 
     }
 
-    public  Map getfiveFlowByVideoid(String videoid) {
+    public  List<VideoDeviceTraffic> getfiveFlowByVideoid(String videoid) {
 
         return traffFlowMapper.getfiveFlowByVideoid(videoid);
 
@@ -106,12 +107,14 @@ public class TraffFlowService {
         //判断token
         String  token= qingZhiLoginCacheService.keepAlive();
         headers.add("token", token);
+        log.info("fiveflow token" +token);
+        log.info("fiveflow url" +url);
         HttpEntity<List<VideoDeviceTraffic>> requestEntity = new HttpEntity<>(list, headers);
         try{
-            ResultObj obj= CompletableFuture.supplyAsync(() -> restTemplate.postForObject(url, requestEntity, ResultObj.class)).get(timeout,TimeUnit.SECONDS);
-            log.info("send devicelist result "+obj.getMsg());
+            ResultObj obj= CompletableFuture.supplyAsync(() -> restTemplate.postForObject(url, requestEntity, ResultObj.class)).get(timeout,TimeUnit.MILLISECONDS);
+            log.info("send devicelist to guangda  result "+obj.getStatus()+obj.getData()+obj.getMsg());
         }catch (Exception ex){
-            log.error("send to guangda  list  error:"+ex.toString());
+            log.error("send devicelist to guangda  error:"+ex.toString());
         }
     }
 }
\ No newline at end of file

src/main/resources/application.properties

@@ -13,15 +13,15 @@ mybatis.configuration.map-underscore-to-camel-case=true
 mybatis.configuration.default-fetch-size=100
 mybatis.configuration.default-statement-timeout=3000
 #mybatis.mapperLocations = classpath:xxx.xml
-logging.level.com.hzjt=debug
+#logging.level.com.hzjt=debug
 
-spring.datasource.username=test
-spring.datasource.password=test
-spring.datasource.url=jdbc:oracle:thin:@192.168.168.212:1523:helowin
+#spring.datasource.username=test
+#spring.datasource.password=test
+#spring.datasource.url=jdbc:oracle:thin:@192.168.168.212:1523:helowin
 
-#spring.datasource.username=hzjt
-#spring.datasource.password=hzjt
-#spring.datasource.url=jdbc:oracle:thin:@33.50.1.22:1521:orcl
+spring.datasource.username=hzjt
+spring.datasource.password=hzjt
+spring.datasource.url=jdbc:oracle:thin:@33.50.1.22:1521:orcl
 
 spring.datasource.driverClassName=oracle.jdbc.OracleDriver
 # druid

src/main/resources/mapper/TraffFlow.xml

@@ -175,26 +175,55 @@
     from VEHICLESTATISTIC ta
     left join TRAFFVICLEWRITERESULT  a on a.VIDEOID=b.VIDEOID
 
-
-
 </select>
-    <select id="getfiveFlowByVideoid" resultType="java.util.HashMap">
-        select NVL(sum(case
-        when ruletag = '2' then
-        1
-        else
-        0
-        end),
-        0)as "downfiveflow",
-        NVL(sum(case
-        when ruletag = '1' then
-        1
-        else
-        0
-        end),
-        0) as "upfiveflow"
-        from vehicle
-        where CREATE_TIME >= SYSDATE - 5 / 1440  and  video_id=#{videoid}
+    <select id="getfiveFlowByVideoid" resultType="com.hzjt.domain.VideoDeviceTraffic">
+        SELECT
+        wbbh deviceId,
+        tdmc deviceName,
+        SYSDATE dateTime,
+        xzbh ssjg,
+        flow flowRate,
+        '1' fx,
+        '' fxmc,
+        'zksy' sjly
+        FROM
+        sbtdspsr b
+        inner JOIN (
+        SELECT
+        NVL(COUNT(*), 0) flow,
+        video_id
+        FROM
+        vehicle
+        WHERE
+        ruletag = '1'
+        AND CREATE_TIME >= SYSDATE - 5 / 1440
+        GROUP BY
+        video_id
+        ) A ON video_id = CONCAT (CONCAT(b.sbbh, '_'), b.tdbh)
+        UNION ALL
+        SELECT
+        wbbh deviceId,
+        tdmc deviceName,
+        SYSDATE dateTime,
+        xzbh ssjg,
+        flow flowRate,
+        '2' fx,
+        '' fxmc,
+        'zksy' sjly
+        FROM
+        sbtdspsr b
+        inner JOIN (
+        SELECT
+        NVL (COUNT(*), 0) flow,
+        video_id
+        FROM
+        vehicle
+        WHERE
+        ruletag = '2'
+        AND CREATE_TIME >= SYSDATE - 5 / 1440
+        GROUP BY
+        video_id
+        ) A ON video_id = CONCAT (CONCAT(b.sbbh, '_'), b.tdbh)
 
     </select>