关于JAVA使用MQTT协议的那些事

站长

2024年07月14日 21:04 · 阅读数 25

MQTT协议是什么？

MQTT（消息队列遥测传输）是ISO标准（ISO/IEC PRF 20922）下基于发布/订阅范式的消息协议。它工作在TCP/IP协议族上，是为硬件性能低下的远程设备以及网络状况糟糕的情况下而设计的发布/订阅型消息协议。MQTT协议是轻量、简单、开放和易于实现的，这些特点使它适用范围非常广泛，尤其在受限的环境中，如机器与机器（M2M）通信和物联网（IoT）。

MQTT协议有什么特点？

开放消息协议，简单易实现。
发布订阅模式，一对多消息发布。
基于TCP/IP网络连接，提供有序、无损、双向连接。
1字节固定报头，2字节心跳报文，最小化传输开销和协议交换，有效减少网络流量。
消息QoS(服务质量)支持可靠传输保证。

MQTT协议的应用？

物联网数据采集及监控平台：MQTT可以用于从各种传感器和物联网设备（如温度传感器、湿度传感器、气压传感器、光照传感器等）收集数据，实时检测设备工作状况，汇总数据并进行可视化监控。
智能家居或智慧城市系统：MQTT可以用于家庭设备、家庭安防、门禁系统、电梯管理、智慧路灯等设备之间的通信和协调，实现智能家居或智慧城市系统。
物流及交通管理系统：MQTT可以用于在车辆、机器人和其他物理设备之间进行通信，实现物流及交通管理系统，为运营商或业务员提供更准确、快速和透明的信息。
环境监控：MQTT可以用于实现物联网设备之间的数据传输和信息互动，例如空气质量、温湿度、水质等的监测，可以对公共安全、环境保护等进行监管和控制。

总结：MQTT协议是一种轻量级、开放的消息传输协议，适用于各种资源受限的设备，尤其在物联网领域有着广泛的应用。

JAVA中如何使用MQTT协议与硬件进行交互？

导入相关库：
- 首先，需要在Java项目中导入支持MQTT协议的库。例如，可以使用Eclipse Paho库，这是一个广泛使用的MQTT客户端库。
理解MQTT角色：
- 在MQTT通讯过程中，有三种身份：发布者(publisher)、代理(broker)、和订阅者(subscriber)。
- 发布者负责发布消息到某个主题(topic)。
- 订阅者负责订阅一个或多个主题，以接收发布者发送的消息。
- 代理是MQTT服务器，负责消息的路由和分发。
配置MQTT连接：
- 设置MQTT Broker的基本连接参数，包括Broker地址、端口号、用户名和密码（如果需要的话）。
- 在Java代码中，可以创建一个MQTT客户端实例，并配置这些参数来连接到MQTT Broker。
编写发布和订阅逻辑：
- 对于与硬件交互的场景，可能需要编写一个Java程序作为发布者，用于将命令或数据发送到硬件。
- 同时，还可以编写另一个Java程序作为订阅者，用于接收硬件发送的状态信息或数据。
处理消息：
- 在发布者和订阅者的代码中，需要编写逻辑来处理MQTT消息。
- 对于发布者，可能需要将Java对象序列化为MQTT消息，然后发布到指定的主题。
- 对于订阅者，需要编写一个回调函数来处理接收到的MQTT消息，并将其解析为Java对象或进行其他处理。
错误处理和重连机制：
- 在与硬件进行交互时，网络波动和硬件故障是常见的问题。因此，在Java代码中实现适当的错误处理和重连机制是很重要的。

有哪些JAVA框架集成了MQTT协议？

Eclipse Paho：Paho是一个开源客户端库，支持多种编程语言，包括Java。它提供了MQTT 3.1和3.1.1协议的完整实现，并且可以在多种类型的设备和网络上运行。Paho库提供了MQTT客户端的API，包括连接、发布、订阅和接收消息等功能。
HiveMQ MQTT Client：HiveMQ是一个高性能的MQTT代理（Broker），但它也提供了一个Java MQTT客户端库。这个库提供了丰富的功能和灵活性，可以方便地集成到Java项目中。HiveMQ MQTT Client支持MQTT 5.0协议，并提供了异步API来提高性能和响应速度。
Spring Integration MQTT：Spring Integration是Spring框架中的一个子项目，它提供了一组用于构建企业集成解决方案的组件。Spring Integration MQTT是其中一个组件，它提供了与MQTT代理进行交互的能力。使用Spring Integration MQTT，可以轻松地将MQTT消息集成到的Spring应用程序中。
MQTT.fx：虽然MQTT.fx主要是一个MQTT客户端工具，但它也提供了一个基于Java的MQTT客户端库。这个库提供了用于连接、发布和订阅MQTT主题的基本功能。MQTT.fx的Java库可以作为一个轻量级的MQTT客户端解决方案。

总结：这些框架各有优缺点，可以根据的具体需求来选择适合的框架。例如，如果正在使用Spring框架构建应用程序，那么Spring Integration MQTT可能是一个不错的选择。如果需要一个轻量级的、易于使用的MQTT客户端库，那么Eclipse Paho或MQTT.fx的Java库可能是更好的选择。

如何使用Eclipse Paho框架写一个简单案例？

import org.eclipse.paho.client.mqttv3.*;  
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;  
  
public class MqttHardwareInteraction {  
  
