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

基于Netty的UDP服务端开发_netty udp

作者：小蓝xlanll | 2024-03-06 20:06:07

赞

踩

netty udp

1.前言

之前基于Netty做了一套TCP与MQTT的服务端，随着系统接入的终端类型越来越多，出现了UDP通讯的设备，虽然这样的设备并非主流，而且通讯机制存在问题，为了考虑系统的兼容性，只能将整套服务做全。

2.UDP通讯优缺点

UDP 是一种面向非连接的协议，面向非连接指的是在正式通信前不必与对方先建立连接，不管对方状态就直接发送数据。至于对方是否可以接收到这些数据，UDP 协议无法控制，所以说 UDP 是一种不可靠的协议。

UDP 协议适用于一次只传送少量数据、对可靠性要求不高的应用环境。

与TCP 协议一样，UDP 协议直接位于 IP 协议之上。实际上，IP 协议属于 OSI 参考模型的网络层协议，而 UDP 协议和 TCP 协议都属于传输层协议。

因为 UDP 是面向非连接的协议，没有建立连接的过程，因此它的通信效率很高，但也正因为如此，它的可靠性不如 TCP 协议。

UDP 协议的主要作用是完成网络数据流和数据报之间的转换在信息的发送端，UDP 协议将网络数据流封装成数据报，然后将数据报发送出去；在信息的接收端，UDP 协议将数据报转换成实际数据内容。

可以认为 UDP 协议的 socket 类似于码头，数据报则类似于集装箱。码头的作用就是负责友送、接收集装箱，而 socket 的作用则是发送、接收数据报。因此，对于基于 UDP 协议的通信双方而言，没有所谓的客户端和服务器端的概念。

3.源码示例

UDP的源码包含了我这边的整个系统架构层面的内容，并非完整的源码，可以根据自己的系统结构进行调整

3.1.启动UDP服务

  1. /**
  2.  * 网关数据监听
  3.  * @author lenny
  4.  * @date 20220314
  5.  */
  6. @Component
  7. public class ApplicationEventListener implements CommandLineRunner {
  8.     //因为我的网关服务是按终端协议节点进行区分的，当终端上下行数据都需要指定这个节点名称然后给到MQ
  9.     @Value("${spring.application.name}")
  10.     private String nodeName;
  11.     //UDP服务的监听端口
  12.     @Value("${gnss.udpserver.udpPort}")
  13.     private int udpPort;
  14.  
  15.     @Override
  16.     public void run(String... args) throws Exception {
  17.         //启动UDP服务
  18.         startUdpServer();
  19.  
  20.         //清除Redis所有此节点的在线终端（业务需要，Demo可以不用）
  21.         RedisService redisService = SpringBeanService.getBean(RedisService.class);
  22.         redisService.deleteAllOnlineTerminals(nodeName);
  23.  
  24.         //将所有此节点的终端设置为离线（业务需要，Demo可以不用）
  25.         RabbitMessageSender messageSender = SpringBeanService.getBean(RabbitMessageSender.class);
  26.         messageSender.noticeAllOffline(nodeName);
  27.     }
  28.  
  29.     /**
  30.      * 启动udp服务
  31.      *
  32.      * @throws Exception
  33.      */
  34.     private void startUdpServer() throws Exception {
  35.         //计数器,必须等到所有服务启动成功才能进行后续的操作
  36.         final CountDownLatch countDownLatch = new CountDownLatch(1);
  37.         //启动UDP服务
  38.         UdpServer udpServer = new UdpServer(udpPort, ProtocolEnum.UDP, countDownLatch);
  39.         udpServer.start();
  40.         //等待启动完成
  41.         countDownLatch.await();
  42.     }
  43. }

