SpringBoot集成MQTT，WebSocket返回前端信息

时间：2025-04-02 21:51:25

MQTT(消息队列遥测传输)是ISO 标准(ISO/IEC PRF 20922)下基于发布/订阅范式的消息协议。它工作在 TCP/IP协议族上，是为硬件性能低下的远程设备以及网络状况糟糕的情况下而设计的发布/订阅型消息协议，为此，它需要一个消息中间件 。

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

直接上代码

pom文件

<!--websocket依赖-->
        <dependency>
            <groupId></groupId>
            <artifactId>spring-boot-starter-websocket</artifactId>
            <version>2.1.</version>
        </dependency>
<!--        mqtt-->
        <dependency>
            <groupId></groupId>
            <artifactId>spring-boot-starter-integration</artifactId>
        </dependency>
        <dependency>
            <groupId></groupId>
            <artifactId>spring-integration-stream</artifactId>
        </dependency>
        <dependency>
            <groupId></groupId>
            <artifactId>spring-integration-mqtt</artifactId>
        </dependency>

配置文件

#mqtt配置
com:
  mqtt:
    #连接地址
    url: tcp://127.0.0.1:7788
    #客户端id(不能重复)
    clientId: mqtt_test1234
    #订阅主题
    topics: test/#
    #MQTT用户名
    username: admin
    #MQTT密码
    password: 123456
    #超时时间
    timeout: 30
    #保持连接数
    keepalive: 20

mqtt的java配置文件

import ;
import ;

/**
 * @Author majinzhong
 * @Date 2023/2/7 15:08
 * @Version 1.0
 * 读取yml
 */
@Component
@ConfigurationProperties(prefix = "")     //对应yml文件中的com下的mqtt文件配置
public class MqttConfiguration {
    private String url;
    private String clientId;
    private String topics;
    private String username;
    private String password;
    private String timeout;
    private String keepalive;
    public String getUrl() {
        return url;
    }
    public void setUrl(String url) {
         = url;
    }
    public String getUsername() {
        return username;
    }
    public void setUsername(String username) {
         = username;
    }
    public String getPassword() {
        return password;
    }
    public void setPassword(String password) {
         = password;
    }
    public String getClientId() {
        return clientId;
    }
    public void setClientId(String clientId) {
         = clientId;
    }
    public String getTopics() {
        return topics;
    }
    public void setTopics(String topics) {
         = topics;
    }
    public String getTimeout() {
        return timeout;
    }
    public void setTimeout(String timeout) {
         = timeout;
    }
    public String getKeepalive() {
        return keepalive;
    }
    public void setKeepalive(String keepalive) {
         = keepalive;
    }
}
import ;
import .;
import ;
import ;
import ;
import ;
import ;
import ;
import ;
import ;
import ;
import ;
import ;
import ;
import ;
import ;
import ;
import ;

/**
 * @Author majinzhong
 * @Date 2023/2/7 15:09
 * @Version 1.0
 * MQTT消费端
 */
@Configuration
@IntegrationComponentScan
public class MqttInboundConfiguration {

    @Autowired
    private MqttConfiguration mqttProperties;

    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    @Bean
    public MqttPahoClientFactory mqttInClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        String[] array = ().split(",");
        MqttConnectOptions options = new MqttConnectOptions();
        (array);
        (());
        (().toCharArray());
        (2);
        //接受离线消息
        (false);
        (options);
        return factory;
    }

    //配置client,监听的topic
    @Bean
    public MessageProducer inbound() {
        String[] inboundTopics = ().split(",");
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
                ()+"_inbound",mqttInClientFactory(), inboundTopics);  //对inboundTopics主题进行监听
        (5000);
        (1);
        (new DefaultPahoMessageConverter());
        (mqttInputChannel());
        return adapter;
    }

    //通过通道获取数据
    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")  //异步处理
    public MessageHandler handler() {
        return new MessageHandler() {
            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                ("----------------------");
                //获取mqtt的topic
                String topic = (String) ().get(MqttHeaders.RECEIVED_TOPIC);
                //使用webSocket返回给前端
                SocketService socketService = new SocketService();
                (().toString(),null,topic);

                ("message:"+());
                ("PacketId:"+().getId());
                ("Qos:"+().get());
                ("topic:"+topic);
            }
        };
    }
}
import .;
import ;
import ;
import ;
import ;
import ;
import ;
import ;
import ;
import ;
import ;

/**
 * @Author majinzhong
 * @Date 2023/2/7 15:09
 * @Version 1.0
 * MQTT生产端
 */
@Configuration
public class MqttOutboundConfiguration {
    @Autowired
    private MqttConfiguration mqttProperties;

