139      财经资讯

SSL 方式接入示例__Java 接入示例_MQTT 接入(物联)_消息队列 MQ-阿里云

本文主要介绍如何使用 MQTT 的 SSL 加密通道来收发消息。

注意：

SSL 方式接入需要服务端支持，目前各个 Region 对 SSL 通道的支持情况请参考环境准备章节的文档。

本文给出的示例均基于 Eclipse Paho Java SDK 实现，SDK 下载请参见 MQTT 接入准备。如使用其他第三方的客户端，请适当修改。

1. 证书下载

如果需要使用 SSL 加密通道，需要首先下载服务端的安全证书到本地。证书下载请点击证书下载。

2. 客户端设置 SSL 属性

在添加好本地的信任证书仓库后，应用的代码中即可设置该仓库到 SSL 的属性中。具体示例如下：

public static void main(String[] args) throws IOException {
        /**
         * 设置MQTT的接入点，请根据应用所在环境选择合适的Region，不支持跨Region访问
         */
        final String broker ="ssl://mqtt-test.cn-qingdao.aliyuncs.com:8883";
        /**
         * 设置阿里云的AccessKey，用于鉴权
         */
        final String acessKey ="XXXXXX";
        /**
         * 设置阿里云的SecretKey，用于鉴权
         */
        final String secretKey ="XXXXXXX";
        /**
         * 发消息使用的一级Topic，需要先在MQ控制台里申请
         */
        final String topic ="XXXX";
        /**
         * MQTT的ClientID，一般由两部分组成，GroupID@@@DeviceID
         * 其中GroupID在MQ控制台里申请
         * DeviceID由应用方设置，可能是设备编号等，需要唯一，否则服务端拒绝重复的ClientID连接
         */
        final String clientId ="GID_XXX@@@ClientID_XXXX";
        String sign;
        MemoryPersistence persistence = new MemoryPersistence();
        try {
            final MqttClient sampleClient = new MqttClient(broker, clientId, persistence);
            final MqttConnectOptions connOpts = new MqttConnectOptions();
            System.out.println("Connecting to broker: " + broker);
            /**
             * 计算签名，将签名作为MQTT的password。
             * 签名的计算方法，参考工具类MacSignature，第一个参数是ClientID的前半部分，即GroupID
             * 第二个参数阿里云的SecretKey
             */
            sign = MacSignature.macSignature(clientId.split("@@@")[0], secretKey);
            connOpts.setUserName(acessKey);
            connOpts.setServerURIs(new String[] { broker });
            connOpts.setPassword(sign.toCharArray());
            connOpts.setCleanSession(false);
            connOpts.setKeepAliveInterval(100);
            /**
            *导入证书的路径，生成加密连接
            */
            SocketFactory socketFactory = initSSLSocket("intermedia.crt");
            connOpts.setSocketFactory(socketFactory);
            sampleClient.setCallback(new MqttCallback() {
                public void connectionLost(Throwable throwable) {
                    System.out.println("mqtt connection lost");
                    throwable.printStackTrace();
                    while(!sampleClient.isConnected()){
                        try {
                            sampleClient.connect(connOpts);
                        } catch (MqttException e) {
                            e.printStackTrace();
                        }
                        try {
                            Thread.sleep(1000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
                public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
                    System.out.println("messageArrived:" + topic + "------" + new String(mqttMessage.getPayload()));
                }
                public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                    System.out.println("deliveryComplete:" + iMqttDeliveryToken.getMessageId());
                }
            });
            sampleClient.connect(connOpts);
            for (int i = 0; i < 10; i++) {
                try {
                    String scontent = new Date()+"MQTT Test body" + i;
                    final MqttMessage message = new MqttMessage(scontent.getBytes());
                    message.setQos(0);
                    System.out.println(i+" pushed at "+new Date()+" "+ scontent);
                    /**
                     *消息发送到某个主题Topic，所有订阅这个Topic的设备都能收到这个消息
                     * 遵循MQTT的发布订阅规范，Topic也可以是多级Topic。此处设置了发送到二级Topic
                     */
                    sampleClient.publish(topic+"/notice/", message);
                    /**
                     * 如果发送P2P消息，二级Topic必须是“p2p”，三级Topic是目标的ClientID
                     * 此处设置的三级Topic需要是接收方的ClientID
                     */
                    String p2pTopic =topic+"/p2p/GID_mqttdelay3@@@DEVICEID_001";
                    sampleClient.publish(p2pTopic,message);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        } catch (Exception me) {
            me.printStackTrace();
        }
    }

private static SSLSocketFactory initSSLSocket(String certFileName) throws Exception {
        InputStream caInput = new BufferedInputStream(ClassLoader.getSystemResourceAsStream(certFileName));
        CertificateFactory cf = CertificateFactory.getInstance("X.509");
        Certificate ca = null;
        try {
            ca = cf.generateCertificate(caInput);
        } catch (CertificateException e) {
            e.printStackTrace();
        } finally {
            caInput.close();
        }
        String keyStoreType = KeyStore.getDefaultType();
        KeyStore keyStore = KeyStore.getInstance(keyStoreType);
        keyStore.load(null, null);
        keyStore.setCertificateEntry("ca", ca);
        String tmfAlgorithm = TrustManagerFactory.getDefaultAlgorithm();
        TrustManagerFactory tmf = TrustManagerFactory.getInstance(tmfAlgorithm);
        tmf.init(keyStore);
        SSLContext context = SSLContext.getInstance("TLSV1.2");
        context.init(null, tmf.getTrustManagers(), null);
        SSLSocketFactory socketFactory = context.getSocketFactory();
        return socketFactory;
    }