任务提交 mqtt 加入ssl连接

ruoyi-framework/pom.xml

@@ -53,6 +53,12 @@
             <artifactId>oshi-core</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>org.bouncycastle</groupId>
+            <artifactId>bcprov-jdk15on</artifactId>
+            <version>1.70</version>
+        </dependency>
+
         <!-- 系统模块-->
         <dependency>
             <groupId>com.ruoyi</groupId>

ruoyi-framework/src/main/java/com/ruoyi/framework/config/mqtt/MqttInboundConfiguration.java

@@ -2,6 +2,7 @@ package com.ruoyi.framework.config.mqtt;
 
 
 
+import com.ruoyi.framework.utils.SSLUtils;
 import lombok.extern.slf4j.Slf4j;
 import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -54,6 +55,13 @@ public class MqttInboundConfiguration {
         options.setCleanSession(false);
         factory.setConnectionOptions(options);
 
+        // 单向ssl/tls
+        try {
+        String caCrtFile = getClass().getResource("/emqxsl-ca.crt").getPath();
+        options.setSocketFactory(SSLUtils.getSingleSocketFactory(caCrtFile));
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
         return factory;
     }
 

ruoyi-framework/src/main/java/com/ruoyi/framework/config/mqtt/SampleCallback.java

@@ -0,0 +1,22 @@
+package com.ruoyi.framework.config.mqtt;
+
+import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
+import org.eclipse.paho.client.mqttv3.MqttCallback;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+
+public class SampleCallback implements MqttCallback {
+    // 连接丢失
+    public void connectionLost(Throwable cause) {
+        System.out.println("connection lost：" + cause.getMessage());
+    }
+
+    //  收到消息
+    public void messageArrived(String topic, MqttMessage message) {
+        System.out.println("Received message: \n  topic：" + topic + "\n  Qos：" + message.getQos() + "\n  payload：" + new String(message.getPayload()));
+    }
+
+    // 消息传递成功
+    public void deliveryComplete(IMqttDeliveryToken token) {
+        System.out.println("deliveryComplete");
+    }
+}

ruoyi-framework/src/main/java/com/ruoyi/framework/utils/SSLUtils.java

@@ -0,0 +1,38 @@
+package com.ruoyi.framework.utils;
+
+import org.bouncycastle.jce.provider.BouncyCastleProvider;
+
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocketFactory;
+import javax.net.ssl.TrustManagerFactory;
+import java.io.BufferedInputStream;
+import java.io.FileInputStream;
+import java.security.KeyStore;
+import java.security.Security;
+import java.security.cert.CertificateFactory;
+import java.security.cert.X509Certificate;
+
+public class SSLUtils {
+    // 单向认证
+    public static SSLSocketFactory getSingleSocketFactory(final String caCrtFile) throws Exception {
+        Security.addProvider(new BouncyCastleProvider());
+        X509Certificate caCert = null;
+
+        FileInputStream caCrtFileInputStream = new FileInputStream(caCrtFile);
+
+        BufferedInputStream bis = new BufferedInputStream(caCrtFileInputStream);
+        CertificateFactory cf = CertificateFactory.getInstance("X.509");
+
+        while (bis.available() > 0) {
+            caCert = (X509Certificate) cf.generateCertificate(bis);
+        }
+        KeyStore caKs = KeyStore.getInstance(KeyStore.getDefaultType());
+        caKs.load(null, null);
+        caKs.setCertificateEntry("cert-certificate", caCert);
+        TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+        tmf.init(caKs);
+        SSLContext sslContext = SSLContext.getInstance("TLSv1.2");
+        sslContext.init(null, tmf.getTrustManagers(), null);
+        return sslContext.getSocketFactory();
+    }
+}

ruoyi-framework/src/main/resources/emqxsl-ca.crt

@@ -0,0 +1,22 @@
+-----BEGIN CERTIFICATE-----
+MIIDrzCCApegAwIBAgIQCDvgVpBCRrGhdWrJWZHHSjANBgkqhkiG9w0BAQUFADBh
+MQswCQYDVQQGEwJVUzEVMBMGA1UEChMMRGlnaUNlcnQgSW5jMRkwFwYDVQQLExB3
+d3cuZGlnaWNlcnQuY29tMSAwHgYDVQQDExdEaWdpQ2VydCBHbG9iYWwgUm9vdCBD
+QTAeFw0wNjExMTAwMDAwMDBaFw0zMTExMTAwMDAwMDBaMGExCzAJBgNVBAYTAlVT
+MRUwEwYDVQQKEwxEaWdpQ2VydCBJbmMxGTAXBgNVBAsTEHd3dy5kaWdpY2VydC5j
+b20xIDAeBgNVBAMTF0RpZ2lDZXJ0IEdsb2JhbCBSb290IENBMIIBIjANBgkqhkiG
+9w0BAQEFAAOCAQ8AMIIBCgKCAQEA4jvhEXLeqKTTo1eqUKKPC3eQyaKl7hLOllsB
+CSDMAZOnTjC3U/dDxGkAV53ijSLdhwZAAIEJzs4bg7/fzTtxRuLWZscFs3YnFo97
+nh6Vfe63SKMI2tavegw5BmV/Sl0fvBf4q77uKNd0f3p4mVmFaG5cIzJLv07A6Fpt
+43C/dxC//AH2hdmoRBBYMql1GNXRor5H4idq9Joz+EkIYIvUX7Q6hL+hqkpMfT7P
+T19sdl6gSzeRntwi5m3OFBqOasv+zbMUZBfHWymeMr/y7vrTC0LUq7dBMtoM1O/4
+gdW7jVg/tRvoSSiicNoxBN33shbyTApOB6jtSj1etX+jkMOvJwIDAQABo2MwYTAO
+BgNVHQ8BAf8EBAMCAYYwDwYDVR0TAQH/BAUwAwEB/zAdBgNVHQ4EFgQUA95QNVbR
+TLtm8KPiGxvDl7I90VUwHwYDVR0jBBgwFoAUA95QNVbRTLtm8KPiGxvDl7I90VUw
+DQYJKoZIhvcNAQEFBQADggEBAMucN6pIExIK+t1EnE9SsPTfrgT1eXkIoyQY/Esr
+hMAtudXH/vTBH1jLuG2cenTnmCmrEbXjcKChzUyImZOMkXDiqw8cvpOp/2PV5Adg
+06O/nVsJ8dWO41P0jmP6P6fbtGbfYmbW0W5BjfIttep3Sp+dWOIrWcBAI+0tKIJF
+PnlUkiaY4IBIqDfv8NZ5YBberOgOzW6sRBc4L0na4UU+Krk2U886UAb3LujEV0ls
+YSEY1QSteDwsOoBrp+uvFRTp2InBuThs4pFsiv9kuXclVzDAGySj4dzp30d8tbQk
+CAUw7C29C79Fv1C5qfPrmAESrciIxpg0X40KPMbp1ZWVbd4=
+-----END CERTIFICATE-----

waterAffairs-admin/pom.xml

@@ -65,7 +65,6 @@
             <artifactId>lombok</artifactId>
             <scope>provided</scope>
         </dependency>
-
     </dependencies>
 
     <build>

waterAffairs-admin/src/main/resources/application.yml

@@ -131,11 +131,11 @@ xss:
 #mqtt
 com:
   mqtt:
-    url: ws://120.79.232.207:8083/mqtt #这里要写 tcp://  不能是 mqtt,不然不是合法的 schema，源码中有
+    url: ssl://qb8dc69b.ala.cn-hangzhou.emqxsl.cn:8883 #这里要写 tcp://  不能是 mqtt,不然不是合法的 schema，源码中有
     clientId: mqtt_test_spring
     receiveTopics: user/#     # IOT device publish msg to web, web receive this to persist
     sendTopics: user/#     # Web Publish msg to iot device, iot device need to subscribe this topic
-    username:
-    password:
+    username: admin
+    password: bw5757124
     timeout: 10
     keepalive: 20