通过工厂模式实现SpringBoot+MQTT-订阅与消费

淡淡的烟草味﹌ 2024-03-26 16:56 39阅读 0赞

## 引言 ##

Spring Boot 是一款用于构建基于 Spring 框架的快速应用程序的开源框架。它的目标是简化 Spring 应用程序的开发和部署过程，Spring Boot 通过提供一些默认配置和自动配置，可以让开发者更快的创建一个独立的、产品级别的 Spring 应用程序。

MQTT 是一种轻量级的消息传输协议，它被设计用于 IoT（物联网）中的低带宽和不可靠的网络环境中。MQTT 协议具有简单、开销小、可扩展等特点，非常适用于物联网场景。

本文将介绍如何使用 Spring Boot 和 MQTT 实现订阅与消费的功能，并通过工厂模式进行实现。

## 1. 准备工作 ##

在开始本文的实现过程之前，我们需要准备一些环境和工具。

**1.1 Spring Boot**

首先，我们需要安装和配置 Spring Boot。在这里，我们将使用 Spring Initializr 工具来快速创建一个 Spring Boot 项目。您可以访问 https://start.spring.io/ 创建一个新的 Spring Boot 项目，也可以使用您自己的 IDE 或构建工具创建项目。

![04875ad7c2684b6aa5c6f5a5eed51874.png][]

**1.2 MQTT**

接下来，我们需要使用 MQTT 客户端用来连接和通信 MQTT 服务器。在这里，我们将使用 Eclipse Paho MQTT 客户端库。您可以访问 https://www.eclipse.org/paho/clients/java/ 下载 Eclipse Paho MQTT 客户端库。

![7ee97cae99b746499926992f3a5c6fb0.png][]

**1.3 MQTT 服务器**

最后，我们需要一个 MQTT 服务器。在这里，我们将使用 Eclipse Mosquitto 作为我们的 MQTT 服务器。您可以访问 https://mosquitto.org/ 下载和安装 Eclipse Mosquitto。

## 2.创建工厂类 ##

我们将使用工厂模式来创建 MQTT 连接和订阅的对象。我们将创建一个 MQTTFactory 类来管理 MQTT 连接和订阅的对象。

**2.1 创建 MQTTFactory 类**

首先，我们创建一个 MQTTFactory 类。这个类将会是一个单例类，它将会负责管理 MQTT 连接和订阅的对象。我们使用静态块来初始化 MQTT 连接。

public class MQTTFactory {
        private static final String MQTT_SERVER_URL = "tcp://localhost:1883";
        private static final String MQTT_CLIENT_ID = "mqtt-client";
        private static final String MQTT_TOPIC = "mqtt-topic";
    
        private static MqttClient mqttClient;
    
        static {
            try {
                mqttClient = new MqttClient(MQTT_SERVER_URL, MQTT_CLIENT_ID);
                mqttClient.connect();
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }
    
        private static MQTTFactory instance;
    
        private MQTTFactory() {}
    
        public static MQTTFactory getInstance() {
            if (instance == null) {
                instance = new MQTTFactory();
            }
            return instance;
        }
    }

在这个类中，我们定义了一些常量来设置 MQTT 服务器的地址、客户端 ID 和订阅的主题。在静态块中，我们使用 Eclipse Paho MQTT 客户端库创建了一个 MqttClient 对象，并连接到了 MQTT 服务器。在 getInstance() 方法中，我们使用懒汉式单例模式创建了一个 MQTTFactory 对象。

**2.2 创建 MQTTSubscriber 类**

接下来，我们创建一个 MQTTSubscriber 类来订阅 MQTT 主题。在 MQTTSubscriber 类中，我们定义了一个 subscribe() 方法，用于订阅 MQTT 主题。

public class MQTTSubscriber {
        private String topic;
        private MqttClient mqttClient;
    
        public MQTTSubscriber(String topic, MqttClient mqttClient) {
            this.topic = topic;
            this.mqttClient = mqttClient;
        }
    
