Java网络编程----Netty入门(NIO框架)

亦凉 2022-06-01 11:45 329阅读 0赞

**一.介绍**

Netty是基于NIO的框架,在介绍Netty之前,对比下这两种方式的通信步骤

**1.1 NIO通信步骤**

(1) 创建ServerSocketChannel,为它配置非阻塞模式

(2) 绑定监听,配置TCP参数,录入backlog大小等

(3) 创建一个独立的IO线程,用于轮询多路复用器Selector

(4) 创建Selector,将之前创建的ServerSocketChannel注册到Selector上,并设置监听标识位SelectionKey.OP\_ACCEPT

(5) 启动IO线程,在循环体中执行Selector.select()方法,轮询就绪的通道

(6) 当轮询到达处于就绪的通道时,需要进行判断操作位,如果是ACCEPT状态,说明是新的客户端接入,则调用accept方法接受新的客户端

(7) 设置新接入客户端的一些参数,如非阻塞,并将其通道继续注册到Selector之中,设置监听标识位等

(8) 如果轮询的通道操作位是READ,则进行读取,构造Buffer对象等

(9) 更细节的还有数据没发送完成继续发送的问题

.......一个简单的NIO服务端程序就是这样执行的

**1.2 Netty通信步骤**

(1) 创建两个NIO线程组,一个专门用于网络事件处理(接受客户端的连接),另一个则进行网络通信读写

(2) 创建一个ServerBootstrap对象,配置Netty的一系列参数,例如接收传出数据的缓存大小等

(3) 创建一个实际处理数据的类ChannelInitializer,进行初始化的准备工作,比如设置接收传出数据的字符集,格式,已经实际处理数据的接口

(4) 绑定端口,执行同步阻塞方法,等待服务器端启动即可

**对比以上两种通信的处理步骤,Netty只使用了四个步骤就实现了其服务端的编写,仅用几十行代码就可以把服务端程序完成得健壮而稳定,这对于单纯的NIO处理,几乎是不可能的,这是Netty框架的一大优势.**

**二.代码示例**

package com.east.netty;
    
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelOption;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    
    /**
     * 对于ChannelOption.SO_BACKLOG的解释
     * 服务器端TCP内核模块维护有两个队列,称之为A,B吧
     * 客户端向服务端Connect的时候,会发送带有SYN标志的包(第一次握手)
     * 服务端收到客户端发来的SYN时,向客户端发送SYN_ACK确认(第二次握手)
     * 此时TCP内核模块把客户端连接加入到A队列中,然后服务端收到客户端发来的ACK(第三次握手)
     * TCP内核模块把客户端连接从A队列移到B队列,连接成功,应用程序的accept会返回
     * 也就是说accept从B队列中取出完成三次握手的连接
     * A队列和B队列的长度之和是backlog.当A,B队列的长度之和大于backlog时,新连接将会被TCP内核拒绝
     * 所以,如果backlog过小,可能会出现accept速度跟不上,A,B队列满了,导致新的客户端无法连接
     * 要注意的是,backlog对程序支持的连接数并无影响,backlog影响的只是还没有被accept取出的连接
     * 
     * @author lhy
     * @time  2018.01.26
     *
     */
    public class Server {
    
    	public static void main(String[] args) throws Exception {
    		// TODO Auto-generated method stub
          EventLoopGroup pGroup = new NioEventLoopGroup();       //一个是用于处理服务器端接受客户端连接的
          EventLoopGroup rGroup = new NioEventLoopGroup();       //一个是进行网络通信的(网络读写的)
          ServerBootstrap b = new ServerBootstrap();             //2.创建辅助工具类,用于服务器通道的一系列配置
          b.group(pGroup,rGroup)                                //绑定两个线程组
          .channel(NioServerSocketChannel.class)                 //指定NIO的模式
          .option(ChannelOption.SO_BACKLOG, 1024)                //设置TCP缓冲区
          .option(ChannelOption.SO_SNDBUF, 32*1024)              //设置发送缓冲大小
          .option(ChannelOption.SO_RCVBUF, 32*1024)              //这是接受缓冲大小
          .option(ChannelOption.SO_KEEPALIVE, true)              //保持连接
          .childHandler(new ChannelInitializer<SocketChannel>() {
        	  protected void initChannel(SocketChannel sc) throws Exception{
    			sc.pipeline().addLast(new ServerHandler());       //3.在这里配置具体数据接受方法的处理(获得管道)
    			sc.pipeline().addLast(new ServerHandler());
    		}
    	});
          
          ChannelFuture cf1 = b.bind(8788).sync();                //4.进行绑定
          
          System.out.println("Server Start ...");
          // Thread.sleep(Integer.MAX_VALUE);
          cf1.channel().closeFuture().sync();                     //5.等待关闭
          
          pGroup.shutdownGracefully();
          rGroup.shutdownGracefully();
          
          //ChannelFuture cf2 = b.bind(8765).sync();
          //cf2.channel().closeFuture().sync();
    	}  
    
    }

