SpringBoot实现MQTT消息发送和接收方式

SpringBoot MQTT消息 MQTT消息发送 MQTT消息接收 2023-03-11 11:03:33 597人浏览安东尼

Python 官方文档：入门教程 => 点击学习

摘要

目录spring integration交互逻辑1、Maven依赖2、yaml配置文件3、MQtt生产者消费者配置类4、消息处理类 5、mQtt发送接口 6、mq

Spring integration交互逻辑

对于发布者：

1.消息通过消息网关发送出去，由 MessageChannel 的实例 DirectChannel 处理发送的细节。

2.DirectChannel 收到消息后，内部通过 MessageHandler 的实例 MqttPahoMessageHandler 发送到指定的 Topic。

对于订阅者：

1.通过注入 MessageProducerSupport 的实例 MqttPahoMessageDrivenChannelAdapter，实现订阅 Topic 和绑定消息消费的 MessageChannel。

2.同样由 MessageChannel 的实例 DirectChannel 处理消费细节。

Channel 消息后会发送给我们自定义的 MqttInboundMessageHandler 实例进行消费。

可以看到整个处理的流程和前面将的基本一致。Spring Integration 就是抽象出了这么一套消息通信的机制，具体的通信细节由它集成的中间件来决定

1、maven依赖

<!-- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-integration -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-integration</artifactId>
    <version>2.5.1</version>
</dependency>
 
<!-- Https://mvnrepository.com/artifact/org.springframework.integration/spring-integration-stream -->
<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-stream</artifactId>
    <version>5.5.5</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.springframework.integration/spring-integration-mqtt -->
<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-mqtt</artifactId>
    <version>5.5.5</version>
</dependency>

2、yaml配置文件

#mqtt配置
mqtt:
  username: 123
  passWord: 123
  #MQTT-服务器连接地址，如果有多个，用逗号隔开
  url: tcp://127.0.0.1:1883
  #MQTT-连接服务器默认客户端ID
  client:
    id: ${random.value}
  default:
    #MQTT-默认的消息推送主题，实际可在调用接口时指定
    topic: topic,mqtt/test/#
    #连接超时
  completionTimeout: 3000

3、mqtt生产者消费者配置类

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
 
import java.util.Arrays;
import java.util.List;
 

@Configuration
@IntegrationComponentScan
@Slf4j
public class MqttSenderAndReceiveConfig {
 
    private static final byte[] WILL_DATA;
 
    static {
        WILL_DATA = "offline".getBytes();
    }
 
    @Autowired
    private MqttReceiveHandle mqttReceiveHandle;
 
    @Value("${mqtt.username}")
    private String username;
 
    @Value("${mqtt.password}")
    private String password;
 
    @Value("${mqtt.url}")
    private String hostUrl;
 
    @Value("${mqtt.client.id}")
    private String clientId;
 
    @Value("${mqtt.default.topic}")
    private String defaultTopic;
 
    @Value("${mqtt.completionTimeout}")
    private int completionTimeout;   //连接超时
 
    
    @Bean(value = "getMqttConnectOptions")
    public MqttConnectOptions getMqttConnectOptions1() {
        MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
        // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接
        mqttConnectOptions.setCleanSession(true);
        // 设置超时时间 单位为秒
        mqttConnectOptions.setConnectionTimeout(10);
        mqttConnectOptions.setAutomaticReconnect(true);
        mqttConnectOptions.setUserName(username);
        mqttConnectOptions.setPassword(password.toCharArray());
        mqttConnectOptions.setServerURIs(new String[]{hostUrl});
        // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送心跳判断客户端是否在线，但这个方法并没有重连的机制
        mqttConnectOptions.seTKEepAliveInterval(10);
        // 设置“遗嘱”消息的话题，若客户端与服务器之间的连接意外中断，服务器将发布客户端的“遗嘱”消息。
        //mqttConnectOptions.setWill("willTopic", WILL_DATA, 2, false);
        return mqttConnectOptions;
    }
 
    
    @Bean
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setConnectionOptions(getMqttConnectOptions1());
        return factory;
    }
 
    
    @Bean
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }
 
    
    @Bean
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
    public MessageHandler mqttOutbound() {
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(clientId + "_producer", mqttClientFactory());
        messageHandler.setAsync(true);
        messageHandler.setDefaultTopic(defaultTopic);
        messageHandler.setAsyncEvents(true); // 消息发送和传输完成会有异步的通知回调
        //设置转换器 发送bytes数据
        DefaultPahoMessageConverter converter = new DefaultPahoMessageConverter();
        converter.setPayloadAsBytes(true);
        return messageHandler;
    }
 
    
    @Bean
    public MessageProducer inbound() {
        List<String> topicList = Arrays.asList(defaultTopic.trim().split(","));
        String[] topics = new String[topicList.size()];
        topicList.toArray(topics);
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(clientId + "_consumer", mqttClientFactory(), topics);
        adapter.setCompletionTimeout(completionTimeout);
        DefaultPahoMessageConverter converter = new DefaultPahoMessageConverter();
        converter.setPayloadAsBytes(true);
        adapter.setConverter(converter);
        adapter.setQos(2);
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }
 
    
    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }
 
    
    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler handler() {
        return new MessageHandler() {
            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                //处理接收消息
                mqttReceiveHandle.handle(message);
            }
        };
    }
}