        public void subscribe() throws MqttException {
            mqttClient.subscribe(topic, new IMqttMessageListener() {
                @Override
                public void messageArrived(String topic, MqttMessage message) throws Exception {
                    System.out.println(String.format("Received message: %s from topic: %s", new String(message.getPayload()), topic));
                }
            });
        }
    }

在这个类中，我们使用构造方法来传递订阅的主题和 MQTT 客户端对象。在 subscribe() 方法中，我们使用 Eclipse Paho MQTT 客户端库的 subscribe() 方法来订阅 MQTT 主题。在 messageArrived() 方法中，我们使用 System.out.println() 方法来打印收到的消息。

## 3.创建 Spring Boot 应用程序 ##

接下来，我们创建一个 Spring Boot 应用程序，来测试我们实现的 MQTT 订阅和消费的功能。

**3.1 创建 Spring Boot 应用程序**

我们可以使用 Spring Initializr 工具创建一个新的 Spring Boot 应用程序。在这里，我们创建一个名为 "mqtt-subscriber" 的应用程序。我们添加了如下的依赖：

*  spring-boot-starter-web
 *  eclipse-paho-mqtt-client

在 pom.xml 文件中，我们添加了如下的依赖：

<dependency>
        <groupId>org.eclipse.paho</groupId>
        <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
        <version>1.2.5</version>
    </dependency>

这个依赖用于引入 Eclipse Paho MQTT 客户端库。

**3.2 创建 Spring Boot 控制器**

我们创建一个名为 "MQTTSubscriberController" 的 Spring Boot 控制器。在这个控制器中，我们注入了一个 MQTTSubscriber 对象，并使用 @PostConstruct 注解来订阅 MQTT 主题。

@RestController
    public class MQTTSubscriberController {
        private final MQTTSubscriber mqttSubscriber;
    
        public MQTTSubscriberController(MQTTSubscriber mqttSubscriber) {
            this.mqttSubscriber = mqttSubscriber;
        }
    
        @PostConstruct
        public void init() {
            try {
                mqttSubscriber.subscribe();
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }
    }

在 init() 方法中，我们调用了 MQTTSubscriber 的 subscribe() 方法来订阅 MQTT 主题。

**3.3 测试应用程序**

现在，我们可以启动应用程序并测试 MQTT 订阅和消费的功能了。我们可以使用 Eclipse Mosquitto 的命令行客户端来发布消息到订阅的主题中。

在命令行中，我们可以使用以下命令来发布消息：

mosquitto_pub -t myTopic -m "Hello, World!"

在这个命令中，我们使用 "mosquitto\_pub" 命令行客户端来发布一条消息到 "myTopic" 主题中。消息内容为 "Hello, World!"。

如果一切正常，我们应该可以在控制台中看到如下的输出：

Received message: Hello, World! from topic: myTopic

这表明我们已经成功订阅并消费了 MQTT 主题中的消息。

## 4.总结 ##

在本文中，我们使用工厂模式实现了一个 MQTT 订阅和消费的示例应用程序。我们使用 Eclipse Paho MQTT 客户端库来连接 MQTT 服务器，并订阅指定的主题。我们还使用 Spring Boot 框架来创建应用程序，并使用 @PostConstruct 注解来订阅 MQTT 主题。最后，我们使用 Eclipse Mosquitto 的命令行客户端来发布消息到订阅的主题中，并在控制台中查看已消费的消息。

![07a6f58a24ac4de4b3d6f2b4f3a14bcc.png][]

## 5.进一步改进 ##

在实际的应用程序中，我们可能需要进一步改进我们的实现。以下是一些可能的改进方案：

**5.1 使用配置文件**

在我们的示例应用程序中，我们硬编码了 MQTT 服务器的地址、端口号、订阅主题等信息。在实际的应用程序中，我们可能希望将这些信息放到配置文件中，以便在部署应用程序时能够轻松地修改这些信息。

我们可以在 application.properties 或 application.yml 文件中添加以下内容：

mqtt.server.host=mqtt.example.com
    mqtt.server.port=1883
    mqtt.topic=myTopic

然后，我们可以在 MQTTFactory 类中读取这些配置信息：

@Configuration
    public class MQTTConfig {
        @Value("${mqtt.server.host}")
        private String mqttServerHost;
    
        @Value("${mqtt.server.port}")
        private int mqttServerPort;
    
        @Value("${mqtt.topic}")
        private String mqttTopic;
    
        @Bean
        public MqttClient mqttClient() throws MqttException {
            String brokerUrl = String.format("tcp://%s:%d", mqttServerHost, mqttServerPort);
            MqttClient mqttClient = new MqttClient(brokerUrl, MqttClient.generateClientId());
            mqttClient.connect();
            return mqttClient;
        }
    
        @Bean
        public MQTTSubscriber mqttSubscriber() throws MqttException {
            MqttClient mqttClient = mqttClient();
            return new MQTTSubscriber(mqttTopic, mqttClient);
        }
    
