Netty相关的UDP 协议（十四）

傷城~ 2022-10-29 13:27 452阅读 0赞

今天我们分析Netty相关的UDP 协议：

**一、什么事UDP**  **协议**

UDP  是面向无连接的通讯协议， UDP  报头由  4  个域组成，其中每个域各占用  2  个字节， 其中包括目的端口号和源端口号信息，数据报的长度域是指包括报头和数据部分在内的总字节数，校验值域来保证数据的安全。由于通讯不需要连接，所以可以实现广播发送。 UDP 通讯时不需要接收方确认，属于不可靠的传输，可能会出现丢包现象，实际应用中要求程序员编程验证。 UDP 与  TCP  位于同一层，但它不管数据包的顺序、错误或重发。因此， UDP  不被应用于那些使用虚电路的面向连接的服务，UDP  主要用于那些面向查询 \--- 应答的服务，例如  NFS 。相对于 FTP  或  Telnet ，这些服务需要交换的信息量较小。

使用  UDP  的服务包括：

1，  NTP （网络时间协议）和 DNS （ DNS  也使用  TCP ），包总量较少的通信（ DNS 、 SNMP  等）；

2. 视频、音频 等多媒体通信（即时通信）；

3. 限定于  LAN  等特定网络中的应用通信；

4. 广播通信（广播、多播）。常用的 QQ ，就是一个以  UDP  为主， TCP  为辅的通讯协议。

TCP 和  UDP  的优缺点无法简单地、绝对地去做比较： TCP  用于在传输层有必要实现可靠传输的情况；而在一方面，UDP  主要用于那些对高速传输和实时性有较高要求的通信或广播通信。TCP  和  UDP  应该根据应用的目的按需使用。

**二、实现**  **UDP**  **单播和广播**

UDP  这样的无连接协议中，并没有持久化连接这样的概念，并且每个消息（一个  UDP  数据报）都是一个单独的传输单元。此外，UDP  也没有  TCP  的纠错机制。

通过类比， TCP  连接就像打电话，其中一系列的有序消息将会在两个方向上流动。相反， UDP 则类似于往邮箱中投入一叠明信片。你无法知道它们将以何种顺序到达它们的目的地，或者它们是否所有的都能够到达它们的目的地。

UDP 的这些方面可能会让你感觉到严重的局限性，但是它们也解释了为何它会比  TCP快那么多：所有的握手以及消息管理机制的开销都已经被消除了。显然，UDP  很适合那些能够处理或者容忍消息丢失的应用程序，但可能不适合那些处理金融交易的应用程序。

本身作为无连接的不可靠的传输协议（适合频繁发送较小的数据包），他不会对数据包进行合并发送（也就没有 Nagle  算法之说了），他直接是一端发送什么数据，直接就发出去了，既然他不会对数据合并，每一个数据包都是完整的（数据+UDP  头 \+IP  头等等发一次数据封装一次）也就**没有粘包** 一说了。

单播的传输模式，定义为发送消息给一个由唯一的地址所标识的单一的网络目的地。面向连接的协议和无连接协议都支持这种模式。

广播——传输到网络（或者子网）上的所有主机。

**三、Netty**  **的**  **UDP**  **相关类**

interface  **AddressedEnvelope** <M, A extends SocketAddress>extends ReferenceCounted

定义一个消息，其包装了另一个消息并带有发送者和接收者地址。其中  M  是消息类型；A 是地址类型。

class  **DefaultAddressedEnvelope** <M, A extends SocketAddress>implements AddressedEnvelope<M,A>提供了  interface AddressedEnvelope  的默认实现

class  **DatagramPacket**  extends DefaultAddressedEnvelope<ByteBuf, InetSocketAddress> implements ByteBufHolder

扩展了  DefaultAddressedEnvelope  以使用  ByteBuf  作为消息数据容器。 DatagramPacket 是 final  类不能被继承，只能被使用。

**通过**  **content()** **来获取消息内容**

**通过**  **sender();** **来获取发送者的消息**

**通过**  **recipient();** **来获取接收者的消息。**

interface  **DatagramChannel**  extends Channel 扩展了 Netty  的  Channel  抽象以支持  UDP  的多播组管理 class  **NioDatagramChannnel**  extends AbstractNioMessageChannel implements DatagramChannel 定义了一个能够发送和接收 Addressed-Envelope  消息的  Channel  类型Netty 的  DatagramPacket  是一个简单的消息容器， DatagramChannel  实现用它来和远程节点通信。类似于在我们先前的类比中的明信片，它包含了接收者（和可选的发送者）的地址以及消息的有效负载本身。