4、消息处理类


@Slf4j
@Component
public class MqttReceiveHandle {
 
    public void handle(Message<?> message) {
        log.info("收到订阅消息: {}", message);
        String topic = message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC).toString();
        log.info("消息主题：{}", topic);
        Object payLoad = message.getPayload();
        byte[] data = (byte[]) payLoad;
        Packet packet = Packet.parse(data);
        log.info("发送的Packet数据{}", JSON.tojsONString(packet));
 
    }
}

5、mqtt发送接口

import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.handler.annotation.Header;
 

@MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
public interface MqttGateway {
 
    
    void sendToMqttObject(@Header(MqttHeaders.TOPIC) String topic, byte[] payload);
 
    
    void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, String payload);
 
    
    void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, String payload);
 
    
    void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, Object payload);
 
    
    void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic, byte[] payload);
}

6、mqtt事件监听类

import lombok.extern.slf4j.Slf4j;
import org.springframework.context.event.EventListener;
import org.springframework.integration.mqtt.event.MqttConnectionFailedEvent;
import org.springframework.integration.mqtt.event.MqttMessageDeliveredEvent;
import org.springframework.integration.mqtt.event.MqttMessageSentEvent;
import org.springframework.integration.mqtt.event.MqttSubscribedEvent;
import org.springframework.stereotype.Component;
 
@Slf4j
@Component
public class MqttListener {
    
    @EventListener(classes = MqttConnectionFailedEvent.class)
    public void listenerAction(MqttConnectionFailedEvent mqttConnectionFailedEvent) {
        log.info("连接失败的事件通知");
    }
 
    
    @EventListener(classes = MqttMessageSentEvent.class)
    public void listenerAction(MqttMessageSentEvent mqttMessageSentEvent) {
        log.info("已发送的事件通知");
    }
 
    
    @EventListener(classes = MqttMessageDeliveredEvent.class)
    public void listenerAction(MqttMessageDeliveredEvent mqttMessageDeliveredEvent) {
        log.info("已传输完成的事件通知");
    }
 
    
    @EventListener(classes = MqttSubscribedEvent.class)
    public void listenerAction(MqttSubscribedEvent mqttSubscribedEvent) {
        log.info("消息订阅的事件通知");
    }
}

7、接口测试

@Resource
    private MqttGateway mqttGateway;
    
    @RequestMapping(value = "/sendMqtt",method = RequestMethod.POST)
    public String sendMqtt(String sendData, String topic) {
        MqttMessage mqttMessage = new MqttMessage();
        mqttGateway.sendToMqtt(topic, sendData);
        //mqttGateway.sendToMqttObject(topic, sendData.getBytes());
        return "OK";
    }

总结

以上为个人经验，希望能给大家一个参考，也希望大家多多支持编程网。

您可能感兴趣的文档:

点击免费下载>>软考高级考试备考技巧/历年真题/备考精华资料

--结束END--

本文标题: SpringBoot实现MQTT消息发送和接收方式

本文链接: https://www.lsjlt.com/news/199331.html(转载时请注明来源链接)

有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

本篇文章演示代码以及资料文档资料下载

下载Word文档到电脑，方便收藏和打印～

下载Word文档

去做题

猜你喜欢

SpringBoot实现MQTT消息发送和接收方式

目录Spring integration交互逻辑1、maven依赖2、yaml配置文件3、mqtt生产者消费者配置类4、消息处理类 5、mqtt发送接口 6、mq...

99+

2023-03-11

SpringBoot MQTT消息 MQTT消息发送 MQTT消息接收
SpringBoot如何实现MQTT消息发送和接收

今天小编给大家分享一下SpringBoot如何实现MQTT消息发送和接收的相关知识点，内容详细，逻辑清晰，相信大部分人都还太了解这方面的知识，所以分享这篇文章给大家参考一下，希望大家阅读完这篇文章后有所收获，下面我们一起来了解一下吧。Spr...

99+

2023-07-05
Springboot中RocketMQ怎么实现消息发送与接收

本文小编为大家详细介绍“Springboot中RocketMQ怎么实现消息发送与接收”，内容详细，步骤清晰，细节处理妥当，希望这篇“Springboot中RocketMQ怎么实现消息发送与接收”文章能帮助大家解决疑惑，下面跟着小编的思路慢慢...

99+

2023-07-02
Springboot详解RocketMQ实现消息发送与接收流程

springboot+rockermq 实现简单的消息发送与接收普通消息的发送方式有3种：单向发送、同步发送和异步发送。下面来介绍下 springboot+rockermq 整合...

99+

2024-04-02
SpringBoot整合RocketMQ实现消息发送和接收的详细步骤

我们使用主流的SpringBoot框架整合RocketMQ来讲解，使用方便快捷；最终项目结构如下：具体步骤如下：第一步：新建SpringBoot项目rocketmq-test...

