devbox
繁依Fanyi0
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

基于SpringBoot整合Netty开发MQTT服务端_netty mqtt

作者：繁依Fanyi0 | 2024-03-06 20:15:26

赞

踩

netty mqtt

Netty认知

Netty是一款基于NIO（Nonblocking I/O，非阻塞IO）开发的网络通信框架，相比传统Socket，在并发性方面有着很大的提升。关于NIO,BIO,AIO之间的区别，可以参考这篇博客：NIO 、 BIO与AIO之间的区别_谁念西风独自凉-CSDN博客

 MQTT服务端实现

首先我们启动一个tcp服务，这里我用到了Redis与RabbitMQ，主要是与分布式WEB平台之间好对接

  1. @Component
  2. public class ApplicationEventListener implements CommandLineRunner {
  3.     @Value("${spring.application.name}")
  4.     private String nodeName;
  5.  
  6.     @Value("${gnss.mqttserver.tcpPort}")
  7.     private int tcpPort;
  8.  
  9.     @Override
  10.     public void run(String... args) throws Exception {
  11.         //启动TCP服务
  12.         startTcpServer();
  13.  
  14.         //清除Redis所有此节点的在线终端
  15.         RedisService redisService = SpringBeanService.getBean(RedisService.class);
  16.         redisService.deleteAllOnlineTerminals(nodeName);
  17.  
  18.         //将所有此节点的终端设置为离线
  19.         RabbitMessageSender messageSender = SpringBeanService.getBean(RabbitMessageSender.class);
  20.         messageSender.noticeAllOffline(nodeName);
  21.     }
  22.  
  23.     /**
  24.      * 启动TCP服务
  25.      *
  26.      * @throws Exception
  27.      */
  28.     private void startTcpServer() throws Exception {
  29.         //计数器,必须等到所有服务启动成功才能进行后续的操作
  30.         final CountDownLatch countDownLatch = new CountDownLatch(1);
  31.         //启动TCP服务
  32.         TcpServer tcpServer = new TcpServer(tcpPort, ProtocolEnum.MqttCommon, countDownLatch);
  33.         tcpServer.start();
  34.         //等待启动完成
  35.         countDownLatch.await();
  36.     }
  37. }

接下来我们编写一个TcpServer类实现TCP服务

  1. @Slf4j
  2. public class TcpServer extends Thread{
  3.     private int port;
  4.  
  5.     private ProtocolEnum protocolType;
  6.  
  7.     private EventLoopGroup bossGroup;
  8.  
  9.     private EventLoopGroup workerGroup;
  10.  
  11.     private ServerBootstrap serverBootstrap = new ServerBootstrap();
  12.  
  13.     private CountDownLatch countDownLatch;
  14.  
  15.     public TcpServer(int port, ProtocolEnum protocolType, CountDownLatch countDownLatch) {
  16.         this.port = port;
  17.         this.protocolType = protocolType;
  18.         this.countDownLatch = countDownLatch;
  19.  
  20.         bossGroup = new NioEventLoopGroup(1);
  21.         workerGroup = SpringBeanService.getBean("workerGroup", EventLoopGroup.class);
  22.         final EventExecutorGroup executorGroup = SpringBeanService.getBean("executorGroup", EventExecutorGroup.class);
  23.         serverBootstrap.group(bossGroup, workerGroup)
  24.                 .channel(NioServerSocketChannel.class)
  25.                 .option(ChannelOption.SO_BACKLOG, 1024)
  26.                 .childOption(ChannelOption.SO_KEEPALIVE, true)
  27.                 .childOption(ChannelOption.TCP_NODELAY, true)
  28.                 .childHandler(new ChannelInitializer<SocketChannel>() {
  29.  
  30.                     @Override
  31.                     protected void initChannel(SocketChannel ch) throws Exception {
  32.                         ch.pipeline().addLast(new IdleStateHandler(MqttConstant.READER_IDLE_TIME, 0, 0, TimeUnit.SECONDS));
  33.                         ch.pipeline().addLast("encoder", MqttEncoder.INSTANCE);
  34.                         ch.pipeline().addLast("decoder", new MqttDecoder());
  35.                         ch.pipeline().addLast(executorGroup, MqttBusinessHandler.INSTANCE);
  36.                     }
  37.                 });
  38.     }
  39.  
  40.     @Override
  41.     public void run() {
  42.         bind();
  43.     }
  44.  
  45.     /**
  46.      * 绑定端口启动服务
  47.      */
  48.     private void bind() {
  49.         serverBootstrap.bind(port).addListener(future -> {
  50.             if (future.isSuccess()) {
  51.                 log.info("{} MQTT服务器启动,端口:{}", protocolType, port);
  52.                 countDownLatch.countDown();
  53.             } else {
  54.                 log.error("{} MQTT服务器启动失败,端口:{}", protocolType, port, future.cause());
  55.                 System.exit(-1);
  56.             }
  57.         });
  58.     }
  59.  
  60.     /**
  61.      * 关闭服务端
  62.      */
  63.     public void shutdown() {
  64.         workerGroup.shutdownGracefully();
  65.         bossGroup.shutdownGracefully();
  66.         log.info("{} TCP服务器关闭,端口:{}", protocolType, port);
  67.     }
  68. }