3.2.UdpServer类

  1. /**
  2.  * UDP服务类
  3.  * @author lenny
  4.  * @date 20220316
  5.  */
  6. @Slf4j
  7. public class UdpServer extends Thread{
  8.     private int port;
  9.  
  10.     private ProtocolEnum protocolType;
  11.  
  12.     private EventLoopGroup workerGroup;
  13.  
  14.     private Bootstrap bootstrap = new Bootstrap();
  15.  
  16.     private CountDownLatch countDownLatch;
  17.  
  18.     public UdpServer(int port, ProtocolEnum protocolType, CountDownLatch countDownLatch) {
  19.         this.port = port;
  20.         this.protocolType = protocolType;
  21.         this.countDownLatch = countDownLatch;
  22.         final EventExecutorGroup executorGroup = SpringBeanService.getBean("executorGroup", EventExecutorGroup.class);
  23.         workerGroup = SpringBeanService.getBean("workerGroup", EventLoopGroup.class);
  24.         bootstrap.group( workerGroup)
  25.                 .channel(NioDatagramChannel.class)
  26.                 .option(ChannelOption.SO_BROADCAST, true)
  27.                 .handler(new ChannelInitializer<NioDatagramChannel>(){
  28.                     @Override
  29.                     protected void initChannel(NioDatagramChannel  ch) throws Exception {
  30.                         ch.pipeline().addLast(new IdleStateHandler(UdpConstant.READER_IDLE_TIME, 0, 0, TimeUnit.SECONDS));
  31.                         //数据初步解析，并进行组包拆包处理
  32.                         ch.pipeline().addLast(new ProtocolDecoder(protocolType));
  33.                         //登录鉴权业务
  34.                         ch.pipeline().addLast(UdpLoginHandler.INSTANCE);
  35.                         //数据完全解码，并根据数据进行业务处理
  36.                         ch.pipeline().addLast(executorGroup, UdpBusinessHandler.INSTANCE);
  37.                     }
  38.                 });
  39.     }
  40.  
  41.     @Override
  42.     public void run() {
  43.         bind();
  44.     }
  45.  
  46.     private void bind(){
  47.         bootstrap.bind(port).addListener(future -> {
  48.             if (future.isSuccess()) {
  49.                 log.info("UDP服务器启动,端口:{}" , port);
  50.                 countDownLatch.countDown();
  51.             } else {
  52.                 log.error("UDP服务器启动失败,端口:{}", port, future.cause());
  53.                 System.exit(-1);
  54.             }
  55.         });
  56.     }
  57.  
  58. }

3.3.ProtocolDecoder数据预处理

  1. /**
  2.  * <p>Description: 常规消息体初步解析解释器</p>
  3.  *
  4.  * @author lenny
  5.  * @version 1.0.1
  6.  * @date 2022-03-16
  7.  */
  8. @Slf4j
  9. public class ProtocolDecoder extends MessageToMessageDecoder<DatagramPacket> {
  10.     /**
  11.      * 协议类型
  12.      */
  13.     private ProtocolEnum protocolType;
  14.  
  15.     public ProtocolDecoder(ProtocolEnum protocolType) {
  16.             this.protocolType=protocolType;
  17.     }
  18.  
  19.     @Override
  20.     protected void decode(ChannelHandlerContext ctx, DatagramPacket datagramPacket, List<Object> out) throws Exception {
  21.         ByteBuf in=datagramPacket.content();
  22.         log.info("收到:{}", ByteBufUtil.hexDump(in));
  23.         Object decoded = null;
  24.         //decodePacket数据粘包与半包处理
  25.         Object obj= PacketUtil.decodePacket(in);
  26.         if(obj!=null) {
  27.             //按协议进行数据初步解析，得到对应的设备ID与消息体，通过设备ID可以进行鉴权处理
  28.             decoded = decodeMessage((ByteBuf) obj);
  29.         } else {
  30.             return;
  31.         }
  32.         if (decoded != null) {
  33.             out.add(decoded);
  34.         }
  35.     }
  36.  
  37.     /**
  38.      * 预处理包（使用的是博实结的一款UDP通讯的产品）
  39.      * @param frame
  40.      * @return
  41.      */
  42.     private Object decodeMessage(ByteBuf frame) {
  43.         //将完整包输出为16进制字符串
  44.         String rowData=ByteBufUtil.hexDump(frame);
  45.         frame.readShort();
  46.         //信令
  47.         int msgId=frame.readUnsignedByte();
  48.         //消息长度（包长字节位置后的第一字节开始直到包尾的长度）
  49.         int bodyLen=frame.readUnsignedShort();
  50.         //伪IP
  51.         byte[] terminalNumArr = new byte[4];
  52.         frame.readBytes(terminalNumArr);
  53.         //终端ID
  54.         String terminalNum= CommonUtil.parseTerminalNum(terminalNumArr);
  55.         //剩余消息体
  56.         byte[] msgBodyArr = new byte[bodyLen-6];
  57.         frame.readBytes(msgBodyArr);
  58.         //校验—+包尾
  59.         frame.readShort();
  60.  
  61.         //处理成统一标准给到处理器去处理后续的解析
  62.         UdpMessage message=new UdpMessage();
  63.         message.setTerminalNumArr(terminalNumArr);
  64.         message.setTerminalNum(terminalNum);
  65.         message.setMsgId(msgId);
  66.         message.setProtocolType(protocolType);
  67.         message.setMsgBodyArr(msgBodyArr);
  68.         message.setMsgBody(Unpooled.wrappedBuffer(msgBodyArr));
  69.         return message;
  70.     }
  71. }

