Netty 学习笔记一、Hello Netty

我不是女神ヾ 2023-10-17 13:35 107阅读 0赞

#### Netty客户端和服务器概述 ####

本节将引导你构建一个完整的Netty服务器和客户端。一般情况下，你可能只关心编写服务器，如一个http服务器的客户端是浏览器。然后在这个例子中，你若同时实现了服务器和客户端，你将会对他们的原理更加清晰。

一个Netty程序的工作图如下

![Center][]

1.  客户端连接到服务器
2.  建立连接后，发送或接收数据
3.  服务器处理所有的客户端连接

从上图中可以看出，服务器会写数据到客户端并且处理多个客户端的并发连接。从理论上来说，限制程序性能的因素只有系统资源和JVM。为了方便理解，这里举了个生活例子，在山谷或高山上大声喊，你会听见回声，回声是山返回的；在这个例子中，你是客户端，山是服务器。喊的行为就类似于一个Netty客户端将数据发送到服务器，听到回声就类似于服务器将相同的数据返回给你，你离开山谷就断开了连接，但是你可以返回进行重连服务器并且可以发送更多的数据。

虽然将相同的数据返回给客户端不是一个典型的例子，但是客户端和服务器之间数据的来来回回的传输和这个例子是一样的。本章的例子会证明这一点，它们会越来越复杂。  
 接下来的几节将带着你完成基于Netty的客户端和服务器的应答程序。

#### 编写应答服务器 ####

写一个Netty服务器主要由两部分组成：

*  配置服务器功能，如线程、端口
 *  实现服务器处理程序，它包含业务逻辑，决定当有一个请求连接或接收数据时该做什么

##### 服务器源代码： #####

package com.netty.hello;
    
    import com.netty.handler.TimeServerHandler;
    
    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;
    
    public class TimeServer {
    	public void bind(int port) throws Exception {
    		// 服务器线程组 用于网络事件的处理 一个用于服务器接收客户端的连接
    		// 另一个线程组用于处理SocketChannel的网络读写
    		EventLoopGroup bossGroup = new NioEventLoopGroup();
    		EventLoopGroup workerGroup = new NioEventLoopGroup();
    
    		try {
    			// NIO服务器端的辅助启动类 降低服务器开发难度
    			ServerBootstrap serverBootstrap = new ServerBootstrap();
    			serverBootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)// 类似NIO中serverSocketChannel
    					.option(ChannelOption.SO_BACKLOG, 1024)// 配置TCP参数
    					.childHandler(new ChildChannelHandler());// 最后绑定I/O事件的处理类
    
    			// 服务器启动后 绑定监听端口 同步等待成功 主要用于异步操作的通知回调 回调处理用的ChildChannelHandler
    			ChannelFuture f = serverBootstrap.bind(port).sync();
    			System.out.println("timeServer启动");
    			// 等待服务端监听端口关闭
    			f.channel().closeFuture().sync();
    		} finally {
    			// 优雅退出 释放线程池资源
    			bossGroup.shutdownGracefully();
    			workerGroup.shutdownGracefully();
    			System.out.println("服务器优雅的释放了线程资源...");
    		}
    	}
    
    	/**
    	 * 网络事件处理器(内部类)
    	 */
    	private class ChildChannelHandler extends ChannelInitializer<SocketChannel> {
    		@Override
    		protected void initChannel(SocketChannel ch) throws Exception {
    			ch.pipeline().addLast(new TimeServerHandler());
    		}
    
    	}
    
    	public static void main(String[] args) throws Exception {
    		// TODO Auto-generated method stub
    		int port = 8000;
    		new TimeServer().bind(port);
    	}
    
    }

从上面这个简单的服务器例子可以看出，启动服务器应先创建一个ServerBootstrap对象，因为使用NIO，所以指定NioEventLoopGroup来接受和处理新连接，指定通道类型为NioServerSocketChannel。  
 接下来，调用childHandler放来指定连接后调用的ChannelHandler，这个方法传ChannelInitializer类型的参数，ChannelInitializer是个抽象类，所以需要实现initChannel方法，这个方法就是用来设置ChannelHandler。  
 最后绑定服务器特定端口，调用sync()方法会阻塞直到服务器完成绑定，然后服务器等待通道关闭，因为使用sync()，所以关闭操作也会被阻塞。现在你可以关闭EventLoopGroup和释放所有资源，包括创建的线程。  
 这个例子中使用NIO，因为它是目前最常用的传输方式，你可能会使用NIO很长时间，但是你可以选择不同的传输实现。例如，这个例子使用OIO方式传输，你需要指定OioServerSocketChannel。Netty框架中实现了多重传输方式，将再后面讲述。  
 本小节重点内容：

*  创建ServerBootstrap实例来引导绑定和启动服务器
 *  创建NioEventLoopGroup对象来处理事件，如接受新连接、接收数据、写数据等等
 *  设置childHandler执行所有的连接请求
 *  都设置完毕了，最后调用ServerBootstrap.bind(port) 方法来绑定服务器

##### 服务器业务逻辑 #####

Netty使用futures和回调概念，它的设计允许你处理不同的事件类型，更详细的介绍将再后面章节讲述，但是我们可以接收数据。你的channel handler必须继承ChannelInboundHandlerAdapter并且重写channelRead方法，这个方法在任何时候都会被调用来接收数据，在这个例子中接收的是字节。  
 下面是handler的实现，其实现的功能是将客户端发给服务器的数据返回给客户端：

