Netty 系列九（支持UDP协议）.

2018-09-18 06:35:50来源：博客园阅读 ()

一、基础知识

UDP 协议相较于 TCP 协议的特点：

1、无连接协议，没有持久化连接；
2、每个 UDP 数据报都是一个单独的传输单元；
3、一定的数据报丢失；
4、没有重传机制，也不管数据报是否可达；
5、速度比TCP快很多，可用来高效处理大量数据 —— 牺牲了握手以及消息管理机制。
6、常用于音频、视频场景，可以忍受一定的数据包丢失，追求速度上的提升。

TCP 协议采用的是一种叫做单播的传输形式，UDP 协议提供了向多个接收者发送消息的额外传输形式（多播、广播）：

单播（TCP 和 UDP）：发送消息给一个由唯一的地址所标识的单一的网络目的地。
多播（UDP）：传输给一个预定义的主机组。
广播（UDP）：传输到网络（或者子网）上的所有主机。

二、功能说明

广播方：打开一个文件，通过 UDP 使用特殊的受限广播地址或者零网络地址 255.255.255.255，把每一行作为一个消息广播到一个指定的端口。

接收方：通过 UDP 广播，只需简单地通过在指定的端口上启动一个监听程序，便可以创建一个事件监视器来接收日志消息。所有的在该 UDP 端口上监听的事件监听器都将会接收到广播信息。

三、实现

下图展示了怎么将我们的文件数据广播为 UDP消息：所有的将要被传输的数据都被封装在了 LogEvent 消息中。 LogEventBroadcaster 将把这些写入到 Channel 中，并通过 ChannelPipeline 发送它们，在那里它们将会被转换（编码）为 DatagramPacket 消息。最后，他们都将通过 UDP 被广播，并由远程节点（监视器）所捕获。

Netty 中支持 UDP 协议主要通过以下相关类：

DatagramPacket：使用 ByteBuf 作为数据源，是 UDP 协议传输的消息容器。

DatagramChannel：扩展了 Netty 的 Channel 抽象以支持 UDP 的多播组管理，它的实现类 NioDatagramChannnel 用来和远程节点通信。

Bootstrap：UDP 协议的引导类，使用 bind() 方法绑定 Channel。

public class LogEvent {
    public static final byte SEPARATOR = ':';
    /**
     * IP套接字地址（IP地址+端口号）
     */
    private final InetSocketAddress inetSocketAddress;
    /**
     * 文件名
     */
    private final String logfile;
    /**
     * 消息内容
     */
    private final String msg;
    
    private final long received;

    /**
     * 用于传入消息的构造函数
     *
     * @param inetSocketAddress
     * @param logfile
     * @param msg
     * @param received
     */
    public LogEvent(InetSocketAddress inetSocketAddress, String logfile, String msg, long received) {
        this.inetSocketAddress = inetSocketAddress;
        this.logfile = logfile;
        this.msg = msg;
        this.received = received;
    }

    /**
     * 用于传出消息的构造函数
     *
     * @param logfile
     * @param msg
     */
    public LogEvent(String logfile, String msg) {
        this(null, logfile, msg, -1);
    }

    public InetSocketAddress getInetSocketAddress() {
        return inetSocketAddress;
    }

    public String getLogfile() {
        return logfile;
    }

    public String getMsg() {
        return msg;
    }

    public long getReceived() {
        return received;
    }
}

文件实体类 LogEvent.java

public class LogEventEncoder extends MessageToMessageEncoder<LogEvent> {
    private final InetSocketAddress remoteAddress;

    public LogEventEncoder(InetSocketAddress remoteAddress) {
        this.remoteAddress = remoteAddress;
    }