**四、UDP**  **单播 测试**

1、应答端：

public class UdpAnswerSide {
    
        public static final int ANSWER_PORT = 8080;
        public final static String ANSWER = "古诗来了：";
    
        public void run(int port) throws Exception{
            EventLoopGroup group = new NioEventLoopGroup();
            try {
                /*udp无连接，没有接受连接说法,即使是接受端，也应该用Bootstrap*/
                Bootstrap b = new Bootstrap();
                /*由于我们用的是UDP协议，所以要用NioDatagramChannel来创建*/
                b.group(group)
                    .channel(NioDatagramChannel.class)
                    .handler(new AnswerHandler());
                //没有接受客户端连接的过程，监听本地端口即可
                ChannelFuture f = b.bind(port).sync();
                System.out.println("应答服务已启动.....");
                f.channel().closeFuture().sync();
            } finally {
                group.shutdownGracefully();
            }
        }
        public static void main(String [] args) throws Exception{
            new UdpAnswerSide().run(ANSWER_PORT);
        }
    }

2、应答端业务代码：

public class AnswerHandler extends
            SimpleChannelInboundHandler<DatagramPacket> {
    
        /*应答的具体内容从常量字符串数组中取得，由nextQuote方法随机获取*/
        private static final String[] DICTIONARY = {
                "只要功夫深，铁棒磨成针。",
                "旧时王谢堂前燕,飞入寻常百姓家。",
                "洛阳亲友如相问，一片冰心在玉壶。",
                "一寸光阴一寸金，寸金难买寸光阴。",
                "老骥伏枥，志在千里，烈士暮年，壮心不已" };
        private static Random r = new Random();
        private String nextQuote(){
            return DICTIONARY[r.nextInt(DICTIONARY.length-1)];
        }
    
        @Override
        protected void channelRead0(ChannelHandlerContext ctx,
                                    DatagramPacket packet)
                throws Exception {
    
            /*获得请求*/
            String req = packet.content().toString(CharsetUtil.UTF_8);
            System.out.println("接收到请求："+req);
            if(UdpQuestionSide.QUESTION.equals(req)){
                String answer = UdpAnswerSide.ANSWER+nextQuote();
                System.out.println("接收到请求："+req);
                /**
                 * 重新 new 一个DatagramPacket对象，
                 * 我们通过packet.sender()来获取发送者的消息。重新发送出去！
                 */
                 ctx.writeAndFlush(
                        new DatagramPacket(
                                Unpooled.copiedBuffer(answer,CharsetUtil.UTF_8),
                                packet.sender()
                        )
                );
            }
        }
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
                throws Exception {
            ctx.close();
            cause.printStackTrace();
        }
    }

3、请求端：

public class UdpQuestionSide {
    
        public final static String QUESTION = "告诉我一句古诗";
    
        public void run(int port) throws Exception{
    
            EventLoopGroup group  = new NioEventLoopGroup();
            try {
                Bootstrap b = new Bootstrap();
                b.group(group)
                        .channel(NioDatagramChannel.class)/*UDP通信*/
                .handler(new QuestoinHandler());
                Channel channel = b.bind(0).sync().channel();
                channel.writeAndFlush(
                        new DatagramPacket(
                                Unpooled.copiedBuffer(QUESTION,CharsetUtil.UTF_8),
                                new InetSocketAddress("127.0.0.1",port)
                        )
                ).sync();
                if(!channel.closeFuture().await(15000)){
                    System.out.println("等待超时");
                }
            } catch (Exception e) {
                group.shutdownGracefully();
            }
        }
        public static void main(String [] args) throws Exception{
            new UdpQuestionSide().run(UdpAnswerSide.ANSWER_PORT);
        }
    }

4、请求端业务：

public class QuestoinHandler extends
            SimpleChannelInboundHandler<DatagramPacket> {
    
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, DatagramPacket msg)
                throws Exception {
            //获得应答，DatagramPacket提供了content()方法取得报文的实际内容
            String response = msg.content().toString(CharsetUtil.UTF_8);
            if (response.startsWith(UdpAnswerSide.ANSWER)) {
                System.out.println(response);
                ctx.close();
            }
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
                throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    }

5、测试执行

应答端执行：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70][]

请求端执行：

![20210219164835894.png][]

**五、UDP 广播测试：**

1、广播端核心：

