IOT云平台 simple（6）springboot netty实现IOT云平台基本的架构（mqtt、Rabbitmq）

青旅半醒 2024-03-30 14:44 13阅读 0赞

本系列教程包括：  
[IOT云平台 simple（0）IOT云平台简介][IOT_ simple_0_IOT]  
[IOT云平台 simple（1）netty入门][IOT_ simple_1_netty]  
[IOT云平台 simple（2）springboot入门][IOT_ simple_2_springboot]  
[IOT云平台 simple（3）springboot netty实现TCP Server][IOT_ simple_3_springboot netty_TCP Server]  
[IOT云平台 simple（4）springboot netty实现简单的mqtt broker][IOT_ simple_4_springboot netty_mqtt broker]  
[IOT云平台 simple（5）springboot netty实现modbus TCP Master][IOT_ simple_5_springboot netty_modbus TCP Master]  
[IOT云平台 simple（6）springboot netty实现IOT云平台基本的架构（mqtt、Rabbitmq）][IOT_ simple_6_springboot netty_IOT_mqtt_Rabbitmq]

本章首先简单介绍了IOT云平台最基本的架构，然后基于springboot netty实现IOT Server；最后进行了测试验证。

测试环境：

1.  mqtt终端，这里用MQTT.fx 工具软件模拟；
2.  IOT server：基于springboot netty进行开发；
3.  Rabbitmq broker；本地安装Windows 64位环境；
4.  Rabbitmq consumer，订阅mq message的server，这里用MQTT Assistant工具软件模拟；

## 1 IOT云平台最基本的架构 ##

本章涉及的IOT云平台最基本架构图：  
![在这里插入图片描述][16a14e3ac11c442abfd65d4a740f29ed.png]  
**说明**：  
1）为了简单，这里只包括mqtt上行链路，即mqtt终端上传数据；  
2）这里通过Rabbitmq 进行消息的分发，也可以用其他mq中间件，如kafka。  
3）Mqtt terminal：mqtt终端，指的是具体的设备传感器或者mqtt协议网关。

**具体流程**：  
第1步，mqtt终端  
实现mqtt协议或者其他协议（modbus、wifi、蓝牙）转换为mqtt协议。

第2步，mqtt终端->IOT server  
mqtt终端publish message到IOT server；

第3步，IOT server->Rabbitmq broker  
IOT Server中mqtt broker的模块收到mqtt message，进行解析，然后通过Rabbitmq producer模块，publish message到Rabbitmq broker；

第4步，Rabbitmq broker->不同server  
Rabbitmq broker收到消息存入指定的queue，然后分发到订阅消息的不同server；

第5步，不同server  
不同server监听收到消息进行相应的处理；如：存入到时序数据库、进行大数据流的计算、具体业务的处理。

## 2 集成开发 ##

第1步：POM文件引入netty、Rabbitmq的依赖：

<dependency>
                <groupId>io.netty</groupId>
                <artifactId>netty-all</artifactId>
                <version>4.1.63.Final</version>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-amqp</artifactId>
                <version>3.0.1</version>
            </dependency>

### 2.1实现mqtt broker ###

创建主要的类：  
1）TCPServer  
server类，实现mqtt broker。  
2 ）TCPServerStartListener  
监听到springboot启动后，启动server。  
3）TCPServerChannelInitializer  
server channel初始化的类

包括两个server channel处理的类：  
1） MqttMessageChannelHandler：  
server channel处理的类；实现mqtt消息的解析；

@Component
    @Slf4j
    @ChannelHandler.Sharable
    public class MqttMessageChannelHandler extends ChannelInboundHandlerAdapter  {
        
    
        @Autowired
        MessageStrategyManager messageStrategyManager;
    
        @Override
        public void channelRead(ChannelHandlerContext channelHandlerContext, Object msg) throws Exception {
        
            MqttMessage mqttMessage = (MqttMessage) msg;
            log.info("--------------------------channelRead begin---------------------------*");
            log.info("from client:" + channelHandlerContext.channel().remoteAddress());
            log.info("receive message：" + mqttMessage.toString());
            try {
        
                MqttMessageType type = mqttMessage.fixedHeader().messageType();
                MessageStrategy messageStrategy =  messageStrategyManager.getMessageStrategy(type);
                if(messageStrategy!=null){
        
                    messageStrategy.sendResponseMessage(channelHandlerContext,mqttMessage);
                }
            }catch (Exception e) {
        
                e.printStackTrace();
            }
            log.info("--------------------------channelRead end---------------------------*");
    
        }
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
         // (4)
            // 当出现异常就关闭连接
            cause.printStackTrace();
            ctx.close();
        }
    }

2） MqMessageChannelHandler：  
server channel处理的类；实现mq消息发布到Rabbitmq broker；

@Component
    @Slf4j
    public class MqMessageChannelHandler extends ChannelInboundHandlerAdapter {
        
        @Autowired
        ProducerService producerService;
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        
            if (!(msg instanceof MqMessage)) {
        
                return;
            }
            MqMessage mqMessage = (MqMessage) msg;
            log.info("转发到Rabbitmq Server：" + mqMessage.data);
            producerService.sendData(mqMessage.data);
        }
    
    }