3.4.登录鉴权

我这边是通过当前设备是否在系统注册进行鉴权处理的

  1. /**
  2.  * 登录业务处理
  3.  * @author Lenny
  4.  * @date 20200314
  5.  */
  6. @Slf4j
  7. @ChannelHandler.Sharable
  8. public class UdpLoginHandler extends SimpleChannelInboundHandler<UdpMessage> {
  9.     public static final UdpLoginHandler INSTANCE = new UdpLoginHandler();
  10.  
  11.     private RedisService redisService;
  12.  
  13.     private RabbitMessageSender messageSender;
  14.  
  15.     private Environment environment;
  16.  
  17.     private UdpLoginHandler() {
  18.         redisService = SpringBeanService.getBean(RedisService.class);
  19.         messageSender = SpringBeanService.getBean(RabbitMessageSender.class);
  20.         environment = SpringBeanService.getBean(Environment.class);
  21.     }
  22.  
  23.     @Override
  24.     protected void channelRead0(ChannelHandlerContext ctx, UdpMessage msg) throws Exception {
  25.         //协议类型
  26.         ProtocolEnum protocolType = msg.getProtocolType();
  27.         //查询终端号码有无在平台注册
  28.         String terminalNum = msg.getTerminalNum();
  29.         TerminalProto terminalInfo = redisService.getTerminalInfoByTerminalNum(terminalNum);
  30.         if (terminalInfo == null) {
  31.             log.error("终端登录失败,未找到终端信息,协议:{},终端号:{},消息:{}", protocolType, terminalNum, msg);
  32.             if (msg.getMsgBody() != null) {
  33.                 ReferenceCountUtil.release(msg.getMsgBody());
  34.             }
  35.             ctx.close();
  36.             return;
  37.         }
  38.  
  39.         //设置协议类型
  40.         terminalInfo.setProtocolType(protocolType);
  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:{},终端号:{}", protocolType, terminalInfo.getTerminalStrId(), terminalNum);
  48.         } else {
  49.             log.info("终端重复登录关闭上一个连接,协议:{},终端ID:{},终端号:{}", protocolType, terminalInfo.getTerminalStrId(), terminalNum);
  50.             oldCtx.close();
  51.         }
  52.         //通知上线
  53.         messageSender.noticeOnline(terminalInfo);
  54.  
  55.         //登录验证通过后移除此handler
  56.         ctx.pipeline().remove(this);
  57.         ctx.fireChannelRead(msg);
  58.     }
  59.  
  60.     @Override
  61.     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
  62.     }
  63.  
  64.     @Override
  65.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  66.         ctx.close();
  67.     }
  68. }