    public static void main(String[] args) {  
        String brokerUrl = "tcp://broker.emqx.io:1883"; // MQTT Broker地址和端口  
        String clientId = "JavaHardwareClient"; // 客户端ID  
  
        try {  
            MqttClient client = new MqttClient(brokerUrl, clientId, new MemoryPersistence());  
            MqttConnectOptions connOpts = new MqttConnectOptions();  
            // 如果有需要，可以设置用户名和密码  
            // connOpts.setUserName("username");  
            // connOpts.setPassword("password".toCharArray());  
            client.connect(connOpts);  
  
            // 假设我们有一个硬件控制命令  
            String hardwareCommand = "turn_on_light";  
            MqttMessage message = new MqttMessage(hardwareCommand.getBytes());  
            message.setQos(2); // 设置QoS级别  
            client.publish("hardware/commands", message); // 发布命令到"hardware/commands"主题  
  
            // 可以添加订阅逻辑来接收硬件的响应或状态更新  
  
            client.disconnect(); // 断开连接  
            client.close(); // 关闭客户端  
        } catch (MqttException e) {  
            e.printStackTrace();  
        }  
    }  
}

*注：这个示例仅用于演示目的，并未包含完整的错误处理和重连逻辑。在实际应用中，需要根据具体需求进行相应的扩展和优化。

Spring框架中如何使用Spring Integration MQTT？

1. 添加依赖

首先，需要在的pom.xml（Maven项目）或build.gradle（Gradle项目）中添加Spring Integration MQTT的依赖。对于Maven，通常需要添加Spring Integration Core和Spring Integration MQTT的依赖。

<dependencies>  
    <!-- Spring Integration Core -->  
    <dependency>  
        <groupId>org.springframework.integration</groupId>  
        <artifactId>spring-integration-core</artifactId>  
        <version>Spring Integration版本</version>  
    </dependency>  
  
    <!-- Spring Integration MQTT -->  
    <dependency>  
        <groupId>org.springframework.integration</groupId>  
        <artifactId>spring-integration-mqtt</artifactId>  
        <version>Spring Integration MQTT版本</version>  
    </dependency>  
  
    <!-- 其他依赖... -->  
</dependencies>

2. 配置 MQTT 连接工厂

在Spring配置（例如，application.yml，application.properties，或者Java Config类）中，需要配置MQTT的入站（inbound）和出站（outbound）通道。这通常涉及到定义MQTT客户端工厂、消息通道、消息处理器等。

@Configuration  
@EnableIntegration  
public class MqttConfig {  
  
    @Value("${mqtt.broker.url}")  
    private String brokerUrl;  
  
    @Value("${mqtt.client.id}")  
    private String clientId;  
  
    // ... 其他配置属性  
  
    @Bean  
    public MqttPahoClientFactory mqttClientFactory() {  
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();  
        // 可以设置其他工厂级别的配置，如 SSL 配置等  
        return factory;  
    }  
  
    @Bean  
    public MessageChannel mqttOutboundChannel() {  
        return new DirectChannel();  
    }  
  
    @Bean  
    public MessageProducer inbound() {  
        MqttPahoMessageDrivenChannelAdapter adapter =  
                new MqttPahoMessageDrivenChannelAdapter(clientId(), mqttClientFactory(), "topic/in");  
        adapter.setCompletionTimeout(5000);  
        adapter.setConverter(new DefaultPahoMessageConverter());  
        adapter.setQos(1); 
        /**
        * QoS 0：最多传输一次
        * QoS 1：至少传输一次
        * QoS 2：恰好传输一次
        */
        adapter.setOutputChannel(mqttInputChannel());  
        return adapter;  
    }  
  
    @Bean  
    public MessageChannel mqttInputChannel() {  
        return new DirectChannel();  
    }  
  
    // ... 可能的 MQTT 发布者和其他配置  
  
    private String clientId() {  
        return clientId + "_" + UUID.randomUUID().toString();  
    }  
}

3. 编写消息处理器

编写消息处理器来处理从MQTT代理接收到的消息，或者准备并发送消息到MQTT代理。这可以通过在配置类中使用@ServiceActivator注解的方法来实现。

@Service  
public class MqttMessageHandler {  
  
    @ServiceActivator(inputChannel = "mqttInputChannel")  
    public void handleMessage(Message<?> message) {  
        // 处理从 MQTT 接收到的消息  
        System.out.println("Received MQTT message: " + message.getPayload());  
    }  
  
    // ... 可能的发送消息到 MQTT 的方法  
}

4. 发送MQTT消息

使用 MessagingGateway 或直接通过 MessageChannel 发送 MQTT 消息。

@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")  
public interface MqttGateway {  
  
    void sendToMqtt(String payload, @Header(MqttHeaders.TOPIC) String topic);  
}  
  
// 在服务或组件中注入并使用这个网关  
@Autowired  
private MqttGateway mqttGateway;  
  
// ...  
mqttGateway.sendToMqtt("Hello, MQTT!", "topic/out");

5. 注意事项

MQTT服务器：确保有一个MQTT服务器可以连接，并正确配置了连接选项（如主机名、端口、用户名和密码等）。
连接选项：在MqttPahoClientFactory中，可以设置MQTT的连接选项，如MqttConnectOptions，以控制连接的细节。
主题和客户端ID：确保订阅和发布的主题（topic）是正确的，并且每个MQTT客户端都有唯一的客户端ID（clientId）。
异常处理：在消息处理器中，确保处理可能的异常，如网络错误或MQTT特定的错误。
日志记录：为应用程序添加日志记录功能，以便在出现问题时可以轻松地跟踪和调试。
测试：在部署到生产环境之前，确保MQTT集成已经经过充分的测试，包括单元测试、集成测试和端到端测试。

结尾：我也是刚学习这块知识，有疑问的地方也是查询资料得来，肯定有说的不足或错误的地方，还望谅解。

转载自:https://juejin.cn/post/7366084203953553434