0
0

SpringBoot集成MQTT服务器和客户端并通过WebSocket将消息实时发送到前端（本地搭建MQTT服务器）

前言：

  1.     在写这篇文章之前，我其实是碰到这个需求的，因为当时我需要这个MQTT服务器功能集成到我的程序中，而不是去用第三方公共的，但是我查阅了大量文章却都没有太大的收获，可能是我搜索的关键字不对，现在自己弄懂了大概之后就顺便做个记录。有错误的地方欢迎大家指出！
  3.     这里就只讲解与之相关的内容了，相关概念可以去看MQTT协议入门、websocket详情
  5.     关于mqtt的代码我是参考的mqtt参考

如何搭建MQTT本地服务器：

1. 要在pom.xml中添加相关maven第三方依赖至于Moquette是什么大家可以自行去查阅
  1. <!-- Moquette是 MQTT Broker实例库-->
  2.         <dependency>
  3.             <groupId>io.moquette</groupId>
  4.             <artifactId>moquette-broker</artifactId>
  5.             <version>0.12.1</version>
  6.             <!-- 这里是因为我的其他配置信息里面好像跟这个冲突了，所以没要它-->            
  7.             <exclusions>
  8.                 <exclusion>
  9.                     <groupId>ch.qos.reload4j</groupId>
  10.                     <artifactId>reload4j</artifactId>
  11.                 </exclusion>
  12.             </exclusions>
  13.         </dependency>
2. 添加mqtt的第三方依赖
  1. <!--添加mqtt依赖-->
  2. <dependency>
  3.     <groupId>org.eclipse.paho</groupId>
  4.     <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
  5.     <version>1.2.5</version>
  6. </dependency>
3. mqtt服务器及其客户端相关配置信息代码
  1. import io.moquette.broker.Server;
  2. import io.moquette.broker.config.MemoryConfig;
  3. import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
  4. import org.springframework.context.annotation.Bean;
  5. import org.springframework.context.annotation.Configuration;
  7. import java.util.Properties;
  9. @Configuration
  10. public class MqttAndBrokerConfig {
  12.     // Mqtt服务器
  13.     @Bean
  14.     public Server MqttBrokerServer() throws Exception{
  15.         // MQTT broker 服务器
  16.         Server mqttBroker = new Server();
  17.         Properties configProps = new Properties();
  18.         configProps.setProperty("port","1883");
  19.         configProps.setProperty("host","127.0.0.1");
  20.         configProps.setProperty("websocket_port","8087");//设置 WebSocket端口，用于支持 WebSocket连接
  21.         try {
  22.             // 启动服务器
  23.             mqttBroker.startServer(new MemoryConfig(configProps));
  24.             System.out.println("Moquette MQTT Broker started. Press Ctrl+C to shutdown.");
  25.             // 程序关闭时执行的操作
  26.             Runtime.getRuntime().addShutdownHook(new Thread(() -> {
  27.                 System.out.println("Stopping Moquette MQTT Broker...");
  28.                 mqttBroker.stopServer();
  29.                 System.out.println("Moquette MQTT Broker Stopped.");
  30.             }));
  31.         }catch (Exception e){
  32.             // 捕获并输出异常信息。
  33.             e.printStackTrace();
  34.         }
  35.         return mqttBroker;
  36.     }
  38.     //这个可以根据自己的需要进行相关的配置，因为我是需求简单才这样写的
  39.     //Mqtt客户端的option配置
  40.     @Bean
  41.     public MqttConnectOptions mqttConnectOptions(){
  42.         MqttConnectOptions options = new MqttConnectOptions();
  43.         options.setKeepAliveInterval(60);//设置保活时间
  44.         options.setConnectionTimeout(30);//设置连接超时时间
  45.         options.setUserName("mqtt_demo");
  46.         options.setPassword("123456".toCharArray());
  47.         options.setCleanSession(true);//配置持久会话。
  48.         //true:表示创建一个新的会话，在客户端断开连接时，会话将自动销毁.
  49.         //false:创建一个持久会话，在客户端断开连接后会话仍然保持，直到会话超时注销.
  50.         return options;
  51.     }
  52. }

实现mqtt的发布/订阅模式

