Java连接MQTT服务-wss方式

特别提示：本人博客部分有参考网络其他博客，但均是本人亲手编写过并验证通过。如发现博客有错误，请及时提出以免误导其他人，谢谢！欢迎转载，但记得标明文章出处：http://www.cnblogs.com/mao2080/

说明：前面介绍的tcp、ws方式适合Java程序在局域网内使用，不涉及到安全问题。但由于Android手机APP需要通过websocket方式来连接,就必须考虑安全性问题了，这时候就采用了wss+CA证书方式进行认证，而且在数据传输中也是加密的。大致与ws方式相同，只不过是加了证书。

1、Java代码

package com.mao.mqtt;

import java.io.FileInputStream;
import java.security.KeyStore;
import java.security.cert.CertificateFactory;
import java.security.cert.X509Certificate;
import java.text.SimpleDateFormat;
import java.util.Date;

import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLSocketFactory;
import javax.net.ssl.TrustManagerFactory;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttTopic;

/**
 * 
 * 功能描述：MQTT测试
 * 创建人： mao2080@sina.com
 * 创建时间：2017年7月4日 下午5:08:59
 * 修改人： mao2080@sina.com
 * 修改时间：2017年7月4日 下午5:08:59
 */
public class MQTTTest_wss {

    /**MQTT服务端ip及端口*/
    private static String host = "wss://ip:443";

    /**账号*/
    private static String username = "li2080";

    /**密码*/
    private static String password = "123";

    /**订阅的主题*/
    private static String subTopic = "a/b/c";
    
    /**clientID*/
    private static String clientId = "li2080";
    
    /**发布的主题*/
    private static String pubTopic = "a/b/c";

    /**MQTT-Client*/
    private static MqttClient client;
    
    /**证书路径*/
    private static String caPath = "E:\mqtt-demo\certfile\CA.crt";
    
    /**
     * @throws InterruptedException 
     * @throws MqttException */
    public static void main(String[] args) throws InterruptedException, MqttException {
        
        // 订阅消息的方法
        subscribe();
//        
        publish();
    }

    /**
     * 
     * 描述：订阅信息
     * @author mao2080@sina.com
     * @created 2017年7月4日 下午4:53:47
     * @since 
     * @return
     */
    public static void subscribe() {
        try {
            // 创建MqttClient
            MQTTTest_wss.getClient().setCallback(new MqttCallback() {

                public void connectionLost(Throwable arg0) {
                    
                }

                public void messageArrived(String topic, MqttMessage message) throws Exception {
                    System.out.println("MQTT Rece:" + message.toString());
                }

                public void deliveryComplete(IMqttDeliveryToken token) {

                }

            });
            MQTTTest_wss.getClient().subscribe(subTopic, 0);
            System.out.println("连接状态:" + client.isConnected());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    /**
     * 
     * 描述：获取MqttClient
     * @author mao2080@sina.com
     * @created 2017年7月6日 上午9:56:37
     * @since 
     * @return
     * @throws MqttException
     */
    public static MqttClient getClient() throws MqttException{
        try {
            if(client == null){
                client = new MqttClient(host, clientId);
                MqttConnectOptions conOptions = new MqttConnectOptions();
                conOptions.setUserName(username);
                conOptions.setPassword(password.toCharArray());
                conOptions.setCleanSession(true);
                conOptions.setSocketFactory(getSSLSocktet(caPath));
                client.connect(conOptions);
            }
            if(!client.isConnected()){
                client.reconnect();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return client;
    }

    /**
     * 
     * 描述：发布信息
     * @author mao2080@sina.com
     * @throws MqttException 
     * @created 2017年7月4日 下午4:53:32
     * @since
     */
    public static void publish() throws MqttException {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        String sendMsg = "{time:"+sdf.format(new Date())+", content:"+com.lds.iot.common.util.UUIDUtil.getLowerLetterNumber(15)+", from: java console}";
        try {
            MqttTopic topic = MQTTTest_wss.getClient().getTopic(pubTopic);
            MqttMessage message = new MqttMessage(sendMsg.getBytes());
            message.setQos(0);
            topic.publish(message);
            System.out.println("MQTT Send:" + sendMsg);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    /**
     * 获取SSLSocketFactory
     * @param caPath
     * @return
     * @throws Exception
     */
    public static SSLSocketFactory getSSLSocktet(String caPath) throws Exception {
        CertificateFactory cAf = CertificateFactory.getInstance("X.509");
        FileInputStream caIn = new FileInputStream(caPath);
        X509Certificate ca = (X509Certificate) cAf.generateCertificate(caIn);
        KeyStore caks = KeyStore.getInstance(KeyStore.getDefaultType());
        caks.load(null, null);
        caks.setCertificateEntry("ca-certificate", ca);
        TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
        tmf.init(caks);
        caIn.close();
        SSLContext context = SSLContext.getInstance("TLSv1");
        context.init(null, tmf.getTrustManagers(), null);
        return context.getSocketFactory();
    }
    
}

2、Maven配置

<dependency>
      <groupId>org.eclipse.paho</groupId>
      <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
      <version>1.2.0</version>
</dependency>

3、服务端配置

请参考另一篇blog：http://www.cnblogs.com/mao2080/p/7772893.html

4、运行效果