public class LogEventBroadcaster {
        private final EventLoopGroup group;
        private final Bootstrap bootstrap;
    
        public LogEventBroadcaster(InetSocketAddress remoteAddress) {
            group = new NioEventLoopGroup();
            bootstrap = new Bootstrap();
            //引导该 NioDatagramChannel（无连接的）
            bootstrap.group(group).channel(NioDatagramChannel.class)
                    /*设置通信选型为广播模式*/
                 .option(ChannelOption.SO_BROADCAST,true)
                 .handler(new LogEventEncoder(remoteAddress));
        }
    
        public void run() throws Exception {
            //绑定 Channel
            Channel ch = bootstrap.bind(0).sync().channel();
            long count = 0;
            //启动主处理循环，模拟日志发送
            for (;;) {
                ch.writeAndFlush(new LogMsg(null, ++count,
                        LogConst.getLogInfo()));
                try {
                    //休眠 2 秒，如果被中断，则退出循环；
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    Thread.interrupted();
                    break;
                }
            }
        }
    
        public void stop() {
            group.shutdownGracefully();
        }
    
        public static void main(String[] args) throws Exception {
    
            //创建并启动一个新的 UdpQuestionSide 的实例
            LogEventBroadcaster broadcaster = new LogEventBroadcaster(
                    new InetSocketAddress("255.255.255.255",
                            LogConst.MONITOR_SIDE_PORT)
            );
            try {
                broadcaster.run();
            }
            finally {
                broadcaster.stop();
            }
        }
    }

2、广播端编码：

public class LogEventEncoder extends MessageToMessageEncoder<LogMsg> {
        private final InetSocketAddress remoteAddress;
    
        //LogEventEncoder 创建了即将被发送到指定的 InetSocketAddress
        // 的 DatagramPacket 消息
        public LogEventEncoder(InetSocketAddress remoteAddress) {
            this.remoteAddress = remoteAddress;
        }
    
        @Override
        protected void encode(ChannelHandlerContext channelHandlerContext,
                              LogMsg logMsg, List<Object> out) throws Exception {
            byte[] bytes = logMsg.getMsg().getBytes(CharsetUtil.UTF_8);
            ByteBuf buf = channelHandlerContext.alloc().buffer(
                    8*2+bytes.length+1);
            buf.writeLong(logMsg.getTime());
            buf.writeLong(logMsg.getMsgId());
            buf.writeByte(LogMsg.SEPARATOR);
            buf.writeBytes(bytes);
            out.add(new DatagramPacket(buf,remoteAddress));
        }
    }

3、接收端监听核心：

public class LogEventMonitor {
        private final EventLoopGroup group;
        private final Bootstrap bootstrap;
    
        public LogEventMonitor(InetSocketAddress address) {
            group = new NioEventLoopGroup();
            bootstrap = new Bootstrap();
            //引导该 NioDatagramChannel
            bootstrap.group(group)
                .channel(NioDatagramChannel.class)
                //设置套接字选项 SO_BROADCAST
                .option(ChannelOption.SO_BROADCAST, true)
    
                .option(ChannelOption.SO_REUSEADDR, true)
    
                .handler( new ChannelInitializer<Channel>() {
                    @Override
                    protected void initChannel(Channel channel)
                        throws Exception {
                        ChannelPipeline pipeline = channel.pipeline();
                        pipeline.addLast(new LogEventDecoder());
                        pipeline.addLast(new LogEventHandler());
                    }
                } )
                .localAddress(address);
        }
    
        public Channel bind() {
            //绑定 Channel。注意，DatagramChannel 是无连接的
            return bootstrap.bind().syncUninterruptibly().channel();
        }
    
        public void stop() {
            group.shutdownGracefully();
        }
    
        public static void main(String[] args) throws Exception {
            //构造一个新的 UdpAnswerSide并指明监听端口
            LogEventMonitor monitor = new LogEventMonitor(
                new InetSocketAddress(LogConst.MONITOR_SIDE_PORT));
            try {
                //绑定本地监听端口
                Channel channel = monitor.bind();
                System.out.println("Udp AnswerSide running");
                channel.closeFuture().sync();
            } finally {
                monitor.stop();
            }
        }
    }

4、接收端业务代码：

