MQTT简介

import java.io.FileNotFoundException;
import java.io.IOException;
import java.security.KeyManagementException;
import java.security.KeyStoreException;
import java.security.cert.CertificateException;

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.persist.MemoryPersistence;

public class MqttPublishSample {

    public static void main(String[] args) throws KeyManagementException, CertificateException, FileNotFoundException, IOException, KeyStoreException {

        String topic        = "MQTT Examples";
        String content      = "Message from MqttPublishSample";
        int qos             = 2;//服务质量，G7中建议为0或者1
        String broker       = "ssl://10.110.111.251:8883";
        String clientId     = "JavaSample";
        MemoryPersistence persistence = new MemoryPersistence();

        try {
            MqttClient sampleClient = new MqttClient(broker, clientId, persistence);
            MqttConnectOptions connOpts = new MqttConnectOptions();
            connOpts.setCleanSession(true);
            System.out.println("Connecting to broker: "+broker);
            sampleClient.connect(connOpts);
            System.out.println("Connected");
            System.out.println("Publishing message: "+content);
            MqttMessage message = new MqttMessage(content.getBytes());
            message.setQos(qos);
            sampleClient.publish(topic, message);
            System.out.println("Message published");
            sampleClient.disconnect();
            System.out.println("Disconnected");
            System.exit(0);
        } catch(MqttException me) {
            System.out.println("reason "+me.getReasonCode());
            System.out.println("msg "+me.getMessage());
            System.out.println("loc "+me.getLocalizedMessage());
            System.out.println("cause "+me.getCause());
            System.out.println("excep "+me);
            me.printStackTrace();
        } 
    }
}

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.InputStreamReader;
import java.security.KeyFactory;
import java.security.KeyStore;
import java.security.PrivateKey;
import java.security.SecureRandom;
import java.security.cert.CertificateFactory;
import java.security.cert.X509Certificate;
import java.security.interfaces.RSAPrivateKey;
import java.security.spec.PKCS8EncodedKeySpec;

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

import org.apache.commons.codec.binary.Base64;
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.persist.MemoryPersistence;

public class PahoTlsExample {

    private static void connect() {

            String topic        = "MQTT Examples";
            String content      = "Message from MqttPublishSample";
            int qos             = 2;
            String broker       = "ssl://10.110.111.251:8883";
            String clientId     = "JavaSample";
            MemoryPersistence persistence = new MemoryPersistence();

            try {
                MqttClient sampleClient = new MqttClient(broker, clientId, persistence);
                MqttConnectOptions connOpts = new MqttConnectOptions();
                connOpts.setCleanSession(true);
                SSLSocketFactory factory = null;
                try {
                    factory = getSSLSocktet("F:/emq/cacert/cacert.crt","F:/emq/cacert/client-cert.crt","F:/emq/cacert/client-key-pkcs8.pem","brt123");
                } catch (Exception e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
                connOpts.setSocketFactory(factory);
                System.out.println("Connecting to broker: "+broker);
                sampleClient.connect(connOpts);
                System.out.println("Connected");
                System.out.println("Publishing message: "+content);
                MqttMessage message = new MqttMessage(content.getBytes());
                message.setQos(qos);
                sampleClient.publish(topic, message);
                System.out.println("Message published");
                sampleClient.disconnect();
                System.out.println("Disconnected");
                System.exit(0);
            } catch(MqttException me) {
                System.out.println("reason "+me.getReasonCode());
                System.out.println("msg "+me.getMessage());
                System.out.println("loc "+me.getLocalizedMessage());
                System.out.println("cause "+me.getCause());
                System.out.println("excep "+me);
                me.printStackTrace();
            }

        }

        private static SSLSocketFactory getSSLSocktet(String caPath,String crtPath, String keyPath, String password) throws Exception {
            // CA certificate is used to authenticate server
            CertificateFactory cAf = CertificateFactory.getInstance("X.509");
            FileInputStream caIn = new FileInputStream(caPath);
            X509Certificate ca = (X509Certificate) cAf.generateCertificate(caIn);
             KeyStore caKs = KeyStore.getInstance("JKS");
             caKs.load(null, null);
             caKs.setCertificateEntry("ca-certificate", ca);
             TrustManagerFactory tmf = TrustManagerFactory.getInstance("PKIX");
             tmf.init(caKs);

            CertificateFactory cf = CertificateFactory.getInstance("X.509");
            FileInputStream crtIn = new FileInputStream(crtPath);
            X509Certificate caCert = (X509Certificate) cf.generateCertificate(crtIn);

            crtIn.close();
            // client key and certificates are sent to server so it can authenticate
            // us
            KeyStore ks = KeyStore.getInstance(KeyStore.getDefaultType());
            ks.load(null, null);
            ks.setCertificateEntry("certificate", caCert);
            ks.setKeyEntry("private-key", getPrivateKey(keyPath), password.toCharArray(),
                    new java.security.cert.Certificate[]{caCert}  );
            KeyManagerFactory kmf = KeyManagerFactory.getInstance("PKIX");
            kmf.init(ks, password.toCharArray());

            // finally, create SSL socket factory
            SSLContext context = SSLContext.getInstance("TLSv1");

            context.init(kmf.getKeyManagers(),tmf.getTrustManagers(), new SecureRandom());

            return context.getSocketFactory();
        }

        public static PrivateKey getPrivateKey(String path) throws Exception{  

            org.apache.commons.codec.binary.Base64 base64=new Base64();
            byte[] buffer=   base64.decode(getPem(path)); 

            PKCS8EncodedKeySpec keySpec= new PKCS8EncodedKeySpec(buffer);
            KeyFactory keyFactory= KeyFactory.getInstance("RSA");  
            return (RSAPrivateKey) keyFactory.generatePrivate(keySpec);  

        } 

         private static String getPem(String path) throws Exception{
                FileInputStream fin=new FileInputStream(path);
                BufferedReader br= new BufferedReader(new InputStreamReader(fin));  
                String readLine= null;  
                StringBuilder sb= new StringBuilder();  
                while((readLine= br.readLine())!=null){  
                    if(readLine.charAt(0)=='-'){  
                        continue;  
                    }else{  
                        sb.append(readLine);  
                        sb.append('\r');  
                    }  
                }  
                fin.close();
                return sb.toString();
            }
}