rabbitmq和springboot实现智能家居远程控制

一时失言乱红尘 2023-02-21 09:12 28阅读 0赞

## 一、什么是mqtt协议 ##

MQTT（Message Queuing Telemetry Transport，消息队列遥测传输协议），是一种基于发布/订阅（publish/subscribe）模式的"轻量级"通讯协议，该协议构建于TCP/IP协议上，由IBM在1999年发布。MQTT最大优点在于，可以以极少的代码和有限的带宽，为连接远程设备提供实时可靠的消息服务。作为一种低开销、低带宽占用的即时通讯协议，使其在物联网、小型设备、移动应用等方面有较广泛的应用。

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

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNDQ3MzAx_size_16_color_FFFFFF_t_70]

## 二、添加相关依赖包 ##

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

## 三、配置文件 ##

iot.mqtt.clientId=client
    iot.mqtt.defaultTopic=mqtt_topic
    iot.mqtt.serverClientId=server
    iot.mqtt.servers=tcp://127.0.0.1:1883
    iot.mqtt.username=admin
    iot.mqtt.password=password

## 四、代码实现 ##

import lombok.Data;
    import org.springframework.boot.context.properties.ConfigurationProperties;
    import org.springframework.integration.annotation.IntegrationComponentScan;
    import org.springframework.stereotype.Component;
    
    
    /** * 其他的中间件 * Mosquitto * Eclipse Paho * RabbitMQ * Apache ActiveMQ * HiveMQ * JoramMQ * ThingMQ * VerneMQ * Apache Apollo * emqttd Xively * IBM Websphere */
    @Data
    @Component
    @IntegrationComponentScan
    @ConfigurationProperties(prefix = "iot.mqtt")
    public class MqttConfig { 
    
        /** * 服务地址 */
        private String servers;
    
        /** * 客户端id */
        private String clientId;
    
        /** * 服务端id */
        private String serverClientId;
    
        /** * 默认主题 */
        private String defaultTopic;
    
        /** * 账号 */
        private String username;
    
        /** * 密码 */
        private String password;
    }

import org.springframework.integration.annotation.MessagingGateway;
    import org.springframework.integration.mqtt.support.MqttHeaders;
    import org.springframework.messaging.handler.annotation.Header;
    
    
    @MessagingGateway(defaultRequestChannel = "iotMqttInputChannel")
    public interface IotMqttGateway { 
    
        void sendMessage2Mqtt(String data);
    
        void sendMessage2Mqtt(String data, @Header(MqttHeaders.TOPIC) String topic);
    
        void sendMessage2Mqtt(@Header(MqttHeaders.TOPIC) String topic, @Header(MqttHeaders.QOS) int qos, String payload);
    }

