[物联网]Java如何接入MQTT协议

MQTT协议简介

MQTT（Message Queuing Telemetry Transport，消息队列遥测传输协议），是一种基于发布/订阅（publish/subscribe）模式的"轻量级"通讯协议，该协议构建于TCP/IP协议上，由IBM在1999年发布。MQTT最大优点在于，可以以极少的代码和有限的带宽，为连接远程设备提供实时可靠的消息服务。作为一种低开销、低带宽占用的即时通讯协议，使其在物联网、小型设备、移动应用等方面有较广泛的应用。

MQTT是一个基于客户端-服务器的消息发布/订阅传输协议。MQTT协议是轻量、简单、开放和易于实现的，这些特点使它适用范围非常广泛。在很多情况下，包括受限的环境中，如：机器与机器（M2M）通信和物联网（IoT）。其在，通过卫星链路通信传感器、偶尔拨号的医疗设备、智能家居、及一些小型化设备中已广泛使用。

Java如何对接MQTT

这里采用EMQX软件进行接入

官方文档https://www.emqx.io/docs/zh/latest/

1. 首先在MQTT中新建用户信息

        String data = "appId".concat(":").concat("appSecret");
        String authorization = "Basic " + Base64.encode(data, Charset.forName("UTF-8"));
        Map headers = new LinkedHashMap();
        headers.put("authorization", authorization);
        Map params = new LinkedHashMap();
        params.put("username", "用户名");
        params.put("password", "密码");
        String res = HttpUtils.doPostJson("EMQX接口地址" + "/api/v4/auth_username", 

1. 创建用户完成后添加Topic权限

  Map params = new LinkedHashMap();
        params.put("username",  "用户名");
        params.put("topic", "topic名称");
        params.put("action", "pub");
        params.put("access", "allow");
        res = HttpUtils.doPostJson("EMQX接口地址" + "/api/v4/acl", JacksonUtils.objectToJson(params), headers);
        resMap = JacksonUtils.jsonToMap(res);

添加完成后可以使用用户该用户链接 测试添加是否成功

Java对接MQTT代码

1. 引用依赖

  org.eclipse.paho
    org.eclipse.paho.client.mqttv3
    1.2.2

1. 监听代码

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
public class EmqxApp {
    public static void main(String[] args) {
        String subTopic = "topic";
        String broker = "地址";
        String clientId = "客户端id";
        String username = "用户名";
        String password = "密码";
        MemoryPersistence persistence = new MemoryPersistence();
        try {
            MqttClient client = new MqttClient(broker, clientId, persistence);
            // MQTT 连接选项
            MqttConnectOptions connOpts = new MqttConnectOptions();
            connOpts.setUserName(username);
            connOpts.setPassword(password.toCharArray());
            // 保留会话
            connOpts.setCleanSession(true);
            // 设置回调
            client.setCallback(new EmqxOnMessageCallback());
            // 建立连接
            System.out.println("Connecting to broker: " + broker);
            client.connect(connOpts);
            System.out.println("Connected");
            // 订阅
            client.subscribe(subTopic);
        } 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 org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
public class EmqxOnMessageCallback implements MqttCallback {
    public void connectionLost(Throwable cause) {
        // 连接丢失后，一般在这里面进行重连
        System.out.println("连接断开，可以做重连");
    }
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        // subscribe后得到的消息会执行到这里面
        System.out.println("接收消息主题:" + topic);
        System.out.println("接收消息Qos:" + message.getQos());
        System.out.println("接收消息内容:" + new String(message.getPayload()));
    }
    public void deliveryComplete(IMqttDeliveryToken token) {
        System.out.println("deliveryComplete---------" + token.isComplete());
    }
}

畅联AIoT开放云平台(www.24hlink.cn)由杭州美畅物联技术有限公司精心打造，不仅能够接入视频，也可以接入、管理各种IoT设备、工业现场设备，在AIoT开发领域能给合作伙伴带来难以置信的的降本增效价值。