package com.netty.handler;
    
    import java.util.Date;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    
    /**
     * server端网络IO事件处理
     * 
     * @author zzg
     *
     */
    public class TimeServerHandler extends ChannelHandlerAdapter {
    
    	@Override
    	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    		// TODO Auto-generated method stub
    		ctx.close();
    		System.out.println("服务器异常退出" + cause.getMessage());
    	}
    
    	@Override
    	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    		// TODO Auto-generated method stub
    		System.out.println("服务器读取到客户端请求...");
    		ByteBuf buf = (ByteBuf) msg;
    		byte[] req = new byte[buf.readableBytes()];
    		buf.readBytes(req);
    		String body = new String(req, "UTF-8");
    		System.out.println("the time server receive order:" + body);
    		String curentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ? new Date(System.currentTimeMillis()).toString()
    				: "BAD ORDER";
    		ByteBuf resp = Unpooled.copiedBuffer(curentTime.getBytes());
    		ctx.write(resp);
    		System.out.println("服务器做出了响应");
    	}
    
    	@Override
    	public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
    		// TODO Auto-generated method stub
    		ctx.flush();
    		System.out.println("服务器readComplete 响应完成");
    	}
    
    }

Netty使用多个Channel Handler来达到对事件处理的分离，因为可以很容易的添加、更新、删除业务逻辑处理handler。Handler很简单，它的每个方法都可以被重写，它的所有的方法中只有channelRead方法是必须要重写的。

##### 捕获异常 #####

重写ChannelHandler的exceptionCaught方法可以捕获服务器的异常，比如客户端连接服务器后强制关闭，服务器会抛出"客户端主机强制关闭错误"，通过重写exceptionCaught方法就可以处理异常，比如发生异常后关闭ChannelHandlerContext。

#### 编写客户端源代码： ####

服务器写好了，现在来写一个客户端连接服务器。应答程序的客户端包括以下几步：

*  连接服务器
 *  写数据到服务器
 *  等待接受服务器返回相同的数据
 *  关闭连接

#### 客户端源代码： ####

package com.netty.hello;
    
    import com.netty.handler.TimeClientHandler;
    
    import io.netty.bootstrap.Bootstrap;
    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.NioSocketChannel;
    
    public class TimeClient {
    	/**
    	 * 连接服务器
    	 * 
    	 * @param port
    	 * @param host
    	 * @throws Exception
    	 */
    	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 {
    							ch.pipeline().addLast(new TimeClientHandler());
    						}
    					});
    			// 异步链接服务器 同步等待链接成功
    			ChannelFuture f = b.connect(host, port).sync();
    			// 等待链接关闭
    			f.channel().closeFuture().sync();
    
    		} finally {
    			group.shutdownGracefully();
    			System.out.println("客户端优雅的释放了线程资源...");
    		}
    
    	}
    
    	public static void main(String[] args) throws Exception {
    		// TODO Auto-generated method stub
    		new TimeClient().connect(8000, "127.0.0.1");
    	}
    
    }

创建启动一个客户端包含下面几步：

*  创建Bootstrap对象用来引导启动客户端
 *  创建EventLoopGroup对象并设置到Bootstrap中，EventLoopGroup可以理解为是一个线程池，这个线程池用来处理连接、接受数据、发送数据
 *  添加一个ChannelHandler，客户端成功连接服务器后就会被执行
 *  调用Bootstrap.connect(host,port)来连接服务器
 *  最后关闭EventLoopGroup来释放资源

##### 客户端的业务逻辑 #####

客户端的业务逻辑的实现依然很简单，更复杂的用法将在后面章节详细介绍。和编写服务器的ChannelHandler一样，在这里将自定义一个类TimeClientHandler继承自ChannelHandlerAdapter类来处理业务；通过重写父类的三个方法来处理感兴趣的事件：

*  channelActive()：客户端连接服务器后被调用  
    
 *  channelRead0()：从服务器接收到数据后调用  
    
 *  exceptionCaught()：发生异常时被调用

package com.netty.handler;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    
    import java.util.logging.Logger;
    /**
     * Client 网络IO事件处理
     * @author xwalker
     *
     */
    public class TimeClientHandler extends ChannelHandlerAdapter {
    	private static final Logger logger=Logger.getLogger(TimeClientHandler.class.getName());
    	private  ByteBuf firstMessage;
    	public TimeClientHandler(){
    		byte[] req ="QUERY TIME ORDER".getBytes();
    		firstMessage=Unpooled.buffer(req.length);
    		firstMessage.writeBytes(req);
    	}
    	@Override
    	public void channelActive(ChannelHandlerContext ctx) throws Exception {
    		ctx.writeAndFlush(firstMessage);
    		System.out.println("客户端active");
    	}
    	
    	@Override
    	public void channelRead(ChannelHandlerContext ctx, Object msg)
    			throws Exception {
    		System.out.println("客户端收到服务器响应数据");
    		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 channelReadComplete(ChannelHandlerContext ctx) throws Exception {
    		ctx.flush();
    		System.out.println("客户端收到服务器响应数据处理完成");
    	}
    	
    	@Override
    	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
    			throws Exception {
    		logger.warning("Unexpected exception from downstream:"+cause.getMessage());
    		ctx.close();
    		System.out.println("客户端异常退出");
    	}
    }

#### 总结 ####

本章介绍了如何编写一个简单的基于Netty的服务器和客户端并进行通信发送数据。介绍了如何创建服务器和客户端以及Netty的异常处理机制。

[Center]: https://img-blog.csdn.net/20170604222512474?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvemhvdXpoaXdlbmdhbmc=/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/Center