关于java(springboot)集成mqtt,复制粘贴修改一下直接带走

绝地灬酷狼 2023-01-08 05:22 93阅读 0赞

在工作中遇到了这样的需求,我也是百度的,然后现在整合一下 算是笔记了(删除了没必要的引入,复制到额时候如果还有爆红 记得删除)  
1,依赖

<dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-integration</artifactId>
                <exclusions>
                    <exclusion>
                        <groupId>org.eclipse.paho</groupId>
                        <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
                    </exclusion>
                </exclusions>
            </dependency>
            <dependency>
                <groupId>org.springframework.integration</groupId>
                <artifactId>spring-integration-stream</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.integration</groupId>
                <artifactId>spring-integration-mqtt</artifactId>
            </dependency>

配置文件:

mqtt:
      username:
        #MQTT-密码password: password 可以为空
      password:
        #MQTT-服务器连接地址，如果有多个，用逗号隔开，如：tcp://127.0.0.1:1883，tcp://192.168.2.133:1883
      url: tcp://112.126.56.53:1883
        #MQTT-连接服务器默认客户端ID
      client:
        id: 随便写
        #MQTT-默认的消息推送主题，实际可在调用接口时指定
      default:
        topic: /RAS/+/+(百度一下mqtt的通配符 )
        yihonWQM: 随便写
        #连接超时
      completionTimeout: 3000

mqtt的发送消息用的 (直接复制走)

import org.springframework.integration.annotation.MessagingGateway;
    import org.springframework.integration.mqtt.support.MqttHeaders;
    import org.springframework.messaging.handler.annotation.Header;
    
    /**
     * mqtt发送消息
     * （defaultRequestChannel = "mqttOutboundChannel" 对应config配置）
     * **/
    @MessagingGateway(defaultRequestChannel = "mqttOutboundChannel")
    public interface MqttGateway {
        
     
        /**
         * 发送信息到MQTT服务器
         *
         * @param data 发送的文本
         */
        void sendToMqtt(String data);
     
        /**
         * 发送信息到MQTT服务器
         *
         * @param topic 主题
         * @param payload 消息主体
         */
        void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic,
                        String payload);
     
        /**
         * 发送信息到MQTT服务器
         *
         * @param topic 主题
         * @param qos 对消息处理的几种机制。
         * 0 表示的是订阅者没收到消息不会再次发送，消息会丢失。
         * 1 表示的是会尝试重试，一直到接收到消息，但这种情况可能导致订阅者收到多次重复消息。
         * 2 多了一次去重的动作，确保订阅者收到的消息有一次。
         * @param payload 消息主体
         */
        void sendToMqtt(@Header(MqttHeaders.TOPIC) String topic,
                        @Header(MqttHeaders.QOS) int qos,
                        String payload);
    }

发送消息用的:

import io.swagger.annotations.Api;
    import io.swagger.annotations.ApiOperation;
    import lombok.SneakyThrows;
    import lombok.Synchronized;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.scheduling.support.CronSequenceGenerator;
    import org.springframework.web.bind.annotation.*;
    
    import javax.annotation.Resource;
    import javax.management.monitor.Monitor;
    import java.text.SimpleDateFormat;
    import java.time.LocalDateTime;
    import java.time.format.DateTimeFormatter;
    import java.util.Date;
    import java.util.List;
    import java.util.Map;
    import java.util.concurrent.locks.ReentrantLock;
    
    @RestController
    @RequestMapping("/mqtt")
    @Api("mqtt尝试") //swagger的注解
    @Slf4j //日志注解 
    public class MqttTestController {
        
    
        @Resource
        private MqttGateway mqttGateway;
    
      
    
        /**
         * sendData 消息
         * qos 消息级别 （对应QOS0、QOS1，QOS2）
         * topic 订阅主题
         **/
        @PostMapping("/test")
        public String sendMqtt(/*接收一些需要的东西*/) {
        
    
    		//调用这个东西就可以直接把请求发送传出去 对应的三个参数 看mqttGateway的详解
            mqttGateway.sendToMqtt(topic, qos, sendData);
            return "OK";
        }
    }

监听消息:  
可以根据实际需求进行业务处理,我是将payload转化成了字符串然后处理的 看个人处理方式

