MQTT小试

发表于

最近项目里要做长连接,初期考虑就使用websocket,后来发现不能满足大批量用户同时连接,转投阿里爸爸的MQTT

前面的话

最近有需求要了解一下各个推送的协议，目前了解到实现推送的三个主要方式：MQTT、XMPP和Google
CloudMessage（GCM）。第三种方式暂不研究，前两种都要看一看，本篇讨论一下MQTT协议吧。本文使
用阿里云Ubuntu云服务器安装代理服务器，使用eclipsepaho实现的MqttClient编写代码。文中的所使
用的账户名和密码在本文发布后将会更改，请各位自行搭建环境。本文包括以下内容：

  • MQTT简介
  • MQTT优势
  • MQTT开发环境搭建
  • 使用PAHO实现MQTT推送

MQTT简介 & MQTT优势

MQTT全称是Message Queuing Telemetry
Transport，MQTT是IBM开发的基于TCP/IP协议的轻量级通讯协议。MQTT是一个客户端服务端架构的发
布-订阅（publish-subscribe）的消息传输协议。它的设计思想是开放、简单、轻量、易于实现。这
些特点使它适用于受限环境。例如，但不仅限于：

  • 网络代价昂贵，带宽低、不可靠
  • 在嵌入式设备中运行，处理器和内存资源有限

MQTT控制报文头部仅有2字节的长度，降低了网络
传输所需要的流量。MQTT支持三种不同级别的服务
质量（Quality of
Service，QoS）为不同场景提供消息可靠性：

  • 级别0：尽力而为。消息发送者会想尽办法发送消息，但是遇到意外并不会重试。
  • 级别1：至少一次。消息接受者如果没有知会或者知会本身丢失，消息发送者会在此发送以保证消息接收者至少会收到一次，当然可能造成重复消息。
  • 级别2：恰好一次。保证这种语义肯定会减少并发或者增加延时，不过丢失或者重复消息是不可接受的时候，级别2是最合适的。

如果各位读完了这些仍然觉得不过瘾，没有戳中各位的痛点，可以去读一下MQTT的协议规范，这里中英文版本都有挑自己爱看的读一下就好。

使用PAHO实现MQTT推送

server端

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

public class MqttServer {
/**
 * 代理服务器ip地址
 */
public static final String MQTT_BROKER_HOST = "tcp://xiasuhuei321.com:61613";

/**
 * 订阅标识
 */
public static final String MQTT_TOPIC = "test";

private static String userName = "admin";
private static String password = "password";

/**
 * 客户端唯一标识
 */
public static final String MQTT_CLIENT_ID = "android_server_xiasuhuei321";
private static MqttTopic topic;
private static MqttClient client;

public static void main(String... args) {
    // 推送消息
    MqttMessage message = new MqttMessage();
    try {
        client = new MqttClient(MQTT_BROKER_HOST, MQTT_CLIENT_ID, new MemoryPersistence());
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(true);
        options.setUserName(userName);
        options.setPassword(password.toCharArray());
        options.setConnectionTimeout(10);
        options.setKeepAliveInterval(20);

        topic = client.getTopic(MQTT_TOPIC);

        message.setQos(1);
        message.setRetained(false);
        message.setPayload("message from server".getBytes());
        client.connect(options);

        while (true) {
            MqttDeliveryToken token = topic.publish(message);
            token.waitForCompletion();
            System.out.println("已经发
            送");
            Thread.sleep(10000);
        }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

这里的逻辑非常简单，创建一个MqttClient，每十秒发送一次消息，订阅了相应topic的客户端将会收到这个消息。接下来编写客户端：

client端

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

public class MyMqttClient {
/**
 * 代理服务器ip地址
 */
public static final String MQTT_BROKER_HOST = "tcp://xiasuhuei321.com:61613";

/**
 * 客户端唯一标识
 */
public static final String MQTT_CLIENT_ID = "android_xiasuhuei321";

/**
 * 订阅标识
 */
public static final String MQTT_TOPIC = "xiasuhuei321";

/**
 *
 */
public static final String USERNAME = "admin";
/**
 * 密码
 */
public static final String PASSWORD = "password";

private volatile static MqttClient mqttClient;
private static MqttConnectOptions options;

public static void main(String... args) {
    try {
        // host为主机名，clientid即连接MQTT的客户端ID，一般以客户端唯一标识符表示，
        // MemoryPersistence设置clientid的保存形式，默认为以内存保存
        // 设备id不要太骚气！！！！！！！
        mqttClient = new MqttClient(MQTT_BROKER_HOST, MQTT_CLIENT_ID, new MemoryPersistence());
        // 配置参数信息
        options = new MqttConnectOptions();
        // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，
        // 这里设置为true表示每次连接到服务器都以新的身份连接
        options.setCleanSession(true);
        // 设置用户名
        options.setUserName(USERNAME);
        // 设置密码
        options.setPassword(PASSWORD.toCharArray());
        // 设置超时时间 单位为秒
        options.setConnectionTimeout(10);
        // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
        options.setKeepAliveInterval(20);
        // 连接
        mqttClient.connect(options);
        // 订阅
        mqttClient.subscribe("test");
        // 设置回调
        mqttClient.setCallback(new MqttCallback() {
            @Override
            public void connectionLost(Throwable throwable) {
                System.out.println("connectionLost");
            }

            @Override
            public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
                System.out.println("Topic: " + s + " Message: " + mqttMessage.toString());
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
            }
        });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

踩坑

频繁重连

  • 确保CLIENT_ID是唯一的
  • 检查是否是多线程影响,因为我的项目中,MQTT运行在服务中,因为是长连接,所以需要开启子线程
  • 销毁时,断开连接要使用disconnect(),重新连接使用reconnect()

client端接收消息

server端发送的string类型消息,client端的接收处理

@Override public void messageArrived(String topic, MqttMessage message) throws Exception {
    Logger.d("MQTT收到消息了:" + new String(message.getPayload()));
    }

接收消息后执行操作

messageArrived()中接收到指令,需要回到主线程去操作,一般使用Handler去做,而不是直接去操作,否则有时候会出现一操作就connectLost()了.

换ip重连

有时候需要连接到其它的mqtt服务器,这个时候就会需要进行重连.
重连用什么形式无所谓,注意调用一个方法,就是disconnect(),先断开跟之前服务器的连接.
因为客户端是不能主动断的,只能通知服务端,让服务端来断你.
另外注意先加个判断,如果已经断开了,就不要再断开了.

耗电量

mqtt在几种常见的长连接方式中,属于轻量级的,所以耗电量相对是最低的,项目初期,心跳保活的间隔是90s, api: setKeepAliveInterval(long m) ; 由于我们的项目是常开的, 这个间隔,一天下来,耗电量也是挺惊人的. 后来查了微信的心跳是300s, 我也调整到了300s,测下来功能也还是正常的,耗电量也下去了.

-------------看啥呢?没了-------------