        @Bean
        public MQTTFactory mqttFactory() {
            return MQTTFactory.getInstance();
        }
    }

现在，我们不再需要在代码中硬编码 MQTT 服务器的地址、端口号、订阅主题等信息，而是从配置文件中读取这些信息。

**5.2 使用注解**

在我们的示例应用程序中，我们使用 @PostConstruct 注解来订阅 MQTT 主题。在实际的应用程序中，我们可能希望使用更具有语义化的注解来标识哪些方法是用于订阅 MQTT 主题的。

我们可以定义一个 @MQTTSubscriber 注解，并在需要订阅 MQTT 主题的方法上使用该注解：

@Retention(RetentionPolicy.RUNTIME)
    @Target(ElementType.METHOD)
    public @interface MQTTSubscriber {
    }

然后，在 MQTTSubscriberController 类中，我们可以使用 @MQTTSubscriber 注解来标识订阅 MQTT 主题的方法：

@RestController
    public class MQTTSubscriberController {
        private final MQTTSubscriber mqttSubscriber;
    
        public MQTTSubscriberController(MQTTSubscriber mqttSubscriber) {
            this.mqttSubscriber = mqttSubscriber;
        }
    
        @MQTTSubscriber
        public void onMessage(String topic, String message) {
            System.out.println(String.format("Received message: %s from topic: %s", message, topic));
        }
    }

在这个例子中，我们定义了一个名为 "onMessage" 的方法，并使用 @MQTTSubscriber 注解来标识该方法是用于订阅 MQTT 主题的。在这个方法中，我们使用两个参数来接收收到的消息的主题和内容。

在我们的应用程序启动时，我们可以使用 Spring AOP 框架来自动扫描 @MQTTSubscriber 注解，并将被注解的方法注册为 MQTT 主题的订阅者。我们可以在 MQTTSubscriberAspect 类中实现这个功能：

@Aspect
    @Component
    public class MQTTSubscriberAspect {
        private final MQTTSubscriber mqttSubscriber;
    
        public MQTTSubscriberAspect(MQTTSubscriber mqttSubscriber) {
            this.mqttSubscriber = mqttSubscriber;
        }
    
        @PostConstruct
        public void subscribe() throws MqttException {
            MqttClient mqttClient = mqttSubscriber.getMqttClient();
            mqttClient.subscribe(mqttSubscriber.getTopic(), (topic, message) -> {
                String payload = new String(message.getPayload(), StandardCharsets.UTF_8);
                Object bean = mqttSubscriber.getBean();
                Method[] methods = bean.getClass().getMethods();
                for (Method method : methods) {
                    if (method.isAnnotationPresent(MQTTSubscriber.class) &&
                            method.getParameterCount() == 2 &&
                            method.getParameterTypes()[0].equals(String.class) &&
                            method.getParameterTypes()[1].equals(String.class)) {
                        method.setAccessible(true);
                        try {
                            method.invoke(bean, topic, payload);
                        } catch (IllegalAccessException | InvocationTargetException e) {
                            e.printStackTrace();
                        }
                    }
                }
            });
        }
    }

在这个类中，我们使用 Spring AOP 框架的 @Aspect 注解来标识该类是一个切面类。在 @PostConstruct 方法中，我们获取 MQTT 订阅者的 MqttClient 实例，并使用该实例订阅指定的主题。然后，我们获取被注解的方法，并遍历这些方法，如果发现某个方法被标注为 @MQTTSubscriber，并且参数个数为 2，第一个参数为 String 类型，第二个参数也为 String 类型，则将该方法注册为订阅者。

现在，我们可以在任何需要订阅 MQTT 主题的方法上使用 @MQTTSubscriber 注解，而不必再使用 @PostConstruct 注解。这使得我们的代码更加具有语义化，并且更加容易阅读和维护。

## 6.结论 ##

在本文中，我们使用工厂模式实现了一个 MQTT 订阅和消费的示例应用程序。我们使用 Eclipse Paho MQTT 客户端库来连接 MQTT 服务器，并订阅指定的主题。我们还使用 Spring Boot 框架来创建应用程序，并使用 @PostConstruct 注解来订阅 MQTT 主题。最后，我们介绍了使用配置文件和注解来改进我们的实现的方法。这些改进方案使我们的代码更加具有语义化，并且更加容易阅读和维护。

MQTT 是一种轻量级的消息传递协议，适用于物联网等场景中的消息传递。它具有简单、可靠、高效等特点，已经被广泛应用于各种场景中。使用工厂模式和 Spring Boot 框架来实现 MQTT 订阅和消费，可以使我们的代码更加清晰和易于维护。

[04875ad7c2684b6aa5c6f5a5eed51874.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/87cdf8895ca9468b89dcd37c61a0eb0c.png
[7ee97cae99b746499926992f3a5c6fb0.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/251ebe83959344438aaecf1a0a6886ed.png
[07a6f58a24ac4de4b3d6f2b4f3a14bcc.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/98a72ae260904641929d2a721f6a5736.png