import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.integration.mqtt.support.MqttHeaders;
    import org.springframework.messaging.Message;
    import org.springframework.stereotype.Component;
    
    import java.time.LocalDateTime;
    
    /**
     * mqtt客户端消息处理类
     **/
    @Slf4j
    @Component
    public class MqttReceiveHandle {
        
    
        public void handle(Message<?> message) {
        
          log.info("主题：{}，QOS:{}，消息接收到的数据：{}", message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC),
                    message.getHeaders().get(MqttHeaders.RECEIVED_QOS), message.getPayload());
        }
    }

最后一个就是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.List;
    
    /**
     *  mqtt 推送and接收 消息类
     * **/
    @Configuration
    @IntegrationComponentScan
    @Slf4j
    public class MqttSenderAndReceiveConfig {
        
     
        private static final byte[] WILL_DATA;
     
        static {
        
            WILL_DATA = "offline".getBytes();
        }
    
        private MqttPahoMessageDrivenChannelAdapter adapter;
     
        @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.default.yihonWQM}")
        private String yihonWQM;
     
        @Value("${mqtt.completionTimeout}")
        private int completionTimeout;   //连接超时
     
        /**
         *  MQTT连接器选项
         * **/
        @Bean(value = "getMqttConnectOptions")
        public MqttConnectOptions getMqttConnectOptions1(){
        
            MqttConnectOptions mqttConnectOptions=new MqttConnectOptions();
            // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接
            mqttConnectOptions.setCleanSession(true);
            // 设置超时时间 单位为秒
            mqttConnectOptions.setConnectionTimeout(32767);
            mqttConnectOptions.setAutomaticReconnect(true);
            mqttConnectOptions.setUserName(username);
            mqttConnectOptions.setPassword(password.toCharArray());
            mqttConnectOptions.setServerURIs(new String[]{
        hostUrl});
            // 设置会话心跳时间 单位为秒 服务器会每隔600秒的时间向客户端发送心跳判断客户端是否在线，但这个方法并没有重连的机制
           // mqttConnectOptions.setKeepAliveInterval(600);
            // 设置“遗嘱”消息的话题，若客户端与服务器之间的连接意外中断，服务器将发布客户端的“遗嘱”消息。
            //mqttConnectOptions.setWill("willTopic", WILL_DATA, 2, false);
            return mqttConnectOptions;
        }
     
        /**
         * MQTT工厂
         * **/
        @Bean
        public MqttPahoClientFactory mqttClientFactory() {
        
            DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
            factory.setConnectionOptions(getMqttConnectOptions1());
            return factory;
        }
     
        /**
         * MQTT信息通道（生产者）
         * **/
        @Bean
        public MessageChannel mqttOutboundChannel() {
        
            return new DirectChannel();
        }
     
        /**
         * MQTT消息处理器（生产者）
         * **/
        @Bean
        @ServiceActivator(inputChannel = "mqttOutboundChannel")
        public MessageHandler mqttOutbound() {
        
            MqttPahoMessageHandler messageHandler =  new MqttPahoMessageHandler("basetnt", mqttClientFactory());
            messageHandler.setAsync(true);
            messageHandler.setDefaultTopic(defaultTopic);
            return messageHandler;
        }
     
        /**
         * 配置client,监听的topic
         * MQTT消息订阅绑定（消费者）
         * **/
        @Bean
        public MessageProducer inbound() {
        
            MqttPahoMessageDrivenChannelAdapter adapter =
                    new MqttPahoMessageDrivenChannelAdapter("begete" + "_inbound1", mqttClientFactory(),
                            defaultTopic,yihonWQM);
            adapter.setCompletionTimeout(completionTimeout);
            adapter.setConverter(new DefaultPahoMessageConverter());
            adapter.setQos(2);
            adapter.setOutputChannel(mqttInputChannel());
            return adapter;
        }
        /**
         * MQTT信息通道（消费者）
         * **/
        @Bean
        public MessageChannel mqttInputChannel() {
        
            return new DirectChannel();
        }
        /**
         * MQTT消息处理器（消费者）
         * **/
        @Bean
        @ServiceActivator(inputChannel = "mqttInputChannel")
        public MessageHandler handler() {
        
            return new MessageHandler() {
        
                @Override
                public void handleMessage(Message<?> message) throws MessagingException {
        
                    //处理接收消息
                    mqttReceiveHandle.handle(message);
                }
            };
        }
    }