netty构建udp服务器以及发送报文到客户端客户端详细案例_netty udp

目录

一、基于netty创建udp服务端以及对应通道设置关键

二、发送数据

三、netty中的ChannelOption常用参数说明

1、ChannelOption.SO_BACKLOG

2、ChannelOption.SO_REUSEADDR

3、ChannelOption.SO_KEEPALIVE

4、ChannelOption.SO_SNDBUF和ChannelOption.SO_RCVBUF

5、ChannelOption.SO_LINGER

6、ChannelOption.TCP_NODELAY

---

一、基于netty创建udp服务端以及对应通道设置关键

@Configuration
@RefreshScope
public class NettyUdpServer {
 
    @Value("${netty.server.udpPort}")
    private int port;
 
    private EventLoopGroup bossGroup;//主线程
 
    private Channel channel;//通道
 
    private ChannelFuture future; //回调
 
    @Autowired
    private DataCollector dataCollector;;
 
    public Channel start() throws InterruptedException {
        //判断是否支持Epoll模式,从而创建不同的线程组
        bossGroup = Epoll.isAvailable()? new EpollEventLoopGroup() : new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            //linux平台下增加SO_REUSEPORT特性提高性能，支持多个进程或者线程绑定到同一个端口，提高服务器程序的吞吐性能
            if(Epoll.isAvailable()) {
                //设置反应器线程组
                b.group(bossGroup)
                 .handler(new EpollUdpServerInitializer(dataCollector))
                //设置nio类型的通道
                 .channel(EpollDatagramChannel.class)
                 .option(ChannelOption.SO_BROADCAST, true)
                 .option(ChannelOption.SO_REUSEADDR, true)
                 .option(ChannelOption.SO_RCVBUF, 1024 * 1024)
                 .option(EpollChannelOption.SO_REUSEPORT, true);
            }else{
                //设置反应器线程组
                b.group(bossGroup)
                 .handler(new UdpServerInitializer(dataCollector))
                 //设置nio类型的通道
                 .channel(NioDatagramChannel.class)
                 //设置通道的参数
                 .option(ChannelOption.SO_BROADCAST, true)
                 .option(ChannelOption.SO_REUSEADDR, true);
            }
            //Channel channel = server.bind(port).sync().channel();
            //开始绑定服务器，通过调用sync()同步方法阻塞直到绑定成功
            //ChannelFuture channelFuture = b.bind(port).sync();
            //等待通道关闭的异步任务结束
            //ChannelFuture closeFuture = channelFuture.channel().closeFuture();
            //closeFuture.sync();
 
            ChannelFuture f = b.bind(port).sync();
            channel = f.channel();
            if(f.isSuccess()){
                //MasterSelector registry = new MasterSelector("","netty-services",  port);
                System.out.println("UDP服务器启动，监听在端口：" + port);
            }else {
                channel.closeFuture().sync();
            }
 
        } finally {
            //bossGroup.shutdownGracefully().sync();
        }
        System.out.println("Udp服务器启动，监听在端口："+port);
        return channel;
    }
}

以上代码中Epoll.isAvailable()用户判断是window还是linux环境，linux环境默认采用Epoll相关通道，所以显式设置EpollDatagramChannel通道。在处理（handler）的设置中要根据不同的通道设置初始化的通道类型：

linux环境下EpollDatagramChannel通道设置 .handler(new EpollUdpServerInitializer(dataCollector))具体代码

public class EpollUdpServerInitializer extends ChannelInitializer<EpollDatagramChannel> {
    private final DataCollector dataCollector;
 
    public EpollUdpServerInitializer(DataCollector dataCollector) {
        this.dataCollector = dataCollector;
    }
 
    @Override
    protected void initChannel(EpollDatagramChannel epollDatagramChannel) throws Exception {
        epollDatagramChannel.pipeline()
                //添加netty空闲超时检查的支持
                .addLast(new UdpServerHandler(dataCollector));
    }

要使 通过服务器端通过EpollDatagramChannel通道发送数据，客户端能够正常接收到数据，下图中标红的泛型通道类要与服务器端设置的通道类一致

同意要支持Nio类型通道为NioDatagramChannel.class时，通道初始化为：

public class UdpServerInitializer extends ChannelInitializer<NioDatagramChannel> {
    private final DataCollector dataCollector;
 
    public UdpServerInitializer(DataCollector dataCollector) {
        this.dataCollector = dataCollector;
    }
 
    @Override
    protected void initChannel(NioDatagramChannel nioDatagramChannel) throws Exception {
        nioDatagramChannel.pipeline()
                //添加netty空闲超时检查的支持
                .addLast(new UdpServerHandler(dataCollector));
    }
}

 要使 通过服务器端通过NioDatagramChannel通道发送数据，客户端能够正常接收到数据，下图中标红的泛型通道类要与服务器端设置的通道类一致

二、发送数据

关键代码，采用writeAndFlush发送数据，注意：要发送udp数据报，

public class UdpServerHandler extends SimpleChannelInboundHandler<DatagramPacket> {
 
    /**设置最大消息大小*/
    private static final int MAX_MESSAGE_SIZE = 2048;
    /**线程池*/
    private ExecutorService executorService;
 
    private final DataCollector dataCollector;
 
