官方文档：Eclipse Paho Java Client

API：Package org.eclipse.paho.client.mqttv3

Paho 基础知识

Paho Java 客户端是用 Java 编写的 MQTT 客户端库，用于开发在 JVM 或其他 Java 兼容平台（例如 Android ）上运行的应用程序 Paho Java 客户端提供了两个 API：MqttAsyncClient 提供了一个完全异步的 API，该活动通过注册的回调通知完成活动。 MqttClient 是 MqttAsyncClient 周围的同步包装器，其中函数与应用程序同步显示。

常用类和接口说明：

IMqttActionListener：异步操作完成后，将通知此接口的实现。
MqttCallback：使与客户端相关的异步事件发生时通知应用程序。
MqttAsyncClient：轻量级客户端，用于使用允许操作在后台运行的非阻塞方法与MQTT服务器通信。
MqttConnectOptions：包含用于控制客户端如何连接到服务器的选项集。
MqttMessage：MQTT消息包含应用程序有效负载和指定如何传递消息的选项。该消息包括表示为byte []的“有效负载”（消息的主体）。

Paho Java 实践

S1：添加依赖

implementation 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.2.2'

S2：引入权限

S3：添加工具类（我自制）

/** * Paho Java Client 工具类 * @author zhanghairong * @version 2018-2-18 */public class MyMqttUtil {    // Log TAG    private final String TAG = "MyMqttUtil";    // MyMqttUtil 上下文环境    private Context mContext;    // MyMqttUtil 单例    private static MyMqttUtil mMyMqttUtilInstance;    // MqttAsyncClient 要连接的服务器    private final String serverURI = "tcp://xxxx:1883";    // MqttAsyncClient ID    private final String clientID = "123456";    // MqttAsyncClient 用于连接服务器的用户名    private final String userName = "root";    // MqttAsyncClient 用于连接服务器的密码    private final String password = "root";    // 断开连接的主题    private String Topic_DisCon = "Topic_DisCon";    // 数据持久化设置    private MqttDefaultFilePersistence dataStore;    // 客户端如何连接到服务器的选项集    private MqttConnectOptions mMqttConnectOptions;    // MqttAsyncClient 实例    private MqttAsyncClient mMqttAsyncClient;    // 私有构造    private MyMqttUtil(Context context) {        this.mContext = context;        initMqttClient();    }    // 单例模式    public static MyMqttUtil getInstance(Context context) {        if (mMyMqttUtilInstance == null) {            synchronized (MyMqttUtil.class) {                if (mMyMqttUtilInstance == null) {                    mMyMqttUtilInstance = new MyMqttUtil(context);                }            }        }        return mMyMqttUtilInstance;    }    // 公共接口IMqttActionListener：异步操作完成后，将通知此接口的实现。在此处监听连接结果    private IMqttActionListener myIMqttActionListener = new IMqttActionListener() {        @Override// 操作成功完成后，将调用此方法。        public void onSuccess(IMqttToken asyncActionToken) {            Log.d(TAG, "MQTT 连接成功");        }        @Override// 操作失败时将调用此方法。        public void onFailure(IMqttToken asyncActionToken, Throwable exception) {            Log.d(TAG, "MQTT 连接失败");        }    };    // 公共接口MqttCallback：使与客户端相关的异步事件发生时通知应用程序。    private MqttCallback mMqttCallback = new MqttCallback() {        @Override// 与服务器的连接丢失时，将调用此方法。        public void connectionLost(Throwable cause) {            Log.d(TAG, "与服务器的连接丢失,尝试重连");        }        @Override// 从服务器收到消息时，将调用此方法。        public void messageArrived(String topic, MqttMessage message) throws Exception {            Log.d(TAG, "从服务器收到消息 topic 是：" + topic + " message 是：" + new String(message.getPayload()));        }        @Override// 在完成消息传递并收到所有确认后调用。        public void deliveryComplete(IMqttDeliveryToken token) {            Log.d(TAG, "完成消息传递并收到所有确认");        }    };    // 初始化配置以及连接服务器    public void initMqttClient(){        try {            // MqttConnectOptions：包含用于控制客户端如何连接到服务器的选项集。            mMqttConnectOptions = new MqttConnectOptions();            // 设置MQTT版本。默认操作是连接版本3.1.1，如果失败则回落到3.1。分别使用MQTT_VERSION_3_1_1或MQTT_VERSION_3_1选项，可以选择特定的版本3.1.1或3.1，而不会回退。            mMqttConnectOptions.setMqttVersion(MqttConnectOptions.MQTT_VERSION_3_1_1);            // 设置连接超时值。该值（以秒为单位）定义了客户端将等待与MQTT服务器建立网络连接的最大时间间隔。默认超时为30秒。值为0将禁用超时处理，这意味着客户端将等待，直到成功建立网络连接或失败。            mMqttConnectOptions.setConnectionTimeout(10);            /*            设置客户端和服务器是否应该记住重新启动和重新连接之间的状态。            如果设置为false，则客户端和服务器都将在客户端，服务器和连接重新启动期间保持状态。维持状态：                即使重新启动客户端，服务器或连接，消息传递也将可靠地满足指定的QOS。                服务器会将订阅视为持久订阅。            如果设置为true，则客户端和服务器将不会在客户端，服务器或连接的重新启动期间保持状态。这表示                如果重新启动客户端，服务器或连接，则无法维持到指定QOS的消息传递                服务器会将订阅视为非持久订阅            */            mMqttConnectOptions.setCleanSession(true);            /*            设置“保持活动”间隔。该值以秒为单位，定义了发送或接收消息之间的最大时间间隔。            它使客户端能够检测服务器是否不再可用，而不必等待TCP / IP超时。            客户端将确保在每个有效期内，至少有一条消息在网络上传播。            在这段时间内，如果没有与数据相关的消息，则客户端会发送一个很小的“ ping”消息，服务器将予以确认。            值为0将禁用客户端中的keepalive处理。             */            mMqttConnectOptions.setKeepAliveInterval(10);            // 设置用于连接的用户名。            mMqttConnectOptions.setUserName(userName);            // 设置用于连接的密码。            mMqttConnectOptions.setPassword(password.toCharArray());            /*            设置连接的“最后的遗嘱”（LWT）。如果此客户端意外断开与服务器的连接，则服务器将使用提供的详细信息向自身发布消息。            topic -要发布到的主题。            payload -消息的字节有效负载。            qos -以（0、1或2）发布消息的服务质量。            retained -是否保留消息。             */            mMqttConnectOptions.setWill(Topic_DisCon, "close".getBytes(), 1, false);            /*            设置如果连接断开，客户端是否将自动尝试重新连接到服务器。            如果设置为false，则在连接丢失的情况下，客户端将不会尝试自动重新连接到服务器。            如果设置为true，则在连接断开的情况下，客户端将尝试重新连接到服务器。它最初将等待1秒钟，然后再尝试重新连接，对于每次失败的重新连接尝试，延迟都会加倍，直到2分钟为止，此时延迟将保持2分钟。             */            mMqttConnectOptions.setAutomaticReconnect(true);            dataStore = new MqttDefaultFilePersistence(System.getProperty("java.io.tmpdir"));            mMqttAsyncClient = new MqttAsyncClient(serverURI, clientID, dataStore);            mMqttAsyncClient.setCallback(mMqttCallback);// 设置一个回调侦听器以用于异步发生的事件。        } catch (MqttException e) {            e.printStackTrace();        }    }    // 连接服务器    public void connectServer(){        try {            if((mMqttAsyncClient != null) && (!mMqttAsyncClient.isConnected())){                mMqttAsyncClient.connect(mMqttConnectOptions, mContext, myIMqttActionListener);// 使用指定的选项连接到MQTT服务器            }        } catch (MqttException e) {            e.printStackTrace();        }    }    // 订阅主题    public void subTopic(String subTopic, int qos) {        if (mMqttAsyncClient != null && mMqttAsyncClient.isConnected()) {            try {                mMqttAsyncClient.subscribe(subTopic, qos);                Log.d(TAG, "订阅主题：" + subTopic + " 成功");            } catch (MqttException e) {                e.printStackTrace();            }        }        else {            Log.d(TAG, "服务器未连接");        }    }    // 推送数据    public void pubMessage(String pubTopic, String message){        if (mMqttAsyncClient != null && mMqttAsyncClient.isConnected()) {            try {                MqttMessage mqttMessage = new MqttMessage(message.getBytes());                mMqttAsyncClient.publish(pubTopic, mqttMessage);                Log.d(TAG, "推送主题：" + pubTopic + "数据 " + message + " 成功");            } catch (MqttException e) {                e.printStackTrace();            }        }        else {            Log.d(TAG, "服务器未连接");        }    }    // 断开连接    public void disConnectServer(){        if (mMqttAsyncClient != null && mMqttAsyncClient.isConnected()) {            try {                mMqttAsyncClient.disconnect();                Log.d(TAG, "服务器断开成功");            } catch (MqttException e) {                e.printStackTrace();            }        }        else {            Log.d(TAG, "服务器未连接");        }    }}

S4：测试

btn_connect.setOnClickListener(new View.OnClickListener() {    @Override    public void onClick(View v) {        MyMqttUtil.getInstance(Sample_MQTT.this).connectServer();    }});btn_publish.setOnClickListener(new View.OnClickListener() {    @Override    public void onClick(View v) {        MyMqttUtil.getInstance(Sample_MQTT.this).pubMessage("Topic_Pub", "123456");    }});btn_subscribe.setOnClickListener(new View.OnClickListener() {    @Override    public void onClick(View v) {        MyMqttUtil.getInstance(Sample_MQTT.this).subTopic("Topic_Sub", 1);    }});btn_disconnect.setOnClickListener(new View.OnClickListener() {    @Override    public void onClick(View v) {        MyMqttUtil.getInstance(Sample_MQTT.this).disConnectServer();    }});

Android(安卓)开发笔记 04 篇：Mqtt 框架 - Paho Java

Paho 基础知识

Paho Java 实践

更多相关文章

随机推荐