### 2.2实现Rabbitmq producer ###

1 定义配置类ProducerConfig：

exchange（交换机）：topic\_exchange  
queue（队列）：topic\_queue  
bindKey（绑定key）：project1.station1.\*

@Slf4j
    @Configuration
    public class ProducerConfig {
        
        String bindKey = "project1.station1.*";
    
        @Bean
        public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory){
        
            RabbitTemplate rabbitTemplate = new RabbitTemplate();
            rabbitTemplate.setConnectionFactory(connectionFactory);
            //开启Mandatory,触发回调函数
            rabbitTemplate.setMandatory(true);
            //ack
            rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
        
                @Override
                public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        
                    log.info("-----------confirm begin---------------");
                    log.info("data:" + correlationData);
                    if(ack){
        
                        log.info("Ack:true");
                    }else{
        
                        log.info("Ack:false");
                    }
                    log.info("cause:" + cause);
                    log.info("-----------confirm end---------------");
                }
            });
            //return
            rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
        
                @Override
                public void returnedMessage(ReturnedMessage returnedMessage) {
        
                    log.info("-----------return begin---------------");
                    log.info("message:"+returnedMessage.getMessage());
                    log.info("reply code:"+returnedMessage.getReplyCode());
                    log.info("reply text:"+returnedMessage.getReplyText());
                    log.info("exchange:"+returnedMessage.getExchange());
                    log.info("routeKey:"+returnedMessage.getRoutingKey());
                    log.info("-----------return end---------------");
                }
            });
    
            return rabbitTemplate;
        }
    
        @Bean("topic_exchange")
        public TopicExchange topicExchange() {
        
            return ExchangeBuilder.topicExchange("topic_exchange").durable(true).build();
        }
    
        @Bean("topic_queue")
        public Queue topicQueue(){
        
            return QueueBuilder.durable("topic_queue").build();
        }
    
        @Bean
        public Binding topicBind(){
        
            return BindingBuilder.bind(topicQueue()).to(topicExchange()).with(bindKey);
        }
    }

2 定义ProducerService类，实现发送消息的功能：  
这里发送：  
exchange（交换机）：topic\_exchange  
routeKey （路由key）：project1.station1.device1

@Slf4j
    @Component
    public class ProducerService {
        
        String exchange = "topic_exchange";
        String routeKey = "project1.station1.device1";
        @Resource
        private RabbitTemplate rabbitTemplate;
    
        public void sendData(String data) {
        
            CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
            log.info("------------topic producer begin------------");
            rabbitTemplate.convertAndSend(exchange, routeKey, data);
            log.info("exchange:"+exchange);
            log.info("routeKey:"+routeKey);
            log.info("send data:"+data);
            log.info("------------topic producer end------------");
        }
    }

## 3 测试验证 ##

第1步：mqtt终端发送消息：temperature is 26, 2023.1.14 17:00。  
![在这里插入图片描述][6952022507264ea482da57a77e0c2262.png]  
第2步：IOT Server接收到mqtt终端发送的消息；然后转发到Rabbitmq Server；  
![在这里插入图片描述][a7198fc1050148089f09e2adb696b68b.png]  
第3步：Rabbitmq Server的topic\_queue中有1条消息；  
![在这里插入图片描述][2f5406cd3efb435e815f02a39452a9c9.png]  
第4步：MQTT Assistant中从topic\_queue中消费消息：  
![在这里插入图片描述][61298f707cb346ab903d58c007d121fa.png]  
可见，对于物联网的数据，实现了从端到平台（解析、分发、存储）的整个流程。

代码详见：  
[https://gitee.com/linghufeixia/iot-simple][https_gitee.com_linghufeixia_iot-simple]  
code5

[IOT_ simple_0_IOT]: https://blog.csdn.net/afei8080/article/details/115539444
[IOT_ simple_1_netty]: https://blog.csdn.net/afei8080/article/details/115539643
[IOT_ simple_2_springboot]: https://blog.csdn.net/afei8080/article/details/115540777
[IOT_ simple_3_springboot netty_TCP Server]: https://blog.csdn.net/afei8080/article/details/127832895
[IOT_ simple_4_springboot netty_mqtt broker]: https://blog.csdn.net/afei8080/article/details/127899022
[IOT_ simple_5_springboot netty_modbus TCP Master]: https://blog.csdn.net/afei8080/article/details/128242813
[IOT_ simple_6_springboot netty_IOT_mqtt_Rabbitmq]: https://blog.csdn.net/afei8080/article/details/128557674
[16a14e3ac11c442abfd65d4a740f29ed.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/dd551244224949f280ea3db7b622e9ff.png
[6952022507264ea482da57a77e0c2262.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/b9c44422e5f34f7b8dd7c0310d8c2e0a.png
[a7198fc1050148089f09e2adb696b68b.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/288e1f56c6424ad5bdf8f993bc9a05b8.png
[2f5406cd3efb435e815f02a39452a9c9.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/21de7155f3234ff98da4f93a33ce97e0.png
[61298f707cb346ab903d58c007d121fa.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/30/9576b3a74e6d4d39936bf35be92d1a03.png
[https_gitee.com_linghufeixia_iot-simple]: https://gitee.com/linghufeixia/iot-simple