mqtt服务

pom.xml

@@ -193,6 +193,15 @@
             <scope>system</scope>
             <systemPath>${pom.basedir}/lib/ojdbc6.jar</systemPath>
         </dependency>
+        <!-- mqtt -->
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-stream</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-mqtt</artifactId>
+        </dependency>
 <!--        <dependency>-->
 <!--            <groupId>jms</groupId>-->
 <!--            <artifactId>orai18n</artifactId>-->

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

+package com.hzjt.config;
+
+public interface IMQTTPublisher {
+    /**
+     * 发布消息
+     *
+     * @param topic   主题
+     * @param message 消息
+     */
+    public void publishMessage(String topic, String message);
+
+    /**
+     * 断开MQTT客户端
+     */
+    public void disconnect();
+}
+

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

+package com.hzjt.config;
+
+/**
+ * <p>
+ * 订阅者接口
+ */
+public interface IMQTTSubscriber {
+
+    /**
+     * 订阅消息
+     *
+     * @param topic
+     */
+    public void subscribeMessage(String topic);
+
+    /**
+     * 断开MQTT客户端
+     */
+    public void disconnect();
+}

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

+package com.hzjt.config;
+
+import com.hzjt.mapper.SbtdspsrMapper;
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.client.mqttv3.*;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+@Component
+@Slf4j
+public class MQTTSubsribe {
+
+    @Value("${spring.mqtt.password}")
+    private String password;
+
+    @Value("${spring.mqtt.username}")
+    private String username;
+
+    @Value("${spring.mqtt.url}")
+    private String url;
+
+    @Value("${spring.mqtt.qos}")
+    private Integer qos;
+
+    // 连接超时时间
+    @Value("${spring.mqtt.completionTimeout}")
+    private int completionTimeout;
+    @Autowired
+    SbtdspsrMapper sbtdspsrMapper;
+
+
+    private String[] topics;
+
+    private int[] qoslist;
+    private MqttClient client;
+
+    private MqttConnectOptions mqttConnectOptions;
+
+    @Autowired
+    private PushCallback pushCallback;
+
+    private ScheduledExecutorService scheduled;
+
+    public void startReconnect() {
+        this.scheduled = Executors.newSingleThreadScheduledExecutor();
+        //    定时任务——重新连接mqtt服务器
+        this.scheduled.scheduleAtFixedRate(new Runnable() {
+            public void run() {
+                if (!MQTTSubsribe.this.client.isConnected()) {
+                    try {
+                        log.info("---mqtt reconnect ---");
+                        List<String> sbbhlist=  sbtdspsrMapper.selectAllWbbh();
+                        client.connect(mqttConnectOptions);
+                        subscribe(sbbhlist);
+//                        client.subscribe(topics, qoslist);
+                    } catch (MqttSecurityException var2) {
+                        var2.printStackTrace();
+                    } catch (MqttException var3) {
+                        var3.printStackTrace();
+                    }
+                }
+
+            }
+        }, 5000L, 10000L, TimeUnit.MILLISECONDS);
+    }
+
+
+    //	对mqttConnectOptions对象的常规设置
+    public MqttConnectOptions getMqttConnectOptions() {
+        this.mqttConnectOptions = new MqttConnectOptions();
+        this.mqttConnectOptions.setCleanSession(true);
+//        this.mqttConnectOptions.setUserName(username);
+//        this.mqttConnectOptions.setPassword(password.toCharArray());
+        this.mqttConnectOptions.setServerURIs(new String[]{url});
+        this.mqttConnectOptions.setConnectionTimeout(completionTimeout);
+        this.mqttConnectOptions.setKeepAliveInterval(2000);
+
+        return mqttConnectOptions;
+    }
+
+    //	连接mqtt服务器订阅信息方法
+//	topic也可作为参数传入
+    public void subscribe(List<String> topics) {
+        if(null!=topics &&topics.size()>0) {
+            try {
+                this.topics = topics.toArray(new String[topics.size()]);
+                this.client = new MqttClient(url, getClientId(), new MemoryPersistence());
+                this.getMqttConnectOptions();
+                this.client.setCallback(this.pushCallback);
+                this.client.connect(this.mqttConnectOptions);
+                //遍历topic ，获得qs
+                int[] qoslist=  new int[topics.size()];
+                for( int i=0;i<topics.size();i++)
+                {qoslist[i]=qos;
+                }
+                this.qoslist=qoslist;
+//            可将订阅的一个或多个topic都存入数组中，同时订阅
+//            String[] topic1 = {defaultTopic};
+                this.client.subscribe(this.topics, qoslist);
+                boolean connected = this.client.isConnected();
+                log.info("连接状态为：" + connected);
+                String flag = connected ? "成功" : "失败";
+            } catch (MqttException e) {
+                log.info(e.toString());
+                startReconnect();
+            }
+        }else {
+            log.info("topic is  empty");
+        }
+    }
+
+    //	随机生成唯一client.id方法
+    public String getClientId() {
+        String nums = "";
+        String[] codeChars = {"0", "1", "2", "3", "4", "5", "6", "7", "8", "9",
+                "a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k", "l", "m", "n", "o", "p", "q", "r", "s", "t", "u", "v", "w", "x", "y", "z",
+                "A", "B", "C", "D", "E", "F", "G", "H", "I", "J", "K", "L", "M", "N", "O", "P", "Q", "R", "S", "T", "U", "V", "W", "X", "Y", "Z"};
+        for (int i = 0; i < 23; i++) {
+            int charNum = (int) Math.floor(Math.random() * codeChars.length);
+            nums = nums + codeChars[charNum];
+        }
+        return nums;
+    }
+
+    public void publishMessage(String topic, String message) {
+        try {
+            MqttMessage mqttmessage = new MqttMessage(message.getBytes());
+            mqttmessage.setQos(this.qos);
+            client.publish(topic, mqttmessage);
+        } catch (MqttException me) {
+            log.info(me.toString());
+        }
+    }
+
+}
\ No newline at end of file

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