    public UdpServerHandler(DataCollector dataCollector) {
        this.dataCollector = dataCollector;
        //根据当前系统可用的处理器数量创建一个固定长度的线程池
        executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
    }
 
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, DatagramPacket datagramPacket) throws Exception {
        ByteBuf buffer = datagramPacket.content();
        //确保不会超出最大消息大小
        if(buffer.readableBytes() > MAX_MESSAGE_SIZE) {
            buffer.release();
            return;
        }
        UdpDatagram udpDatagram = parseUdpDatagram(buffer);
        UdpDatagram respUdpDatagram = dataCollector.processUdpDatagram(udpDatagram);
        if (null != respUdpDatagram) {
             handleReceivedData(ctx, respUdpDatagram, datagramPacket);
        }
    }
 
    /**
     * 处理接收到的数据
     * @param ctx
     * @param udpDatagram
     */
    public void handleReceivedData(ChannelHandlerContext ctx, UdpDatagram udpDatagram, DatagramPacket datagramPacket) throws ExecutionException, InterruptedException {
        Channel channel = ctx.channel();
        if (log.isInfoEnabled()) {
            log.info("received udp message: sessionId: {}, opCode: {}, short messageId: {}",
                    ctx.channel().id(), udpDatagram.getMessageTypeEnum(), udpDatagram.getShortMessageId());
        }
        byte[] payloadBytes = udpDatagram.getPayloadBytes();
        ByteBuf copiedBuffer = Unpooled.copiedBuffer(payloadBytes);
        ChannelFuture channelFuture = channel.writeAndFlush(new DatagramPacket(copiedBuffer.retain(), datagramPacket.sender()));
        channelFuture.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture channelFuture) throws Exception {
                if (channelFuture.isSuccess()) {
                    // 数据发送成功
                    log.info("数据发送成功:sender host: {}, sender port:{}, sender address:{}",datagramPacket.sender().getHostName(),datagramPacket.sender().getPort(), datagramPacket.sender().getAddress());
                } else {
                    // 数据发送失败
                    log.error("数据发送失败: {}",channelFuture.cause().getStackTrace());
                    channelFuture.cause().printStackTrace();
                }
            }
        });
    }
 
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        dataCollector.tcpConnect(ctx.channel());
    }
 
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        if (log.isWarnEnabled()) {
            log.warn("udp session throw an exception, sessionId:{} exception message: {}",
                    ctx.channel().id().asLongText(), cause.getMessage());
        }
    }
 
    //当客户端关闭链接时关闭通道
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        dataCollector.tcpChannelDisconnect(ctx.channel());
 
    }
}

处理类继承SimpleChannelInboundHandler类泛型类为DatagramPacket 

 writeAndFlush方法中发送的数据类型要是DatagramPacket

三、netty中的ChannelOption常用参数说明

1、ChannelOption.SO_BACKLOG

ChannelOption.SO_BACKLOG对应的是tcp/ip协议listen函数中的backlog参数。函数listen(int socketfd, int backlog)用来初始化服务端可连接队列。

服务端处理客户端连接请求是顺序处理的，所以同一时间只能处理一个客户端连接，多个客户端来的时候，服务端将不能处理的客户端连接请求放在队列中等待处理，backlog参数指定了队列的大小。

2、ChannelOption.SO_REUSEADDR

ChanneOption.SO_REUSEADDR对应于套接字选项中的SO_REUSEADDR，这个参数表示允许重复使用本地地址和端口。

比如，某个服务器进程占用了TCP的80端口进行监听，此时再次监听该端口就会返回错误，使用该参数就可以解决问题，该参数允许共用该端口，这个在服务器程序中比较常使用。

比如某个进程非正常退出，该程序占用的端口可能要被占用一段时间才能允许其他进程使用，而且程序死掉以后，内核一需要一定的时间才能够释放此端口，不设置SO_REUSEADDR就无法正常使用该端口。

3、ChannelOption.SO_KEEPALIVE

Channeloption.SO_KEEPALIVE参数对应于套接字选项中的SO_KEEPALIVE，该参数用于设置TCP连接，当设置该选项以后，连接会测试链接的状态，这个选项用于可能长时间没有数据交流的连接。

当设置该选项以后，如果在两小时内没有数据的通信时，TCP会自动发送一个活动探测数据报文。

4、ChannelOption.SO_SNDBUF和ChannelOption.SO_RCVBUF

ChannelOption.SO_SNDBUF参数对应于套接字选项中的SO_SNDBUF，ChannelOption.SO_RCVBUF参数对应于套接字选项中的SO_RCVBUF这两个参数用于操作发送缓冲区大小和接受缓冲区大小。

接收缓冲区用于保存网络协议站内收到的数据，直到应用程序读取成功，发送缓冲区用于保存发送数据，直到发送成功。

5、ChannelOption.SO_LINGER

ChannelOption.SO_LINGER参数对应于套接字选项中的SO_LINGER，Linux内核默认的处理方式是当用户调用close（）方法的时候，函数返回，在可能的情况下，尽量发送数据，不一定保证会发送剩余的数据，造成了数据的不确定性，使用SO_LINGER可以阻塞close()的调用时间，直到数据完全发送。

6、ChannelOption.TCP_NODELAY

ChannelOption.TCP_NODELAY参数对应于套接字选项中的TCP_NODELAY,该参数的使用与Nagle算法有关。

Nagle算法是将小的数据包组装为更大的帧然后进行发送，而不是输入一次发送一次，因此在数据包不足的时候会等待其他数据的到来，组装成大的数据包进行发送，虽然该算法有效提高了网络的有效负载，但是却造成了延时。

而该参数的作用就是禁止使用Nagle算法，使用于小数据即时传输。和TCP_NODELAY相对应的是TCP_CORK，该选项是需要等到发送的数据量最大的时候，一次性发送数据，适用于文件传输。

  

 ChannelOption通用配置