package com.east.netty;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    
    public class ServerHandler extends ChannelHandlerAdapter {
    
    	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception{
    		ByteBuf buf = (ByteBuf) msg;
    		byte [] req = new byte[buf.readableBytes()];
    		buf.readBytes(req);
    		
    		String body = new String(req,"utf-8");
    		System.out.println("Server" + body);
    		
    		//服务端给客户端的响应
    		String response = "Hi Client!!";
    		ctx.writeAndFlush(Unpooled.copiedBuffer(response.getBytes()));
    //		.addListener(ChannelFutureListener.CLOSE);
    		
    	}
    	
    	public void exceptionCaught(ChannelHandlerContext ctx,Throwable t) throws Exception{
    		ctx.close();
    	}
    
    }

package com.east.netty;
    
    import io.netty.bootstrap.Bootstrap;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioSocketChannel;
    
    public class Client {
    
    	public static void main(String[] args) throws Exception {
    		// TODO Auto-generated method stub
    		EventLoopGroup group = new NioEventLoopGroup();     //生成一个线程组
    		Bootstrap b = new Bootstrap();
    		b.group(group).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {
    			@Override
    			protected void initChannel(SocketChannel sc) throws Exception {
    				// TODO Auto-generated method stub
    				sc.pipeline().addLast(new ClientHandler());        //可以绑定多个,添加一个个的关系器或方法
    			}
    		});
          
    		ChannelFuture cf1 = b.connect("127.0.0.1",8788).sync();
    		System.out.println("Client Connected ...");                //连接成功,同时向服务端发送数据
    		cf1.channel().writeAndFlush(Unpooled.copiedBuffer("hello haige!!".getBytes()));
    		Thread.sleep(1000);
    		cf1.channel().writeAndFlush(Unpooled.copiedBuffer("hello haige!!".getBytes()));
    		Thread.sleep(1000);
    		cf1.channel().writeAndFlush(Unpooled.copiedBuffer("hello haige!!".getBytes()));
    		Thread.sleep(1000);
    		
    		cf1.channel().closeFuture().sync();
    		group.shutdownGracefully();
    	}
    
    }

package com.east.netty;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.util.ReferenceCountUtil;
    
    public class ClientHandler extends ChannelHandlerAdapter {
    
    	public void channelRead(ChannelHandlerContext ctx,Object msg) throws Exception{
    		try {
    			ByteBuf buf = (ByteBuf) msg;
    			byte [] req = new byte [buf.readableBytes()];
    			buf.readBytes(req);
    			String body = new String(req,"utf-8");
    			System.out.println("Client :" + body);
    		} catch (Exception e) {
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		}finally {
    			ReferenceCountUtil.release(msg);
    		}
    	}
    	
    	public void channelActive(ChannelHandlerContext ctx) throws Exception{
    		System.out.println("通道激活...");
    	}
    	
    	public void channelReadComplete(ChannelHandlerContext ctx) throws Exception{
    		System.out.println("读取完毕...");
    	}
    
    }