编写一个解码器MqttBusinessHandler，实现对MQTT消息接收与处理

  1. @Slf4j
  2. @ChannelHandler.Sharable
  3. public class MqttBusinessHandler extends SimpleChannelInboundHandler<Object> {
  4.     public static final MqttBusinessHandler INSTANCE = new MqttBusinessHandler();
  5.     private MqttMsgBack mqttMsgBack;
  6.     private MqttBusinessHandler() {
  7.         mqttMsgBack= MqttMsgBack.INSTANCE;
  8.     }
  9.  
  10.     /**
  11.      * 接收到消息后处理
  12.      * @param ctx
  13.      * @param msg
  14.      * @throws Exception
  15.      */
  16.     @Override
  17.     protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
  18.         if (null != msg) {
  19.             MqttMessage mqttMessage = (MqttMessage) msg;
  20.             MqttFixedHeader mqttFixedHeader = mqttMessage.fixedHeader();
  21.             Channel channel = ctx.channel();
  22.             if(mqttFixedHeader.messageType().equals(MqttMessageType.CONNECT)){
  23.                 //在一个网络连接上，客户端只能发送一次CONNECT报文。服务端必须将客户端发送的第二个CONNECT报文当作协议违规处理并断开客户端的连接
  24.                 //建议connect消息单独处理，用来对客户端进行认证管理等 这里直接返回一个CONNACK消息
  25.                 mqttMsgBack.connectionAck(ctx, mqttMessage);
  26.             }
  27.  
  28.             switch (mqttFixedHeader.messageType()){
  29.                 //客户端发布消息
  30.                 case PUBLISH:
  31.                     mqttMsgBack.publishAck(ctx, mqttMessage);
  32.                     break;
  33.                 //发布释放
  34.                 case PUBREL:
  35.                     mqttMsgBack.publishComp(ctx, mqttMessage);
  36.                     break;
  37.                 //订阅主题
  38.                 case SUBSCRIBE:
  39.                     mqttMsgBack.subscribeAck(ctx, mqttMessage);
  40.                     break;
  41.                 //取消订阅主题
  42.                 case UNSUBSCRIBE:
  43.                     mqttMsgBack.unsubscribeAck(ctx, mqttMessage);
  44.                     break;
  45.                 //客户端发送心跳报文
  46.                 case PINGREQ:
  47.                     mqttMsgBack.pingResp(ctx, mqttMessage);
  48.                     break;
  49.                 //客户端主动断开连接
  50.                 case DISCONNECT:
  51.                     break;
  52.                 default:
  53.                     break;
  54.             }
  55.         }
  56.     }
  57.  
  58.     @Override
  59.     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
  60.         log.info("终端关闭连接,IP信息:{}", CommonUtil.getClientAddress(ctx));
  61.     }
  62.  
  63.     @Override
  64.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  65.         ctx.close();
  66.         log.error("终端连接异常,IP信息:{}", CommonUtil.getClientAddress(ctx), cause);
  67.     }
  68.  
  69.     /**
  70.      * 	服务端当读超时时会调用这个方法
  71.      */
  72.     @Override
  73.     public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception, IOException {
  74.         ctx.close();
  75.         log.error("读超时,IP信息:{}", CommonUtil.getClientAddress(ctx), evt);
  76.     }