99+

2024-04-02
Go语言实现服务端消息接收和发送

本文实例为大家分享了Go实现服务端消息接收和发送的具体代码，供大家参考，具体内容如下一. 服务端接收数据并返回数据服务端代码 package main import ( ...

99+

2024-04-02
C#使用udp如何实现消息的接收和发送

目录使用udp实现消息的接收和发送C# 运用UDP服务端界面客户端界面总结使用udp实现消息的接收和发送代码比较简单，但是别忘记关闭防火墙进行测试。首先便是服务端，使用Socke...

99+

2023-02-26

C#使用udp udp消息接收 udp消息发送
C#怎么使用udp实现消息的接收和发送

本篇内容主要讲解“C#怎么使用udp实现消息的接收和发送”，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让小编来带大家学习“C#怎么使用udp实现消息的接收和发送”吧!使用udp实现消息的接收和发送代码比较简单，但是...

99+

2023-07-05
springboot整合netty-mqtt-client实现Mqtt消息的订阅和发布示例

目录1.添加依赖2.源码3.运行测试1.添加依赖 <dependency> <groupId>org.jetlinks</groupId>...

99+

2024-04-02
SpringBoot2.0集成MQTT消息推送功能实现

这几天在弄后端管理系统向指定的Android客户端推送消息的功能模块，查阅了网上很多博...

99+

2024-04-02
如何进行SpringBoot+RabbitMQ方式收发消息

本篇文章给大家分享的是有关如何进行SpringBoot+RabbitMQ方式收发消息，小编觉得挺实用的，因此分享给大家学习，希望大家阅读完这篇文章后可以有所收获，话不多说，跟着小编一起来看看吧。本篇会和SpringBoot做整合，采用自动配...

99+

2023-06-16
PHP实现在线即时通讯系统消息的发送和接收

随着人们生活方式的改变和网络技术的不断发展，即时通讯已经成为现代人必不可少的沟通方式之一。在这种趋势下，各种在线即时通讯系统应运而生。既然在不同地方的人可以在同一时间内交换消息，那么在线即时通讯系统是如何实现消息的传递的呢？本文将介绍通过 ...

99+

2023-05-24

PHP 消息发送在线通讯
react-native消息推送实现方式

目录react-native极光推送一、安装插件二、配置安卓配置IOS配置三、使用解决ios角标无法清除总结react-native极光推送先去官网注册：https://www.j...

99+

2023-02-18

react-native消息推送 react-native 消息推送
SpringBoot+WebSocket实现消息推送功能

目录背景WebSocket简介协议原理WebSocket与HTTP协议的区别WebSocket特点应用场景系统集成Websocketjar包引入Websocket配置具体实现测试示例...

99+

2022-11-13

SpringBoot WebSocket消息推送 SpringBoot 消息推送 SpringBoot WebSocket
springboot实现公众号接收回复消息和超过5秒被动回复消息

目录1.首先第一步要接收微信消息，需要在公众号里设置与开发-基本配置里配置一下服务器配置2.配置好公众号以后，开始接收微信消息本次就是记录一下我的开发过程，不是教程，纯属自己做个笔记...

99+

2024-04-02
PHP使用Socket和EPOLL实现在线即时通讯系统消息的发送和接收

随着互联网的不断发展，即时通讯(IM)系统已成为人们生活、工作中不可或缺的工具。而其中，如何保证消息的快速传输和实时性就成为了系统设计中至关重要的一环。本文将介绍如何使用PHP的Socket和EPOLL实现在线即时通讯系统消息的发送和接收。...

99+

2023-05-24

PHP socket EPOLL
esp8266模块--MQTT协议连接服务器实现数据接收和发送+源码

首先推荐中国移动的代码，我觉得中国移动的代码更为合理：（但是有一些其他的模块在里面） OneNET开发板代码、资料--2020-09-27--标准板、Mini板bug修复 - 开发板专区 - OneNET设备云论坛 (10086.cn) 以...

99+

2023-09-20

笔记嵌入式硬件 stm32
kafka发送消息的方式有哪些

今天小编给大家分享一下kafka发送消息的方式有哪些的相关知识点，内容详细，逻辑清晰，相信大部分人都还太了解这方面的知识，所以分享这篇文章给大家参考一下，希望大家阅读完这篇文章后有所收获，下面我们一起来了解一下吧。kafka发送消息的方式p...

99+

2023-07-05
python实现定时发送qq消息

因为生活中老是忘记各种事情，刚好又在学python，便突发奇想通过python实现提醒任务的功能（尽管TIM有定时功能），也可定时给好友、群、讨论组发送qq消息。其工作流程是：访问数据库提取最近计划——>根据数据内容（提醒时间、提醒...

99+

2023-01-31

消息 python qq
java发送kafka事务消息的实现方法

前言事务对java开发的同学来说并不陌生，我们使用事务的目的在于避免产生重复数据或者说利用数据存储中间件的事务特性确保数据的精准性，比如大家熟悉的mysql，我们在程序开始时，只需...

99+

2024-04-02