public class LogEventHandler
        extends SimpleChannelInboundHandler<LogMsg> {
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx,
                                    Throwable cause) throws Exception {
            //当异常发生时，打印栈跟踪信息，并关闭对应的 Channel
            cause.printStackTrace();
            ctx.close();
        }
    
        @Override
        public void channelRead0(ChannelHandlerContext ctx,
                                 LogMsg event) throws Exception {
            //创建 StringBuilder，并且构建输出的字符串
            StringBuilder builder = new StringBuilder();
            builder.append(event.getTime());
            builder.append(" [");
            builder.append(event.getSource().toString());
            builder.append("] ：[");
            builder.append(event.getMsgId());
            builder.append("] ：");
            builder.append(event.getMsg());
            //打印 LogMsg 的数据
            System.out.println(builder.toString());
        }
    }

5、接收端解码：

public class LogEventDecoder extends MessageToMessageDecoder<DatagramPacket> {
    
        @Override
        protected void decode(ChannelHandlerContext ctx,
                              DatagramPacket datagramPacket, List<Object> out)
            throws Exception {
            ByteBuf data = datagramPacket.content();
            //获得发送时间
            long sendTime = data.readLong();
            System.out.println("接受到"+sendTime+"发送的消息");
            //获得消息的id
            long msgId = data.readLong();
            //获得分隔符SEPARATOR
            byte sepa = data.readByte();
            //获取读索引的当前位置，就是分隔符的索引+1
            int idx = data.readerIndex();
            //提取日志消息，从读索引开始，到最后为日志的信息
            String sendMsg = data.slice(idx ,
                    data.readableBytes()).toString(CharsetUtil.UTF_8);
            //构建一个新的 LogMsg 对象，并且将它添加到（已经解码的消息的）列表中
            LogMsg event = new LogMsg(datagramPacket.sender(),
                    msgId, sendMsg);
            //作为本handler的处理结果，交给后面的handler进行处理
            out.add(event);
        }
    }

6、启动执行：

广播端启动后

接收端：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1][]

**六、UDT 总结**

基于  UDP  的数据传输协议（ UDP-based Data Transfer Protocol ，简称  UDT ）是一种互联网数据传输协议。UDT  的主要目的是支持 **高速广域网上的海量数据传输** ，最典型的例子就是建立在光纤广域网上的网格计算，一些研究所在这样的网络上运行他们的分布式的数据密集程式，例如，远程访问仪器、分布式数据挖掘和高分辨率的多媒体流。

而互联网上的标准数据传输协议  TCP  在高带宽长距离网络上性能很差。顾名思义， UDT建于 UDP  之上，并引入新的拥塞控制和数据可靠性控制机制。 UDT  是面向连接的双向的应用层协议。

UDT  的特性主要包括在以下几个方面：

1、基于  UDP  的应用层协议： 有基本网络知识的朋友都知道  TCP  和  UDP  的区别和使用场景，但是有没有一种协议能同时兼顾 TCP  协议的安全可靠和  UDP  协议的高效，那么  UDT  就是一种。

2、面向连接的协议：面向连接意味着两个使用协议的应用在彼此交换数据之前必须先建立一个连接，当然 UDT  是逻辑上存在的连接通道。这种连接的维护是基于握手、 Keep-alive （保活）以及关闭连接。

3、可靠的协议：依靠包序号机制、接收者的  ACK  响应和丢包报告、 ACK  序号机制、重传机制( 基于丢包报告和超时处理 ) 来实现数据传输的可靠性。

4、双工的协议：每个  UDT  实例包含发送端和接收端的信息。 新的拥塞算法，并且具有可扩展

5、新的拥塞控制框架：新的拥塞控制算法不同于基于窗口的TCP 拥塞控制算法 ( 慢启动和拥塞避免 ) ，是混合的基于窗口的、基于速率的拥塞控制算法。 可扩展的拥塞控制框架开源的代码和拥塞控制的 C++ 类架构，可支持开发者派生专用的拥塞控制算法。

6、带宽估计： UDT  使用对包 (PP -- Packet pair) 的机制来估计带宽值。即每  16  个包为一组 , 最后一个是对包, 即发送方不用等到下一个发送周期内再发送。接收方接收到对包后对其到达时间进行记录, 可结合上次记录的值计算出链路的带宽 ( 计算的方法称为中值过滤法 ),  并在下次 ACK  中进行反馈。

到此，Netty相关的UDP 协议分享完毕，下篇分享Netty相关的服务器推送技术，敬请期待！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70]: /images/20221024/622ded3ba880442689ed6a1fbc38aacb.png
[20210219164835894.png]: /images/20221024/8c9a3e039b0242c3b21b01fdd7270e76.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1]: /images/20221024/934d3fdbe9a64d7bbc46ad7cadee52ca.png