MQTT入门教程：轻松订阅消息

2023-07-12 04:47:54

MQTT：轻量级消息传输协议

在机器与机器通信和物联网领域，MQTT（消息队列遥测传输）协议以其轻量、开放和简单而备受推崇。它采用发布/订阅模式，使设备能够轻松可靠地交换信息。本文将深入探讨 MQTT 的基本概念、Springboot 集成和常见的常见问题解答，帮助您充分利用这一强大的消息传输协议。

MQTT 的基本概念

MQTT 遵循一种称为发布/订阅的通信方式。发布者 将消息发布到指定主题，而订阅者 订阅该主题，以便接收发布的任何消息。每个主题标识一个特定的消息类别，允许订阅者过滤出对其应用程序相关的信息。MQTT 默认采用广播模型，即所有订阅相同主题的订阅者都会收到发布的消息。

Springboot 集成 MQTT

Springboot 框架提供了对 MQTT 客户端的内置支持，使开发人员能够轻松地将其集成到他们的应用程序中。以下是 Springboot 集成 MQTT 的步骤：

1. 引入依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-mqtt</artifactId>
</dependency>

2. 配置 MQTT 客户端

在 application.yaml 文件中指定 MQTT 服务器的连接信息：

spring:
  mqtt:
    host: localhost  # MQTT 服务器地址
    port: 1883      # MQTT 服务器端口
    username: admin  # MQTT 服务器用户名
    password: password # MQTT 服务器密码

3. 创建 MQTT 客户端实例

@SpringBootApplication
public class MqttApplication {

    public static void main(String[] args) {
        SpringApplication.run(MqttApplication.class, args);
    }

    @Bean
    public MqttConnectOptions mqttConnectOptions() {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName("admin");
        options.setPassword("password".toCharArray());
        options.setServerURIs(new String[] { "tcp://localhost:1883" });
        return options;
    }

    @Bean
    public MqttClient mqttClient() {
        MqttClient client = new MqttClient("tcp://localhost:1883", "ClientId", new MemoryPersistence());
        client.connect(mqttConnectOptions());
        return client;
    }
}

4. 订阅主题

订阅所需的主题，以便从发布者接收消息：

@SpringBootApplication
public class MqttApplication {

    public static void main(String[] args) {
        SpringApplication.run(MqttApplication.class, args);
    }

    @Bean
    public MqttConnectOptions mqttConnectOptions() {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName("admin");
        options.setPassword("password".toCharArray());
        options.setServerURIs(new String[] { "tcp://localhost:1883" });
        return options;
    }

    @Bean
    public MqttClient mqttClient() {
        MqttClient client = new MqttClient("tcp://localhost:1883", "ClientId", new MemoryPersistence());
        client.connect(mqttConnectOptions());
        client.subscribe("test/topic", 1);
        return client;
    }
}

5. 接收消息

使用回调处理接收到的消息：

@SpringBootApplication
public class MqttApplication {

    public static void main(String[] args) {
        SpringApplication.run(MqttApplication.class, args);
    }

    @Bean
    public MqttConnectOptions mqttConnectOptions() {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName("admin");
        options.setPassword("password".toCharArray());
        options.setServerURIs(new String[] { "tcp://localhost:1883" });
        return options;
    }

    @Bean
    public MqttClient mqttClient() {
        MqttClient client = new MqttClient("tcp://localhost:1883", "ClientId", new MemoryPersistence());
        client.connect(mqttConnectOptions());
        client.subscribe("test/topic", 1);
        client.setCallback(new MqttCallback() {
            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {
                System.out.println("收到消息：" + new String(message.getPayload()));
            }

            @Override
            public void connectionLost(Throwable cause) {
                System.out.println("连接丢失：" + cause.getMessage());
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {
                System.out.println("消息发送成功：" + token.getMessageId());
            }
        });
        return client;
    }
}