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

一、基于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