1. mqtt controller实现层
  1.     注意：到这里的时候已经可以启动相关程序，并且可以调用其方法。我还是讲一下吧。
  3.     init()：在执行方法之前，分别创建发布者/订阅者的客户端，并连接服务器。其中我还设置了当程序启动时自动订阅一个topic（这个可以根据自己的需要来进行设置）
  5.     cleanup()：在关闭程序时执行，即容器销毁bean时调用。
  7.     publisher方法：发布者发布信息时调用的方法。
  9.     subscriber方法：订阅者订阅的消息的方法（这个可以不用像我这样添加，因为订阅者是不需要做出操作就可以获取信息的，我只是顺便写了），这个方法也可以直接放在publisher方法中直接使用。
  11.     注意：相关类在文章最下面
  1. import com.example.intelligentparking.pojo.Result;
  2. import jakarta.annotation.PostConstruct;
  3. import jakarta.annotation.PreDestroy;
  4. import org.eclipse.paho.client.mqttv3.*;
  5. import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
  6. import org.springframework.beans.factory.annotation.Autowired;
  7. import org.springframework.context.annotation.DependsOn;
  8. import org.springframework.web.bind.annotation.*;
  10. import java.util.UUID;
  12. @RestController
  13. @RequestMapping("/mqtt")
  14. @DependsOn("MqttBrokerServer")//确保MqttBrokerServer在controller层之前被初始化
  15. public class MqttController {
  16.     private MqttClient publisherClient;
  17.     private MqttClient subscriberClient;
  18.     private final String brokerUrl = "tcp://localhost:1883";
  19.     private int qos = 1;
  20.     private String topic = "mqtt/demo";
  22.     @Autowired
  23.     private MqttConnectOptions options;
  25.     @PostConstruct //在类的依赖项被注入后调用，但在类被使用之前执行
  26.     public void init(){
  27.         try {
  28.             // 发布者
  29.             // 创建 MQTT 客户端
  30.             publisherClient = new MqttClient(brokerUrl, "publisher_" + UUID.randomUUID().toString(), new MemoryPersistence());
  31.             //new MemoryPersistence()使用内存持久化存储。
  32.             // 设置连接配置
  33.             publisherClient.connect(options);
  34.             System.out.println("发布者客户端连接成功");
  36.             // 订阅者
  37.             subscriberClient = new MqttClient(brokerUrl, "subscriber_" + UUID.randomUUID().toString(), new MemoryPersistence());
  38.             subscriberClient.connect(options);
  39.             System.out.println("订阅者客户端连接成功");
  41.             // 在应用启动时订阅主题
  42.             subscriberClient.subscribe(topic, qos);
  43.             System.out.println("Subscribed to topic: " + topic );
  45.             // 设置回调函数
  46.             subscriberClient.setCallback(new MqttCallback() {
  47.                 // 连接丢失时调用
  48.                 @Override
  49.                 public void connectionLost(Throwable throwable) {
  50.                     System.out.println("connectionLost: " + throwable.getMessage());
  51.                 }
  53.                 // 接收到消息时被调用
  54.                 @Override
  55.                 public void messageArrived(String topic, MqttMessage message) throws Exception {
  56.                     String messageContent = new String(message.getPayload());
  57.                     System.out.println("我是订阅消息的！");
  58.                     System.out.println("topic: " + topic);
  59.                     System.out.println("Qos: " + message.getQos());
  60.                     System.out.println("message content: " + messageContent);
  62.                   
  63.                 }
  65.                 // 消息发送完成时被调用
  66.                 @Override
  67.                 public void deliveryComplete(IMqttDeliveryToken token) {
  68.                     System.out.println("deliveryComplete---------" + token.isComplete());
  69.                 }
  70.             });
  72.         } catch (MqttException e) {
  73.             e.printStackTrace();
  74.         }
  75.     }
  77.     @PostMapping("/publish")
  78.     public Result publisher(@RequestParam String topic, @RequestParam String content) throws MqttException {
  79.         // 创建消息(在 MQTT 协议中，消息的内容是以字节形式发送的,所以用.getBytes()方法)并设置 QoS
  80.         MqttMessage message = new MqttMessage(content.getBytes());
  81.         message.setQos(1);
  82.         // 发布消息
  83.         publisherClient.publish(topic, message);
  84.         System.out.println("我是发布消息的！");
  85.         System.out.println("Message published");
  86.         System.out.println("topic: " + topic);
  87.         System.out.println("message content: " + content);
  88.         return Result.success("发布成功");
  89.     }
  91.     @PostMapping("/subscribe")
  92.     public Result subscriber(@RequestParam String topic) throws MqttException {
  93.         subscriberClient.subscribe(topic, qos);
  94.         System.out.println("Subscribed to topic: " + topic);
  95.         return Result.success("订阅成功");
  96.     }
  98.     @PreDestroy  //Spring 容器销毁 bean 时调用
  99.     public void cleanup() {
  100.         try {
  101.             if (publisherClient != null && publisherClient.isConnected()) {
  102.                 publisherClient.disconnect();
  103.                 publisherClient.close();
  104.                 System.out.println("Publisher client disconnected and closed.");
  105.             }
  106.             if (subscriberClient != null && subscriberClient.isConnected()) {
  107.                 // 关闭连接
  108.                 subscriberClient.disconnect();
  109.                 // 关闭客户端
  110.                 subscriberClient.close();
  111.                 System.out.println("MQTT client disconnected and closed.");
  112.             }
  113.         } catch (MqttException e) {
  114.             e.printStackTrace();
  115.         }
  116.     }
  117. }