我们对接收到的消息进行业务处理

  1. @Slf4j
  2. public class MqttMsgBack {
  3.     public static final MqttMsgBack INSTANCE = new MqttMsgBack();
  4.     private RedisService redisService;
  5.     private RabbitMessageSender messageSender;
  6.     private Environment environment;
  7.     private MessageServiceProvider messageServiceProvider;
  8.  
  9.     private MqttMsgBack() {
  10.         redisService = SpringBeanService.getBean(RedisService.class);
  11.         messageSender = SpringBeanService.getBean(RabbitMessageSender.class);
  12.         environment = SpringBeanService.getBean(Environment.class);
  13.         messageServiceProvider = SpringBeanService.getBean(MessageServiceProvider.class);
  14.     }
  15.  
  16.     /**
  17.      * 	确认连接请求
  18.      * @param ctx
  19.      * @param mqttMessage
  20.      */
  21.     public void connectionAck (ChannelHandlerContext ctx, MqttMessage mqttMessage) {
  22.         MqttConnectMessage mqttConnectMessage = (MqttConnectMessage) mqttMessage;
  23.         MqttFixedHeader mqttFixedHeaderInfo = mqttConnectMessage.fixedHeader();
  24.         MqttConnectVariableHeader mqttConnectVariableHeaderInfo = mqttConnectMessage.variableHeader();
  25.         //构建返回报文， 可变报头
  26.         MqttConnAckVariableHeader mqttConnAckVariableHeaderBack = new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_ACCEPTED, mqttConnectVariableHeaderInfo.isCleanSession());
  27.         //构建返回报文， 固定报头
  28.         MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.CONNACK,mqttFixedHeaderInfo.isDup(), MqttQoS.AT_MOST_ONCE, mqttFixedHeaderInfo.isRetain(), 0x02);
  29.         //构建连接回复消息体
  30.         MqttConnAckMessage connAck = new MqttConnAckMessage(mqttFixedHeaderBack, mqttConnAckVariableHeaderBack);
  31.         ctx.writeAndFlush(connAck);
  32.         //获取连接者的ClientId
  33.         String clientIdentifier = mqttConnectMessage.payload().clientIdentifier();
  34.         //查询终端号码有无在平台注册
  35.         TerminalProto terminalInfo = redisService.getTerminalInfoByTerminalNum(clientIdentifier);
  36.         if (terminalInfo == null) {
  37.             log.error("终端登录失败,未找到终端信息,终端号:{},IP信息:{}", clientIdentifier, CommonUtil.getClientAddress(ctx));
  38.             ctx.close();
  39.             return;
  40.         }
  41.         //设置节点名
  42.         terminalInfo.setNodeName(environment.getProperty("spring.application.name"));
  43.         //保存终端信息和消息流水号到上下文属性中
  44.         Session session = new Session(terminalInfo);
  45.         ChannelHandlerContext oldCtx = SessionUtil.bindSession(session, ctx);
  46.         if (oldCtx == null) {
  47.             log.info("终端登录成功,终端ID:{},终端号:{},IP信息:{}", terminalInfo.getTerminalStrId(), clientIdentifier, CommonUtil.getClientAddress(ctx));
  48.         } else {
  49.             log.info("终端重复登录关闭上一个连接,终端ID:{},终端号:{},IP信息:{}", terminalInfo.getTerminalStrId(), clientIdentifier, CommonUtil.getClientAddress(ctx));
  50.             oldCtx.close();
  51.         }
  52.         //通知上线
  53.         messageSender.noticeOnline(terminalInfo);
  54.         log.info("终端登录成功,终端号:{},IP信息:{}", clientIdentifier, CommonUtil.getClientAddress(ctx));
  55.     }
  56.  
  57.     /**
  58.      * 	根据qos发布确认
  59.      * @param ctx
  60.      * @param mqttMessage
  61.      */
  62.     public void publishAck (ChannelHandlerContext ctx, MqttMessage mqttMessage) {
  63.         MqttPublishMessage mqttPublishMessage = (MqttPublishMessage) mqttMessage;
  64.         MqttFixedHeader mqttFixedHeaderInfo = mqttPublishMessage.fixedHeader();
  65.         MqttQoS qos = (MqttQoS) mqttFixedHeaderInfo.qosLevel();
  66.         //得到主题
  67.         String topicName = mqttPublishMessage.variableHeader().topicName();
  68.         //获取消息体
  69.         ByteBuf msgBodyBuf = mqttPublishMessage.payload();
  70.         log.info("收到:{}", ByteBufUtil.hexDump(msgBodyBuf));
  71.         MqttCommonMessage msg=new MqttCommonMessage();
  72.         msg.setTerminalNum(SessionUtil.getTerminalInfo(ctx).getTerminalNum());
  73.         msg.setStrMsgId(topicName);
  74.         //根据主题获取对应的主题消息处理器
  75.         BaseMessageService messageService = messageServiceProvider.getMessageService(topicName);
  76.         try {
  77.             Object result = messageService.process(ctx, msg, msgBodyBuf);
  78.             log.info("收到{}({}),终端ID:{},内容:{}", messageService.getDesc(), topicName,msg.getTerminalNum(), msg.getMsgBodyItems());
  79.         } catch (Exception e) {
  80.             log.error("收到{}({}),消息异常,终端ID:{},消息体:{}", messageService.getDesc(), topicName,msg.getTerminalNum(),ByteBufUtil.hexDump(msgBodyBuf), e);
  81.         }
  82.         switch (qos) {
  83.             //至多一次
  84.             case AT_MOST_ONCE:
  85.                 break;
  86.             //至少一次
  87.             case AT_LEAST_ONCE:
  88.                 //构建返回报文， 可变报头
  89.                 MqttMessageIdVariableHeader mqttMessageIdVariableHeaderBack = MqttMessageIdVariableHeader.from(mqttPublishMessage.variableHeader().packetId());
  90.                 //构建返回报文， 固定报头
  91.                 MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.PUBACK,mqttFixedHeaderInfo.isDup(), MqttQoS.AT_MOST_ONCE, mqttFixedHeaderInfo.isRetain(), 0x02);
  92.                 //构建PUBACK消息体
  93.                 MqttPubAckMessage pubAck = new MqttPubAckMessage(mqttFixedHeaderBack, mqttMessageIdVariableHeaderBack);
  94.                 log.info("Qos:AT_LEAST_ONCE:{}",pubAck.toString());
  95.                 ctx.writeAndFlush(pubAck);
  96.                 break;
  97.             //刚好一次
  98.             case EXACTLY_ONCE:
  99.                 //构建返回报文，固定报头
  100.                 MqttFixedHeader mqttFixedHeaderBack2 = new MqttFixedHeader(MqttMessageType.PUBREC,false, MqttQoS.AT_LEAST_ONCE,false,0x02);
  101.                 //构建返回报文，可变报头
  102.                 MqttMessageIdVariableHeader mqttMessageIdVariableHeaderBack2 = MqttMessageIdVariableHeader.from(mqttPublishMessage.variableHeader().packetId());
  103.                 MqttMessage mqttMessageBack = new MqttMessage(mqttFixedHeaderBack2,mqttMessageIdVariableHeaderBack2);
  104.                 log.info("Qos:EXACTLY_ONCE回复:{}"+mqttMessageBack.toString());
  105.                 ctx.writeAndFlush(mqttMessageBack);
  106.                 break;
  107.             default:
  108.                 break;
  109.         }
  110.     }
  111.  
  112.     /**
  113.      * 发布完成 qos2
  114.      * @param ctx
  115.      * @param mqttMessage
  116.      */
  117.     public void publishComp (ChannelHandlerContext ctx, MqttMessage mqttMessage) {
  118.  
  119.         MqttMessageIdVariableHeader messageIdVariableHeader = (MqttMessageIdVariableHeader) mqttMessage.variableHeader();
  120.         //构建返回报文， 固定报头
  121.         MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.PUBCOMP,false, MqttQoS.AT_MOST_ONCE,false,0x02);
  122.         //构建返回报文， 可变报头
  123.         MqttMessageIdVariableHeader mqttMessageIdVariableHeaderBack = MqttMessageIdVariableHeader.from(messageIdVariableHeader.messageId());
  124.         MqttMessage mqttMessageBack = new MqttMessage(mqttFixedHeaderBack,mqttMessageIdVariableHeaderBack);
  125.         log.info("发布完成回复:{}"+mqttMessageBack.toString());
  126.         ctx.writeAndFlush(mqttMessageBack);
  127.     }
  128.  
  129.     /**
  130.      * 	订阅确认
  131.      * @param ctx
  132.      * @param mqttMessage
  133.      */
  134.     public void subscribeAck(ChannelHandlerContext ctx, MqttMessage mqttMessage) {
  135.         MqttSubscribeMessage mqttSubscribeMessage = (MqttSubscribeMessage) mqttMessage;
  136.         MqttMessageIdVariableHeader messageIdVariableHeader = mqttSubscribeMessage.variableHeader();
  137.         //构建返回报文， 可变报头
  138.         MqttMessageIdVariableHeader variableHeaderBack = MqttMessageIdVariableHeader.from(messageIdVariableHeader.messageId());
  139.         Set<String> topics = mqttSubscribeMessage.payload().topicSubscriptions().stream().map(mqttTopicSubscription -> mqttTopicSubscription.topicName()).collect(Collectors.toSet());
  140.         List<Integer> grantedQoSLevels = new ArrayList<>(topics.size());
  141.         for (int i = 0; i < topics.size(); i++) {
  142.             grantedQoSLevels.add(mqttSubscribeMessage.payload().topicSubscriptions().get(i).qualityOfService().value());
  143.         }
  144.         //	构建返回报文	有效负载
  145.         MqttSubAckPayload payloadBack = new MqttSubAckPayload(grantedQoSLevels);
  146.         //	构建返回报文	固定报头
  147.         MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.SUBACK, false, MqttQoS.AT_MOST_ONCE, false, 2+topics.size());
  148.         //	构建返回报文	订阅确认
  149.         MqttSubAckMessage subAck = new MqttSubAckMessage(mqttFixedHeaderBack,variableHeaderBack, payloadBack);
  150.         log.info("订阅回复:{}", subAck.toString());
  151.         ctx.writeAndFlush(subAck);
  152.     }
  153.  
  154.     /**
  155.      * 取消订阅确认
  156.      * @param ctx
  157.      * @param mqttMessage
  158.      */
  159.     public void unsubscribeAck(ChannelHandlerContext ctx, MqttMessage mqttMessage) {
  160.         MqttMessageIdVariableHeader messageIdVariableHeader = (MqttMessageIdVariableHeader) mqttMessage.variableHeader();
  161.         //	构建返回报文	可变报头
  162.         MqttMessageIdVariableHeader variableHeaderBack = MqttMessageIdVariableHeader.from(messageIdVariableHeader.messageId());
  163.         //	构建返回报文	固定报头
  164.         MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.UNSUBACK, false, MqttQoS.AT_MOST_ONCE, false, 2);
  165.         //	构建返回报文	取消订阅确认
  166.         MqttUnsubAckMessage unSubAck = new MqttUnsubAckMessage(mqttFixedHeaderBack,variableHeaderBack);
  167.         log.info("取消订阅回复:{}",unSubAck.toString());
  168.         ctx.writeAndFlush(unSubAck);
  169.     }
  170.  
  171.     /**
  172.      * 心跳响应
  173.      * @param ctx
  174.      * @param mqttMessage
  175.      */
  176.     public void pingResp (ChannelHandlerContext ctx, MqttMessage mqttMessage) {
  177.         MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PINGRESP, false, MqttQoS.AT_MOST_ONCE, false, 0);
  178.         MqttMessage mqttMessageBack = new MqttMessage(fixedHeader);
  179.         log.info("心跳回复:{}", mqttMessageBack.toString());
  180.         ctx.writeAndFlush(mqttMessageBack);
  181.     }
  182. }

我们可以根据客户端发布消息的主题匹配不同的处理器

 最后，我们在对应的处理器里面实现对主题消息的处理逻辑，比如：定位消息，指令消息等等，比如简单实现对定位数据Location主题的消息处理

  1. @Slf4j
  2. @MessageService(strMessageId = "Location", desc = "定位")
  3. public class LocationMessageService extends BaseMessageService<MqttCommonMessage> {
  4.     @Autowired
  5.     private RabbitMessageSender messageSender;
  6.  
  7.     @Override
  8.     public Object process(ChannelHandlerContext ctx, MqttCommonMessage msg, ByteBuf msgBodyBuf) throws Exception {
  9.         byte[] msgByteArr = new byte[msgBodyBuf.readableBytes()];
  10.         msgBodyBuf.readBytes(msgByteArr);
  11.         String data = new String(msgByteArr);
  12.         msg.putMessageBodyItem("位置", data);
  13.         return null;
  14.     }
  15. }

 后续

目前仅仅是实现MQTT服务端消息接收与消息回复，后续可以根据接入的物联网设备进行对应主题消息的业务处理

有物联网兴趣的同学可以对加我微信一起交流学习

请添加图片描述

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/繁依Fanyi0/article/detail/201291?site
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号