如果想要知道 emqtt 服务器 配置ssl ，安装客户端验证是否配置成功, 或者想要了解 ssl/tls 、自签证书创建、单向认证、双向认证 等 可以看下我的另外一篇博文:https://mp.csdn.net/mdeditor/88885198

源码是java代码，Android端 也可以使用。

废话不多说了，直接上源码 （这个源码是我实际操作验证过的，源码也是我从项目代码中提取出来的部分代码，写的一个测试 ssl 单向认证 客户端无证书 的代码,想要完整mqtt操作源码 可以关注我的微博，评论里面提出来，有时间我会分享出来）。

下面引用的mqtt客户端jar包 org.eclipse.paho.client.mqttv3-1.2.0.jar 下载地址:
https://pan.baidu.com/s/1AptFyDKmae4BO7V0ZCFMeQ
提取码：8mrl

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import javax.net.SocketFactory;
import javax.net.ssl.SSLContext;
import javax.net.ssl.TrustManager;
import javax.net.ssl.X509TrustManager;
import java.security.KeyManagementException;
import java.security.NoSuchAlgorithmException;
import java.security.cert.CertificateException;
import java.security.cert.X509Certificate;

public class MqttTLSTest {

    static MqttClientCallback mqttClientCallback = new MqttClientCallback();
    static MqttAsyncClient mqttClient = null;

	static String username = "xxx";  //注意这里 填你自己的mqtt账号密码
    static String password = "xxxxx"; //注意这里 填你自己的mqtt账号密码
    //String broker = "tcp://xxx.xx.xxx.xxx:1883"; //注意这里要填自己mqtt服务器所在地址
    static String broker = "ssl://xxx.xx.xxx.xxx:8883"; //注意这里要填自己mqtt服务器所在地址
    
    public static void main(String[] args) throws InterruptedException {
        start();
        while (true){
            Thread.sleep(10000);
        }
    }

    public static void start() {
        String clientId = "mqttserver" + String.valueOf(System.currentTimeMillis());

        try {
            mqttClient = new MqttAsyncClient(broker, clientId, new MemoryPersistence());
            mqttClient.setCallback(mqttClientCallback);

            //订阅 连接mqtt服务器
            subscribeConnect();

            //发布 连接mqtt服务器
            //... 略

        } 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();
        }
    }

    public static void subscribeConnect() {
        System.out.println("订阅连接");
        if (mqttClient != null) {
            try {
                MqttConnectOptions connOpts = new MqttConnectOptions();

                connOpts.setCleanSession(true);
                connOpts.setMaxInflight(100000);
				
				//如果匿名登录连接，这2行注释掉
                connOpts.setUserName(username);
                connOpts.setPassword(password.toCharArray());

                //ssl 连接 , 这里的 TrustManager 是自己实现的，没有去校验服务端的证书
                TrustManager[] trustAllCerts = new TrustManager[1];
                TrustManager tm = new MyTM();
                trustAllCerts[0] = tm;
                SSLContext sc = SSLContext.getInstance("SSL");
                sc.init(null, trustAllCerts, null);
                SocketFactory factory = sc.getSocketFactory();
                connOpts.setSocketFactory(factory);
                //

                mqttClient.connect(connOpts, null, new IMqttActionListener() {
                    @Override
                    public void onSuccess(IMqttToken asyncActionToken) {
                        try {
                        	//订阅 topic 为test 的消息,消息质量1
                            mqttClient.subscribe("test", 1);
                            System.out.println("成功订阅topic为test的消息"); 
                        } 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();
                        }
                    }

                    @Override
                    public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                        System.out.println("mqtt 没有连接上:" + exception.getMessage());
                        exception.printStackTrace();
                    }
                });
            } 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();
            } catch (NoSuchAlgorithmException e) {
                e.printStackTrace();
            } catch (KeyManagementException e) {
                e.printStackTrace();
            }
        }
    }

	//MyTM 是自己实现的认证管理类，里面并有校验服务端的证书就返回true,永久成功！
    static class MyTM implements TrustManager, X509TrustManager {
        @Override
        public X509Certificate[] getAcceptedIssuers() {
            return null;
        }

        public boolean isServerTrusted(X509Certificate[] certs) {
            return true;
        }

        public boolean isClientTrusted(X509Certificate[] certs) {
            return true;
        }

        @Override
        public void checkServerTrusted(X509Certificate[] certs, String authType)
                throws CertificateException {
            return;
        }

        @Override
        public void checkClientTrusted(X509Certificate[] certs, String authType)
                throws CertificateException {
            return;
        }
    }

    public static class MqttClientCallback implements MqttCallback{

        @Override
        public void connectionLost(Throwable arg0)
        {
            System.out.println("mqtt 失去了连接");
        }

        @Override
        public void deliveryComplete(IMqttDeliveryToken arg0)
        {
            System.out.println("mqtt 发送完成！");
        }

        @Override
        public void messageArrived(String topic, MqttMessage message)
                throws Exception
        {
            String content = new String(message.getPayload(), "utf-8");
            System.out.println("收到mqtt消息,topic: "+topic+" ,content: "+content);
        }
    }
}

运行代码，没报异常 表示连接成功， 然后我在别的地方发送mqtt 消息,如果没有别的可以发送的客户端可以参考我之前介绍的mqtt图形客户端安装的博文:https://blog.csdn.net/a704397849/article/details/88533875#mqttjavaapp_115

• 上面介绍的是 客户端无证书 ssl 单向认证连接 。 如果不想要ssl 安全连接，仅仅是tcp 连接的话 将上面的broker 换成tcp 的broker ，ssl 部分注释掉即可，自己试试吧。

最后
这里仅仅是介绍了 客户端无证书认证连接 ，如果想要 有证书认证连接 自己网上找找吧 ，不难！