    @Bean
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }

    @Bean
    public MqttPahoClientFactory mqttOutClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        String[] array = ().split(",");
        MqttConnectOptions options = new MqttConnectOptions();
        (array);
        (());
        (().toCharArray());
        // 接受离线消息
        (false); //告诉代理客户端是否要建立持久会话   false为建立持久会话
        (options);
        return factory;
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
    public MessageHandler mqttOutbound() {
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(
                ()+"outbound", mqttOutClientFactory());
        (true);

        return messageHandler;
    }
}

websocket的java配置文件

import ;
import ;
import ;

/**
 * @authoer:majinzhong
 * @Date: 2022/11/7
 * @description:
 */
@Configuration
public class WebSocketConfig {
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }
}

MQTT的controller文件

import ;
import ;
import ;
import ;

/**
 * @Author majinzhong
 * @Date 2023/2/7 15:10
 * @Version 1.0
 */
@RestController
public class MqttPubController {

    @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
    @Autowired
    private MqttGateway mqttGateway;


    @RequestMapping("/hello")
    public String hello() {
        return "hello!";
    }

    @RequestMapping("/sendMqtt")
    public String sendMqtt(String  sendData){
        (sendData);
        ("进入sendMqtt-------"+sendData);
        ("topic01",(String) sendData);
        return "Test is OK";
    }



    @RequestMapping("/sendMqttTopic")
    public String sendMqtt(String  sendData,String topic){
        //(sendData+"   "+topic);
        //("进入inbound发送："+sendData);
        (topic,(String) sendData);
        return "Test is OK";
    }
}

MQTT的service文件

import ;
import ;
import ;
import ;

/**
 * @Author majinzhong
 * @Date 2023/2/7 15:21
 * @Version 1.0
 */
@Service
@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
public interface MqttGateway {
    void sendToMqtt(@Header() String topic01, String sendData);
}

websocket的sevice文件

import ;
import ;
import ;
import ;
import ;

import .*;
import ;
import ;
import ;
import ;
import ;
import ;

/**
 * @authoer:majinzhong
 * @Date: 2022/11/16
 * @description:
 */
@Component
@ServerEndpoint(value = "/socket/{nickname}")
@CrossOrigin
@Service
public class SocketService {

    /**
     * 用来存放每个客户端对应的MyWebSocket对象。
     **/
    private static CopyOnWriteArraySet<SocketService> socketSet = new CopyOnWriteArraySet<>();
    /**
     * 与某个客户端的连接会话，需要通过它来给客户端发送数据
     **/
    private Session session;
    /**
     * 用户名称
     **/
    private String nickname;
    /**
     * 用来记录sessionId和该session进行绑定
     **/
    private static Map<String,Session> map = new HashMap<String, Session>();

    /**
     * 连接建立成功调用的方法
     */
    @OnOpen
    public void onOpen(Session session,@PathParam("nickname") String nickname) {
         = session;
        =nickname;

        (nickname, session);
        (this);

        ("有新连接加入:"+nickname+",当前在线人数为" + ());
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose() {
        (this);
        List<String> nickname = ().get("nickname");
        for(String nick:nickname) {
            (nick);
        }
        ("有一连接关闭！当前在线人数为" + ());
    }

    /**
     * 收到客户端消息后调用的方法
     */
    @OnMessage
    public void onMessage(String message, Session session,@PathParam("nickname") String nickname) {
        ("来自客户端的消息-->"+nickname+": " + message);
        //将mqtt发送过来的数据返回给前端
        try {
            JSON parse = (message);
            Map<String,Object> valuesMap = (Map<String, Object>) ("values");
            String tag1 = ("tag1").toString();
            String replace = ("\r\n", "");
            String substring = ((":") + 1).trim().replace(" ",":");
            String tag2 = ("tag2").toString();
            String substring1=substring+"tag2:"+tag2;
            String result="{"+substring1+"}";
            //发送给前端，用作页面渲染
            Session fromSession = (nickname);
            ().sendText(result);
        } catch (Exception e) {
            ();
        }
    }

    /**
     * 发生错误时调用
     */
    @OnError
    public void onError(Session session, Throwable error) {
        ("发生错误");
        ();
    }

    /**
     * 群发自定义消息
     */
//    public void broadcast(String message) {
//        for (SocketService item : socketSet) {
//            /**
//             * 同步异步说明参考：/who_is_xiaoming/article/details/53287691
//             *
//             * ().sendText(message);
//             **/
//            ().sendText(message);
//        }
//    }
}

总结：

和MQTT是通过MQTT的topic进行区分消息的消费，所以在和前端进行联调的时候，需要的nickname需要是MQTT的topic，其中消息的整理在WebSocket里面

2.在调用MqttPubController里的方法时，需要正确的topic入参，不然在MQTTX里面是接收不到数据的

补充：文件里面的url可以是多个，使用逗号进行拼接，topics也可以是多个，使用逗号进行拼接，其中#是通配符，代表test下的所有

相关文章