import org.springframework.stereotype.Controller;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RequestParam;
    import org.springframework.web.bind.annotation.ResponseBody;
    import javax.annotation.Resource;
    
    
    @Controller
    @RequestMapping("/mqtt")
    public class IotMqttController { 
    
        @Resource
        private IotMqttGateway iotMqttGateway;
    
        @RequestMapping("/send")
        @ResponseBody
        public String sendMqtt(@RequestParam(value = "topic") String topic, @RequestParam(value = "message") String message) { 
            iotMqttGateway.sendMessage2Mqtt(message);
            return "SUCCESS";
        }
    }

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.mqtt.core.DefaultMqttPahoClientFactory;
    import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
    import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
    import org.springframework.messaging.MessageChannel;
    import org.springframework.messaging.MessageHandler;
    
    
    
    @Configuration
    @IntegrationComponentScan
    public class IotMqttProducerConfig { 
    
        @Value("${iot.mqtt.username}")
        private String username;
    
        @Value("${iot.mqtt.password}")
        private String password;
    
        @Value("${iot.mqtt.servers}")
        private String hostUrl;
    
        @Value("${iot.mqtt.clientId}")
        private String clientId;
    
        @Value("${iot.mqtt.defaultTopic}")
        private String defaultTopic;
    
        @Autowired
        private MqttConfig mqttConfig;
    
        @Bean
        public MqttConnectOptions getMqttConnectOptions(){ 
            MqttConnectOptions mqttConnectOptions=new MqttConnectOptions();
            mqttConnectOptions.setUserName(username);
            mqttConnectOptions.setPassword(password.toCharArray());
            mqttConnectOptions.setServerURIs(new String[]{ hostUrl});
            mqttConnectOptions.setKeepAliveInterval(2);
            return mqttConnectOptions;
        }
    
    
        @Bean
        public MqttPahoClientFactory mqttClientFactory() { 
            DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
            factory.setConnectionOptions(getMqttConnectOptions());
            return factory;
        }
    
        @Bean
        public MessageChannel iotMqttInputChannel() { 
            return new DirectChannel();
        }
    
        @Bean
        @ServiceActivator(inputChannel = "iotMqttInputChannel")
        public MessageHandler mqttOutbound() { 
            MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(mqttConfig.getServerClientId(), mqttClientFactory());
            messageHandler.setAsync(false);
            messageHandler.setDefaultQos(2);
            messageHandler.setDefaultTopic(mqttConfig.getDefaultTopic());
            return messageHandler;
        }
    }

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.support.DefaultPahoMessageConverter;
    import org.springframework.messaging.MessageChannel;
    import org.springframework.messaging.MessageHandler;
    import org.springframework.messaging.MessagingException;
    
    
    @Slf4j
    @Configuration
    @IntegrationComponentScan
    public class IotMqttSubscriberConfig { 
    
        @Value("${iot.mqtt.username}")
        private String username;
    
        @Value("${iot.mqtt.password}")
        private String password;
    
        @Value("${iot.mqtt.servers}")
        private String hostUrl;
    
        @Value("${iot.mqtt.clientId}")
        private String clientId;
    
        @Value("${iot.mqtt.defaultTopic}")
        private String defaultTopic;
    
        @Autowired
        private MqttConfig mqttConfig;
    
        @Bean
        public MqttConnectOptions getMqttConnectOptions(){ 
            MqttConnectOptions mqttConnectOptions=new MqttConnectOptions();
            mqttConnectOptions.setUserName(username);
            mqttConnectOptions.setPassword(password.toCharArray());
            mqttConnectOptions.setServerURIs(new String[]{ hostUrl});
            mqttConnectOptions.setKeepAliveInterval(2);
            return mqttConnectOptions;
        }
    
        @Bean
        public MqttPahoClientFactory mqttClientFactory() { 
            DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
            factory.setConnectionOptions(getMqttConnectOptions());
            return factory;
        }
    
        @Bean
        public MessageChannel iotMqttInputChannel() { 
            return new DirectChannel();
        }
    
        @Bean
        public MessageProducer inbound() { 
            MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(mqttConfig.getClientId(), mqttClientFactory(), mqttConfig.getDefaultTopic());
            adapter.setCompletionTimeout(5000);
            adapter.setConverter(new DefaultPahoMessageConverter());
            adapter.setQos(2);
            adapter.setOutputChannel(iotMqttInputChannel());
            return adapter;
        }
    
    
        @Bean
        @ServiceActivator(inputChannel = "iotMqttInputChannel")
        public MessageHandler handlerTest() { 
            return message -> { 
                try { 
                    String string = message.getPayload().toString();
                    System.out.println("接收到消息：" + string);
                } catch (MessagingException e) { 
                    log.info(e.getMessage());
                }
            };
        }
    }

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
    import org.eclipse.paho.client.mqttv3.MqttCallback;
    import org.eclipse.paho.client.mqttv3.MqttMessage;
    
    
    public class CallBacks implements MqttCallback { 
    
        @Override
        public void connectionLost(Throwable throwable) { 
            System.out.println(throwable);
        }
    
        @Override
        public void messageArrived(String s, MqttMessage mqttMessage) { 
            System.out.println("消息接收" + s + mqttMessage);
        }
    
        @Override
        public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) { 
    
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNDQ3MzAx_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200629212223774.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNDQ3MzAx,size_16,color_FFFFFF,t_70