+package com.hzjt.config;
+
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
+import org.eclipse.paho.client.mqttv3.MqttCallback;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.core.RedisTemplate;
+
+import java.util.concurrent.ScheduledExecutorService;
+
+@Configuration
+@Slf4j
+public class PushCallback implements MqttCallback,IMQTTPublisher {
+
+    @Autowired
+    private MQTTSubsribe mqttSubsribe;
+
+    @Autowired
+    private RedisTemplate redisTemplate;
+
+    //    定时任务——定时缓存查询的数据
+    private ScheduledExecutorService scheduled;
+
+    @Override
+    public void connectionLost(Throwable throwable) {
+        // 连接丢失后，一般在这里面进行重连
+        log.info("连接断开，可以做重连");
+        this.mqttSubsribe.startReconnect();
+    }
+
+
+    @Override
+
+    public void messageArrived(String topic, MqttMessage message) throws Exception {
+        log.info("topic result"+topic+new String(message.getPayload()));
+        // subscribe后得到的消息会执行到这里面
+        String[] topics = topic.split("/");
+//        this.scheduled = Executors.newSingleThreadScheduledExecutor();
+//        //周期定时方法，可以在里面进行定时数据存储操作，我测试时是先将数据存储到了redis中，可做实时数据来用
+//        this.scheduled.scheduleAtFixedRate(new Runnable() {
+//            @Override
+//            public void run() {
+//                String key = topics[0] + topics[1];
+//                String value = new String(message.getPayload());
+//                log.info("topic result");
+////                if (redisTemplate.hasKey(key)){
+////                    redisTemplate.delete(key);
+////                }
+////                redisTemplate.opsForValue().set(key,value);
+////                System.out.println("redis缓存数据"+value);
+////                //下面是我对数据的一些处理，仅供参考
+////                JSONObject jsonObject = JSONObject.parseObject(value);
+////                Map map = jsonObject;
+////                List<Map<String,Object>> list = (List<Map<String, Object>>) ((Map) map.get("lines")).get("data");
+////                log.error("-----------------------------"+list);
+//            }
+//            //此处 120为每120秒执行一次
+//        },0,120, TimeUnit.SECONDS);
+    }
+
+    @Override
+    public void deliveryComplete(IMqttDeliveryToken token) {
+        log.info("deliveryComplete---------" + token.isComplete());
+    }
+
+    @Override
+    public void publishMessage(String topic, String message) {
+
+    }
+
+    @Override
+    public void disconnect() {
+        log.info("disconnect---------" );
+    }
+
+}
\ No newline at end of file

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

@@ -2,6 +2,7 @@ package com.hzjt.controller;
 
 import com.hzjt.domain.ResultObj;
 import com.hzjt.domain.Traffalarmrecord;
+import com.hzjt.domain.WriteResultObj;
 import com.hzjt.service.EventWriteService;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.web.bind.annotation.GetMapping;
