1. Netty服务器以及客户端开发步骤

r囧r小猫 2021-09-27 19:02 252阅读 0赞

### Netty 服务器开发步骤： ###

1.  配置服务器端的线程组，新建两个服务器端的线程组 NioEventLoopGroup实例，它包含一组NIO线程，专门用于网络事件的处理，实际上他们就是Reactor线程组。创建两个原因是：一个用于服务端接受客户端的连接，另一是用于进行SocketChannel网络读写。

>     EventLoopGroup bossGroup = new NioEventLoopGroup();
>     EventLoopGroup workerGroup = new NioEventLoopGroup();

2.  创建ServerBootStrap对象，ServerBootStrap是Netty用于启动NIO服务端的辅助启动类，目的是降低服务端的开发复杂度

>     ServerBootstrap b = new ServerBootstrap();

> b.group(bossGroup,workerGroup)                        //将两个线程组传到group的方法中  
>              .channel(NioServerSocketChannel.class)           //  设置创建Channel  类似与JDK中的NIO  
>              .option(ChannelOption.SO\_BACKLOG, 1024)   //配置NioServerSocketChannel参数  
>              .childHandler(new ChildChannelHandler ());      //绑定IO处理事件

3.1  绑定IO处理事件  继承ChannelInitializer  重写initChannel方法

class ChildChannelHandler extends ChannelInitializer<SocketChannel>{
    
    	@Override
    	protected void initChannel(SocketChannel arg0) throws Exception {
    		// TODO Auto-generated method stub
    		arg0.pipeline().addLast(new TimeServerHandle());
    	}
    }

3.2 具体方法

public class TimeServerHandle extends ChannelHandlerAdapter {
    
    	@Override
    	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                        //做的类型转换，转换成ByteBuf对象，ByteBuf类似于ByteBuffer对象，只不过更强大一些，通过readableBytes()方法可以获得缓存区可读的字节数，根据字节数来创建byte数组，通过readBytes方法，可以将缓存区的字节数复制到新建的数组中
    		ByteBuf buf =  (ByteBuf)msg;  
    		byte[] req = new byte[buf.readableBytes()];
    		buf.readBytes(req);
    		            //转化成String
    		String body = new String(req,"UTF-8");
    		System.out.println("The time server receive order:"+body);
    
                        //对请求消息进行判断
    		String currentTime = "QUERY TIME ORDER".equals(body)? new java.util.Date(System.currentTimeMillis()).toString():"BAD ORDER";
    		ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
    
                    //通过ChannelHandlerContext的write方法来异步发送应答消息给客户端
    		ctx.write(resp);
    	}
    
    	@Override
    	public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
                    //为了不频繁的唤醒Selector进行消息发送，调用write方法只是将消息放到缓存数组中，在通过flush方法，将消息全部写入到SocketChannel中
    		ctx.flush();
    	}
    
    	@Override
    	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    		ctx.close();
    	}
    
    	
    }

4.   ServerBootstrap 实例 b 调用bind方法，调用sync（同步阻塞方法）等待绑定操作的完成 ,返回一个ChannelFuture。随后ChannelFuture实例调用channel().closeFuture().sync();等待服务器链路层关闭 才退出

ChannelFuture f;
    		try {
    			f = b.bind(port).sync();
    			//等待服务端监听窗口关闭
    			f.channel().closeFuture().sync();
    		} catch (InterruptedException e) {
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		}

5.  关闭NIO线程组相关资源

finally{
    			bossGroup.shutdownGracefully();
    			workerGroup.shutdownGracefully();
    		}

### Netty 客户端开发步骤： ###

1.  创建connect方法，参数是端口号和ip

2.  配置NIO线程组（用于读写）

3.  创建BootStrap实例，调用group方法，将线程组传入到方法中，配置Channel option 随后添加handler

4.  发起异步连接操作     ChannelFuture f = b.connect(host,port).sync();

5.  关闭

public class TimeClient {
    	
    	public void connect(int port,String host) throws Exception{
    		// 配置nio线程组
    		EventLoopGroup group = new NioEventLoopGroup();
    		
    		try {
    			Bootstrap b = new Bootstrap();
    			b.group(group).channel(NioSocketChannel.class)
    			.option(ChannelOption.TCP_NODELAY, true)
    			.handler(new ChannelInitializer<SocketChannel>() {
    				@Override
    				protected void initChannel(SocketChannel ch) throws Exception {
    					// TODO Auto-generated method stub
    					ch.pipeline().addLast(new TimeClientHandler());
    				}
    			});
    			// 发起异步连接操作
    			ChannelFuture f = b.connect(host,port).sync();
    			f.channel().closeFuture().sync();
    		} finally {
    			group.shutdownGracefully();
    		}
    	}
    	
    	public static void main(String[] args) {
    		int port = 8080;
    		if(args!=null && args.length>0){
    			port = Integer.valueOf(args[0]);
    		}
    		
    		try {
    			new TimeClient().connect(port, "localhost");
    		} catch (Exception e) {
    			// TODO Auto-generated catch block
    			e.printStackTrace();
    		}
    	}
    }

public class TimeClientHandler extends ChannelHandlerAdapter {
    	
    	private static final Logger logger = Logger.getLogger(TimeClientHandler.class.getName());
    	
    	private final ByteBuf firstMessage;
    	
    	public TimeClientHandler(){
    		byte[] req = "QUERY TIME ORDER".getBytes();
    		firstMessage = Unpooled.buffer(req.length);
    		firstMessage.writeBytes(req);
    	}
    
                    //客户端和服务器数据链路层连接成功调用此方法，发送查询事件指令给服务器，调用wirteAndFlush方法
    	@Override
    	public void channelActive(ChannelHandlerContext ctx) throws Exception {
    		ctx.writeAndFlush(firstMessage);
    	}
                    //当服务器返回应答消息时候，执行此方法，收到服务相应的信息，并且打印
    	@Override
    	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("Now is"+body);
    	}
    
                //发生异常时候，打印异常日志，释放客户端资源
    	@Override
    	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    		// 释放资源
    		logger.warning(""+cause.getMessage());
    		ctx.close();
    	}
    
    	
    	
    }