    @Override
    protected void encode(ChannelHandlerContext ctx, LogEvent msg, List<Object> out) throws Exception {
        byte[] file = msg.getLogfile().getBytes(CharsetUtil.UTF_8);
        byte[] content = msg.getMsg().getBytes(CharsetUtil.UTF_8);
        ByteBuf byteBuf = ctx.alloc().buffer(file.length + content.length + 1);
        byteBuf.writeBytes(file);
        byteBuf.writeByte(LogEvent.SEPARATOR);
        byteBuf.writeBytes(content);
        out.add(new DatagramPacket(byteBuf, remoteAddress));
    }
}

编码器 LogEventEncoder.java

该编码器实现了将 LogEvent 实体类内容转换为 DatagramPacket UDP数据报。

public class LogEventBroadcaster {
    private final EventLoopGroup group;
    private final Bootstrap bootstrap;
    private final File file;

    public LogEventBroadcaster(InetSocketAddress address, File file) {
        group = new NioEventLoopGroup();
        bootstrap = new Bootstrap();
        bootstrap.group(group)
                //引导该 NioDatagramChannel（无连接的）
                .channel(NioDatagramChannel.class)
                // 设置 SO_BROADCAST 套接字选项
                .option(ChannelOption.SO_BROADCAST, true)
                .handler(new LogEventEncoder(address));
        this.file = file;
    }
    
    public void run() throws InterruptedException, IOException {
        //绑定 Channel，UDP 协议的连接用 bind() 方法
        Channel channel = bootstrap.bind(0).sync().channel();
        long pointer = 0;
        //长轮询 监听是否有新的日志文件生成
        while (true) {
            long length = file.length();
            if (length < pointer) {
                // 如果有必要，将文件指针设置到该文件的最后一个字节
                pointer = length;
            } else {
                RandomAccessFile raf = new RandomAccessFile(file, "r");
                // 确保当前的文件指针，以确保没有任何的旧数据被发送
                raf.seek(pointer);
                String line;
                while ((line = raf.readLine()) != null) {
                    //对于每个日志条目，写入一个 LogEvent 到 Channel 中，最后加入一个换行符号
                    channel.writeAndFlush(new LogEvent(file.getAbsolutePath(), line + System.getProperty("line.separator")));
                }
                pointer = raf.getFilePointer();
                raf.close();
            }
            try {
                // 休眠一秒，如果被中断，则退出循环，否则重新处理它
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                while (!Thread.interrupted()) {
                    break;
                }
            }
        }
    }

    public void stop() {
        group.shutdownGracefully();
    }

    public static void main(String[] args) throws IOException, InterruptedException {
        InetSocketAddress socketAddress = new InetSocketAddress("255.255.255.255", 8888);
        File file = new File("E:\\2018-09-12.log");
        LogEventBroadcaster logEventBroadcaster = new LogEventBroadcaster(socketAddress, file);
        try {
            logEventBroadcaster.run();
        } finally {
            logEventBroadcaster.stop();
        }
    }
}

现在，我们来测试一下这个 UDP 广播类，首先我们需要一个工具 nmap ，用它来监听 UDP 的 8888 端口，以接收我们广播的日志文件。下载地址： https://nmap.org/dist/nmap-7.70-win32.zip

下载完成后，命令行进入安装目录，执行命令：ncat.exe -l -u -p 8888 ，监听 UDP 端口。

当然，也可以自己写个测试类监听 UDP 端口，打印日志查看。这里我没有用 Netty 写监听类，直接用了 java 原生的 DatagramSocket 和 DatagramPacket 写的监听类，如下：

public class UDPServer {

    public static void main(String[] args) {
        DatagramSocket server = null;
        try {
            server = new DatagramSocket(8888);
            byte[] datas = new byte[1024];
            //用一个字节数组接收UDP包，字节数组在传递给构造函数时是空的
            while (true) {
                DatagramPacket datagramPacket = new DatagramPacket(datas, datas.length);
                server.receive(datagramPacket);
                System.out.println(new String(datas));
            }
        } catch (SocketException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            server.close();
        }
    }
}