@@ -24,7 +25,7 @@ public class EventWriteController {
     }
 
     @PostMapping("/sendEvents")
-    public List<ResultObj> sendEvents(@RequestParam("recordid") String recordid){
+    public List<WriteResultObj> sendEvents(@RequestParam("recordid") String recordid){
         List<Long> longrecorid=new ArrayList<>();
         for(String id:recordid.split(","))
         {
@@ -34,7 +35,7 @@ public class EventWriteController {
     }
 
     @GetMapping("/sendEventsByids")
-    public List<ResultObj> sendEventsByids(){
+    public List<WriteResultObj> sendEventsByids(){
         List<Long> recordid=new ArrayList<>();
         recordid.add(Long.valueOf(93228));
         recordid.add(Long.valueOf(93834));

src/main/java/com/hzjt/domain/MqttAlarm.java

+package com.hzjt.domain;
+
+import lombok.Data;
+
+import java.math.BigDecimal;
+import java.util.List;
+import java.util.Map;
+
+@Data
+public class MqttAlarm {
+    private String deviceId;
+    private String eventTypeId;
+    private String companyId;
+    private String alarmTime;
+    private String x;
+    private String y;
+    private String  height;
+    private String width;
+    private String imagePath;
+    private String videoPath;
+    private String content;
+}

src/main/java/com/hzjt/domain/ResponseEnum.java

@@ -25,7 +25,7 @@ public enum ResponseEnum {
     E_1010(1010,"数据保存失败"),
     E_1011(1011,"数据重复"),
     E_9999(9999,"系统异常"),
-
+    E_9991(9991,"发送异常"),
     SUCCESS(200,"请求成功");
 
     private int code;

src/main/java/com/hzjt/domain/WriteResultObj.java

@@ -16,7 +16,11 @@ public class WriteResultObj {
         this.message = message;
         this.status = status;
     }
-
+    public WriteResultObj(Integer status, String message,String data) {
+        this.message = message;
+        this.status = status;
+        this.data=data;
+    }
     @Override
     public String toString() {
         return "WriteResultObj{" +

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

@@ -81,8 +81,8 @@ public class WebSocket {
      *
      * @param message
      */
-    public void GroupSending(String message) {
-
+    public boolean GroupSending(String message){
+        boolean send=true;
             for (String name : webSocketSet.keySet()) {
                 try {
 
@@ -91,9 +91,10 @@ public class WebSocket {
                             webSocketSet.get(name).session.getBasicRemote().sendText(message);
                            }
                     } catch (Exception e) {
-                    e.printStackTrace();
+                    send=false;
                 }
             }
+            return send;
     }
 
     /**

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

@@ -30,4 +30,6 @@ public interface SbtdspsrMapper extends BaseMapper<Sbtdspsr>, ConditionMapper<Sb
     List<Sbtdspsr> selectBySbbh(@Param("sbbh") String sbbh,@Param("tdbh") Integer tdbh);
 
     List<SbtdspsrParam> selectDeviceWrite();
+
+    List<String> selectAllWbbh();
 }
\ No newline at end of file

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

@@ -33,4 +33,6 @@ public interface TraffAlarmRecordMapper extends BaseMapper<Traffalarmrecord>, Co
 
     String   seletManualStatus();
 
+    String seletmqttbh(@Param("recordtype") String recordtype);
+
 }
\ No newline at end of file

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

 package com.hzjt.service;
 
 import com.hzjt.domain.*;
+import com.hzjt.handler.WebSocket;
 import com.hzjt.mapper.SbtdspsrMapper;
 import com.hzjt.mapper.TraffAlarmRecordMapper;
-import com.hzjt.util.DateUtils;
-import com.hzjt.util.FTPUtil;
-import com.hzjt.util.StringEnum;
-import com.hzjt.util.ThreadPoolUtil;
+import com.hzjt.util.*;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -67,7 +65,11 @@ public class EventWriteService {
     @Autowired
     private TraffAlarmRecordMapper traffalarmrecordMapper;
 
+    @Autowired
+    WebSocket webSocket;
+
 
+    BASE64Encoder base64Encoder = new BASE64Encoder();
     private static  CompletionService<ResultObj> completionService  = new ExecutorCompletionService<ResultObj>(ThreadPoolUtil.getPool());
     public ResultObj sendEvent(Long recordid) {
         Traffalarmrecord traffalarmrecord = traffalarmrecordMapper.selectByPrimaryKey(recordid);
@@ -84,21 +86,73 @@ public class EventWriteService {
         return resultObj;
     }
 
-    public List<ResultObj> sendEvents(List<Long> recordid) {
+    public List<WriteResultObj> sendEvents(List<Long> recordid) {
         Condition condition = new Condition(Traffalarmrecord.class);
         Example.Criteria criteria = condition.createCriteria();
         criteria.andIn("recordid",recordid);
         List<Traffalarmrecord> traffalarmrecordList = traffalarmrecordMapper.selectByCondition(condition);
-        List<ResultObj> resultObj=new ArrayList<>();
-        try {
-            resultObj = sendAllMessage(traffalarmrecordList);
+        List<WriteResultObj> resultObj=new ArrayList<>();
+        //将返回结果转化为Alarm格式
+        List<Alarm> alarmList=new ArrayList<>();
+        for(Traffalarmrecord record : traffalarmrecordList){
+            Alarm  alarm=new Alarm();
+            alarm.setDept(record.getAreaid().toString());
+            alarm.setIncident_type(record.getRecordtype());
+            alarm.setType("TRAFFIC_INCIDENT_ALARM");
+            alarm.setTs(String.valueOf(record.getRecordtime().getTime()));
+            alarm.setVideo_id(record.getFdid());
+            List<String> img_urls=new ArrayList<>();
+            img_urls.add(record.getImg1urlfrom());
+            img_urls.add(record.getImg2urlfrom());
+            img_urls.add(record.getImg3urlfrom());
+            img_urls.add(record.getImg4urlfrom());
+            img_urls.add(record.getImg5urlfrom());
+            alarm.setImg_urls(img_urls);
+            alarm.setId(record.getRecordid().toString());
+            alarm.setObjLabel(record.getObjlable());
+           // alarm.setObj_location(record.getLocation());
+            List<String> img_base64=new ArrayList<>();
+            if(record.getImg1path() != null ){
+                byte[]  Img=FTPUtil.getFtpPicBytes(record.getImg1path());
+                img_base64.add(Img!=null?base64Encoder.encode(Img) : null);
+            }
+            if(record.getImg2path() != null ){
+                byte[]  Img=FTPUtil.getFtpPicBytes(record.getImg2path());
+                img_base64.add(Img!= null ? base64Encoder.encode(Img) : null);
+            }
+            if(record.getImg3path() != null ) {
+                byte[] Img = FTPUtil.getFtpPicBytes(record.getImg3path());
+                img_base64.add(Img != null ? base64Encoder.encode(Img) : null);
 
-        } catch (TimeoutException e) {
-            log.error("eventwrite - sendEvents 请求超时:" + e.toString());
-            resultObj.add(ResultObj.error(ResponseEnum.E_1008.getCode(), ResponseEnum.E_1008.getMsg()));
-        } catch (Exception e) {
-            log.error("eventwrite - sendEvents 异常:" + e.toString());
-             resultObj.add(ResultObj.error(ResponseEnum.E_9999.getCode(), e.toString()));
+            }
+            if(record.getImg4path() != null ) {
+                byte[] Img = FTPUtil.getFtpPicBytes(record.getImg4path());
+                img_base64.add(Img != null ? base64Encoder.encode(Img) : null);
+            }
+            if(record.getImg5path() != null ) {
+                byte[] Img = FTPUtil.getFtpPicBytes(record.getImg5path());
+                img_base64.add(Img != null ? base64Encoder.encode(Img) : null);
+            }
+            alarm.setImg_base64(img_base64);
+            if(record.getVideopath() != null ) {
+                alarm.setVideo_record_url(record.getVideopath());
+            }
+            //推送给第三方
+                boolean send =webSocket.GroupSending(JsonUtil.beanToString(alarm));
+            if(send){
+                //更新推送状态
+                record.setPushstatus(0);
+                record.setPushdesc("发送成功");
+                traffAlarmRecordMapper.updatePushEvent(record);
+                resultObj.add(new  WriteResultObj(200,"推送成功",record.getRecordid().toString()));
+            }
+            else {
+                //推送失败
+                record.setPushstatus(1);
+                record.setPushdesc("发送失败");
+                traffAlarmRecordMapper.updatePushEvent(record);
+                resultObj.add(new  WriteResultObj(ResponseEnum.E_9991.getCode(),"发送失败",record.getRecordid().toString()));
+               }
         }
         return resultObj;
     }

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

@@ -44,6 +44,7 @@ public class TraffdevicewriteresultService {
 
     @Value("${qingzhi.devicewritesupplier.name}")
     String devicesupplier;
+
     @Value("${qingzhi.redis.token}")
     String qztoken;
 

src/main/resources/application.properties

@@ -13,16 +13,16 @@ 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
 
 
 #   Mysql���ݿ�-����Դ����
-#spring.datasource.username=hzjt
-#spring.datasource.password=hzjt
-#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.65.250.179:1521:helowin
+spring.datasource.url=jdbc:oracle:thin:@192.168.168.212:1521:helowin
+#spring.datasource.username=hzjt
+#spring.datasource.password=hzjt
+#spring.datasource.url=jdbc:oracle:thin:@33.65.219.103:1521:helowin
 spring.datasource.driverClassName=oracle.jdbc.OracleDriver
 # druid
 spring.datasource.type=com.alibaba.druid.pool.DruidDataSource
@@ -43,8 +43,8 @@ spring.datasource.connectionProperties=druid.stat.mergeSql=true;druid.stat.slowS
 spring.datasource.useGlobalDataSourceStat=true 
 
 #redis
-spring.redis.host=33.65.250.179
-#spring.redis.host=127.0.0.1
+#spring.redis.host=33.65.219.103
+spring.redis.host=127.0.0.1
 spring.redis.port=6379
 #spring.redis.password=123456
 spring.redis.database=0
@@ -54,12 +54,14 @@ spring.redis.pool.max-idle=8
 spring.redis.pool.min-idle=0
 spring.redis.timeout=2000
 spring.session.store-type=none
-ip.host=33.65.250.179
+ip.host=33.65.219.103
 port=21
 message.dept=33030
 message.rate=10
 
-ftp.host=33.65.250.179:21:hzjt:1qaz2wsx
+ftp.host=33.65.219.40:21:hzjt:1qaz2wsx
+ftppath=hzjt:1qaz2wsx@33.65.219.40:21
+ftpServiceUrl=http://33.65.219.103:8089/api/alg/files
 
 alarm.subscribe.data.key=gs:traff:alarmlist
 
@@ -89,4 +91,17 @@ flv.url=http://33.50.1.21:57080/ecvs/get_play_list
 qingzhi.login.username=zksy
 qingzhi.login.password=zksy@123
 qingzhi.login.url=http://33.50.1.213:38080/api/auth/login
-qingzhi.login.keepaliveurl=http://33.50.1.213:38080/api/auth/token/keepalive
\ No newline at end of file
+qingzhi.login.keepaliveurl=http://33.50.1.213:38080/api/auth/token/keepalive
+
+#mqtt配置 - start
+#用户名
+spring.mqtt.username =
+#密码
+spring.mqtt.password =
+#服务器连接地址
+spring.mqtt.url = tcp://12.1.97.11:1883
+#连接超时
+spring.mqtt.completionTimeout=3000
+spring.mqtt.qos=2
+
+#mqtt配置 - end

src/main/resources/mapper/SbtdspsrMapper.xml

@@ -83,7 +83,7 @@
     </select>
 
     <select id="selectBySbbh" resultType="com.hzjt.domain.Sbtdspsr">
-        SELECT A.* FROM SBTDSPSR A where  sbbh=#{sbbh} and tdbh=#{tdbh}
+        SELECT A.*  FROM SBTDSPSR A where  sbbh=#{sbbh} and tdbh=#{tdbh}
     </select>
 
     <select id="selectDeviceWrite" resultType="com.hzjt.domain.SbtdspsrParam">
@@ -92,4 +92,8 @@
         LEFT JOIN TRAFFDEVICECONFIG C ON C.FDID = A.SBBH AND C.CHANNELID = A.TDBH
         WHERE B.PUSHSTATUS != 0 OR B.PUSHDESC IS NULL
     </select>
+
+    <select id="selectAllWbbh" resultType="java.lang.String">
+        select concat(concat('event/',wbbh),'/videoEvent') from  sbtdspsr where wbbh is not null
+    </select>
 </mapper>
\ No newline at end of file

src/main/resources/mapper/Traffalarmrecord.xml

@@ -179,4 +179,8 @@
 <select id="seletManualStatus" resultType="java.lang.String">
     select name  from  t_code  where type=4
 </select>
+
+    <select id="seletmqttbh" resultType="java.lang.String">
+    select mqttbh  from  t_code  where type=1 and  key=#{recordtype}
+</select>
 </mapper>
\ No newline at end of file