3.5.业务处理

  1. /**
  2.  * 业务处理
  3.  * @author lenny
  4.  * @date 20220314
  5.  */
  6. @Slf4j
  7. @Sharable
  8. public class UdpBusinessHandler extends SimpleChannelInboundHandler<UdpMessage> {
  9.     public static final UdpBusinessHandler INSTANCE = new UdpBusinessHandler();
  10.  
  11.     private RabbitMessageSender messageSender;
  12.  
  13.     private MessageServiceProvider messageServiceProvider;
  14.  
  15.     private UdpBusinessHandler() {
  16.         messageSender = SpringBeanService.getBean(RabbitMessageSender.class);
  17.         messageServiceProvider = SpringBeanService.getBean(MessageServiceProvider.class);
  18.     }
  19.  
  20.     @Override
  21.     protected void channelRead0(ChannelHandlerContext ctx, UdpMessage msg) throws Exception {
  22.         //根据消息类型获取对应的消息处理器
  23.         int msgId = msg.getMsgId();
  24.         //找到对应的业务处理器
  25.         BaseMessageService messageService = messageServiceProvider.getMessageService(msgId);
  26.         ByteBuf msgBody = msg.getMsgBody();
  27.         try {
  28.             //通过业务处理器进行处理
  29.             messageService.process(ctx, msg, msgBody);
  30.         } catch (Exception e) {
  31.             printExceptionLog(msg, messageService, e);
  32.         } finally {
  33.             ReferenceCountUtil.release(msgBody);
  34.         }
  35.     }
  36.     /**
  37.      * 打印异常日志
  38.      *
  39.      * @param msg
  40.      * @param messageService
  41.      * @param e
  42.      */
  43.     private void printExceptionLog(UdpMessage msg, BaseMessageService messageService, Exception e) {
  44.         byte[] msgBodyArr = msg.getMsgBodyArr();
  45.         log.error("收到{}({}),消息异常,协议:{},终端号码:{},消息体:{}", messageService.getDesc(), NumberUtil.formatMessageId(msg.getMsgId())
  46.                 , msg.getProtocolType(), msg.getTerminalNum(), ByteBufUtil.hexDump(msgBodyArr), e);
  47.     }
  48.  
  49.     @Override
  50.     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
  51.         Session session = SessionUtil.getSession(ctx.channel());
  52.         TerminalProto terminalInfo = session.getTerminalInfo();
  53.         boolean unbindResult = SessionUtil.unbindSession(ctx);
  54.         if (unbindResult) {
  55.             //通知离线
  56.             messageSender.noticeOffline(terminalInfo);
  57.         }
  58.     }
  59.  
  60.     @Override
  61.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  62.         TerminalProto terminalInfo = SessionUtil.getTerminalInfo(ctx);
  63.         log.error("终端连接异常,终端ID:{},终端号码:{}", terminalInfo.getTerminalStrId(), terminalInfo.getTerminalNum(), cause);
  64.         ctx.close();
  65.     }
  66.  
  67.     @Override
  68.     public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
  69.         if (evt instanceof IdleStateEvent) {
  70.             TerminalProto terminalInfo = SessionUtil.getTerminalInfo(ctx);
  71.             if (terminalInfo == null) {
  72.                 log.info("{}秒未读到数据,关闭连接", UdpConstant.READER_IDLE_TIME);
  73.             } else {
  74.                 log.info("{}秒未读到数据,关闭连接,终端ID:{},终端号码:{}", UdpConstant.READER_IDLE_TIME, terminalInfo.getTerminalStrId()
  75.                         , terminalInfo.getTerminalNum());
  76.             }
  77.             ctx.close();
  78.         }
  79.     }
  80. }

3.6.业务处理器举例

比如我们收到一个消息ID为0x81的数据

  1. /**
  2.  * 点名查看
  3.  * @author Lenny
  4.  * @date 20220314
  5.  */
  6. @Slf4j
  7. @MessageService(messageId = 0x81, desc = "点名查看")
  8. public class Message81Service extends BaseMessageService<UdpMessage> {
  9.  
  10.     @Autowired
  11.     private RabbitMessageSender messageSender;
  12.  
  13.     @Override
  14.     public Object process(ChannelHandlerContext ctx, UdpMessage msg, ByteBuf msgBodyBuf) throws Exception {
  15.         //这里需要对业务进行实际解码，目前并未处理，目的是先验证整个流程的通畅性
  16.         log.info(JSON.toJSONString(msg));
  17.         return null;
  18.     }
  19. }

4.说明

目前我这边已经利用了Netty完成了TCP、MQTT、UDP的通讯服务层，集成了近百款终端协议产品，包含了部标全套的服务，以及第三方企业标准标准几十家，有兴趣的朋友可以联系我。

请添加图片描述

 

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

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

  

闽ICP备14008679号