websocket相关配置信息

1. 添加相关依赖
  1. <!--引入 WebSocket 依赖-->
  2.         <dependency>
  3.             <groupId>org.springframework.boot</groupId>
  4.             <artifactId>spring-boot-starter-websocket</artifactId>
  5.         </dependency>
2. 配置websocket
  1. import com.example.intelligentparking.service.MqttWebSocketHandler;
  2. import org.springframework.context.annotation.Configuration;
  3. import org.springframework.web.socket.config.annotation.EnableWebSocket;
  4. import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
  5. import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry;
  7. @Configuration
  8. @EnableWebSocket
  9. public class WebSocketConfig implements WebSocketConfigurer {
  11.     @Override
  12.     public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
  13.         //在指定路径 /mqtt-websocket 上添加 MqttWebSocketHandler处理器。这个路径是客户端连接时使用的 WebSocket URL。
  14.         registry.addHandler(new MqttWebSocketHandler(), "/mqtt-websocket")
  15.                 //允许所有来源的客户端连接（CORS 设置），这对于开发环境通常是方便的，但在生产环境中应该指定允许的来源以提高安全性
  16.                 .setAllowedOrigins("*");
  17.     }
  18. }
3. websocket处理器
  1. import org.springframework.web.socket.CloseStatus;
  2. import org.springframework.web.socket.TextMessage;
  3. import org.springframework.web.socket.WebSocketSession;
  4. import org.springframework.web.socket.handler.TextWebSocketHandler;
  6. import java.io.IOException;
  7. import java.util.concurrent.CopyOnWriteArrayList;
  9. public class MqttWebSocketHandler extends TextWebSocketHandler {
  10. //声明了一个名为 sessions 的变量，用于存储所有当前活跃的 WebSocket连接（即客户端会话），并使用线程安全的 CopyOnWriteArrayList 来确保在多线程环境中安全访问和修改这个列表。
  11.     private static final CopyOnWriteArrayList<WebSocketSession> sessions = new CopyOnWriteArrayList<>();
  13.     @Override
  14.     public void afterConnectionEstablished(WebSocketSession session) {
  15.         sessions.add(session);
  16.     }
  18.     @Override
  19.     public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
  20.         sessions.remove(session);
  21.     }
  23. //将指定的消息发送到所有连接的 WebSocket 客户端。
  24.     public void sendMessageToAll(String message) {
  25.         for (WebSocketSession session : sessions) {
  26.             try {
  27.                 session.sendMessage(new TextMessage(message));
  28.             } catch (IOException e) {
  29.                 e.printStackTrace();
  30.             }
  31.         }
  32.     }
  33. }

注意：现在这两个还不能直接传递消息，因为mqtt和websocket这两个算是跨域信息，所以需要配置CORS信息，以允许来自特定源或所有源的连接。

配置CORS信息:

  1. import org.springframework.context.annotation.Configuration;
  2. import org.springframework.web.servlet.config.annotation.CorsRegistry;
  3. import org.springframework.web.servlet.config.annotation.WebMvcConfigurer;
  5. @Configuration
  6. public class WebConfig implements WebMvcConfigurer {
  7.     @Override
  8.     public void addCorsMappings(CorsRegistry registry) {
  9.         registry.addMapping("/**") // 所有接口
  10.                 .allowCredentials(true) // 是否发送 Cookie
  11.                 .allowedOriginPatterns("*") // 支持域
  12.                 .allowedMethods("GET", "POST", "PUT", "DELETE") // 支持方法
  13.                 .allowedHeaders("*") // 表示所有请求头都被允许，客户端可以发送任何请求头到服务器。
  14.                 .exposedHeaders("*");// 表示所有响应头都将被暴露给客户端，客户端可以访问所有的响应头信息。
  15.     }
  16. }

注意：这个时候就可以实现信息跨域，但是前面的代码中需要实现将消息传递到前端

信息跨域实现：

MqttController类中添加

  1. import com.example.intelligentparking.service.MqttWebSocketHandler;
  2. import com.example.intelligentparking.pojo.MessageDTO;
  3. import com.fasterxml.jackson.databind.ObjectMapper;
  5. private final MqttWebSocketHandler webSocketHandler = new MqttWebSocketHandler();
  6. private ObjectMapper objectMapper = new ObjectMapper();
  8. // 接收到消息时被调用
  9.                 @Override
  10.                 public void messageArrived(String topic, MqttMessage message) throws Exception {
  11.                     String messageContent = new String(message.getPayload());
  12.                     System.out.println("我是订阅消息的！");
  13.                     System.out.println("topic: " + topic);
  14.                     System.out.println("Qos: " + message.getQos());
  15.                     System.out.println("message content: " + messageContent);
  17.                     // 将消息内容转换为 JSON（websocket中以json数据传递）
  18.                     MessageDTO messageDTO = new MessageDTO(messageContent);
  19.                     String jsonMessage = objectMapper.writeValueAsString(messageDTO);
  21.                     // 通过 WebSocket 发送消息到前端
  22.                     webSocketHandler.sendMessageToAll(jsonMessage);
  23.                 }

对应websocket前端的js代码

  1. // client.js
  2. const WebSocket = require('ws');
  4. // 创建 WebSocket 连接
  5. const socket = new WebSocket('ws://localhost:8086/mqtt-websocket');
  7. // 连接成功时的回调
  8. socket.on('open', () => {
  9.     console.log('WebSocket 连接已建立');
  10. });
  12. // 接收到消息时的回调
  13. socket.on('message', (res) => {
  14.     console.log('收到消息:', JSON.parse(res));
  15.     // 在这里可以进行消息验证或处理
  16. });
  18. // 连接关闭时的回调
  19. socket.on('close', () => {
  20.     console.log('WebSocket 连接已关闭');
  21. });
  23. // 连接错误时的回调
  24. socket.on('error', (error) => {
  25.     console.error('WebSocket 错误:', error);
  26. });

相关类：Result、MessageDTO

  1. package com.example.intelligentparking.pojo;
  3. import lombok.AllArgsConstructor;
  4. import lombok.Data;
  5. import lombok.NoArgsConstructor;
  7. // 统一响应结果
  8. @NoArgsConstructor//无参构造
  9. @AllArgsConstructor//全参构造
  10. @Data
  11. public class Result<T> {
  13.     private Integer code; //业务状态码   0-成功    1-失败
  14.     private String message; //提示信息
  15.     private T data; //响应数据
  17.     //返回操作成功响应结果（带参数）
  18.     public static <E> Result<E> success(E data){
  19.         return new Result<>(0,"操作成功",data);
  20.     }
  22.     //返回操作成功响应结果
  23.     public static Result success(){
  24.         return new Result(0,"操作成功",null);
  25.     }
  27.     //返回操作失败的响应结果
  28.     public static Result error(String message){
  29.         return new Result(  1,message,null);
  30.     }
  31. }
  1. package com.example.intelligentparking.pojo;
  3. public class MessageDTO {
  4.     private String content;
  6.     public MessageDTO(String content) {
  7.         this.content = content;
  8.     }
  10.     public String getContent() {
  11.         return content;
  12.     }
  14.     public void setContent(String content) {
  15.         this.content = content;
  16.     }
  17. }

综上就可以实现title的功能了

标签： websocket 运维 服务器
本文转载自: https://blog.csdn.net/m0_74773277/article/details/143337779
版权归原作者 緋儚 所有， 如有侵权，请联系我们删除。
登录后发布评论

“SpringBoot集成MQTT服务器和客户端并通过WebSocket将消息实时发送到前端（本地搭建MQTT服务器）”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？

文章导航
没有找到文章内容分段。。。