Netty 核心概念介绍和简单示例

迷南。 2023-06-15 13:24 64阅读 0赞

Netty是一个异步的，事件驱动的网络框架，我们可以使用它来开发高性能客户端和服务器。在本文中，我将解释一些基本的netty概念，以便当我们开始使用netty框架来构建我们自己的Http Web服务器和客户端时，很容易理解Netty术语。

## 核心概念 ##

### Channel ###

Channel是Java NIO的基础。它表示一个开放的连接，能够执行IO操作，例如读取和写入。

Channel向开发人员提供以下内容：

*  通道的当前状态（例如，是否打开？是否已连接？）
 *  通道的配置参数（例如，接收缓冲区大小）
 *  通道支持的I / O操作（例如，读取，写入，连接和绑定）
 *  在ChannelPipeline处理与通道相关联的所有I / O事件和请求。

### ChannelFuture ###

Netty的操作是异步的：即他们不保证完成就立即返回。此时最合乎逻辑的问题是：“ 如何确保在此任务完成之后而不是之前完成某些任务？为了做到这一点，Netty中的所有异步方法都返回一个ChannelFuture实例。此类具有在完成时通过管道传递其他任务的方法，以确保任务以非阻塞方式一个接一个地执行。

### ChannelFutureListener ###

在ChannelFuture接口中，其addListener()方法注册了一个ChannelFuturerListener，以便在某个操作完成时（无论是否成功）得到通知。

### ChannelPipeline ###

Netty使用管道来处理通过管道的数据。

Netty管道的工作方式都是相同的。管道中的每个类执行一个独立的任务。当请求/响应通过管道时，每个处理程序执行一个独立的任务，并将请求/响应传递到队列中的下一个处理程序。

### Netty Handlers ###

Netty中有两种主要类型的处理程序，分别是Channel Inbound Handlers和Channel Outbound Handlers。

入站处理程序用于处理入站流量，即从套接字通道到应用程序。同样，出站处理程序处理出站流量，即从应用程序到套接字通道。

从套接字读取数据是入站事件，将数据写入套接字是出站事件。因此，在netty中，处理程序会响应这些入站和出站事件而执行，我们可以在这些处理程序中定义数据处理逻辑。

### Encoders and Decoders ###

在使用网络协议时，我们需要执行数据序列化和反序列化。为此，Netty 为能够解码传入的数据引入了解码器。大多数解码器的基类是ByteToMessageDecoder。

为了对传出的数据进行编码，Netty引入了编码器。MessageToByteEncoder是大多数编码器实现的基础。

我们可以使用编码器和解码器将消息从字节序列转换为Java对象，反之亦然。

## 示例工程 ##

### 工程依赖 ###

首先，需要在pom.xml中提供Netty依赖项：

<dependency>
    	<groupId>io.netty</groupId>
    	<artifactId>netty-all</artifactId>
    	<version>4.1.30.Final</version>
    </dependency>

### 数据模型 ###

请求数据

public class RequestData {
    	private int intValue;
    	private String stringValue;
    
    	public int getIntValue() {
    		return intValue;
    	}
    
    	public void setIntValue(int intValue) {
    		this.intValue = intValue;
    	}
    
    	public String getStringValue() {
    		return stringValue;
    	}
    
    	public void setStringValue(String stringValue) {
    		this.stringValue = stringValue;
    	}
    
    	@Override
    	public String toString() {
    		return "RequestData{" + "intValue=" + intValue + ", stringValue='"
    				+ stringValue + '\'' + '}';
    	}
    }

响应数据

public class ResponseData {
    	private int intValue;
    
    	public int getIntValue() {
    		return intValue;
    	}
    
    	public void setIntValue(int intValue) {
    		this.intValue = intValue;
    	}
    
    	@Override
    	public String toString() {
    		return "ResponseData{" + "intValue=" + intValue + '}';
    	}
    }

### 客户端 ###

#### 请求数据编码 ####

现在，我们需要为我们的协议消息创建编码器和解码器。

应当注意，Netty与套接字接收缓冲区一起使用，该缓冲区不是以队列的形式而是以一堆字节的形式表示。这意味着当服务器未收到完整消息时，就可以调用我们的入站处理程序。

我们必须确保在处理之前已收到完整的消息，我们可以创建一个临时的ByteBuf并将所有入站字节附加到该字节上，同时将字符串长度先写入：

public class RequestDataEncoder extends MessageToByteEncoder<RequestData> {
    
    	private final Charset charset = Charset.forName("UTF-8");
    
    	@Override
    	protected void encode(ChannelHandlerContext ctx, RequestData msg,
    			ByteBuf out) throws Exception {
    		out.writeInt(msg.getIntValue());
    		// 记录写入游标
            int writerIndex = out.writerIndex();
            // 预写入一个假的length
            out.writeInt(0);
            // 写入UTF-8字符序列
            int length = ByteBufUtil.writeUtf8(out, msg.getStringValue());
            // 覆盖length
            out.setInt(writerIndex, length);
    	}
    }

在设计Netty的自定义协议的时候，一旦字符串类型的属性中出现中文就会出现解码异常的现象，这个异常并不一定出现了Exception，而是出现了解码之后字符截断出现了人类不可读的字符。

对于这个问题Netty已经提供了内置的工具类io.netty.buffer.ByteBufUtil。可以先记录一下writerIndex，先写一个假的值（例如0），再使用ByteBufUtil\#writeUtf8()写字符序列，然后根据返回的写入的字节长度，通过writerIndex覆盖之前写入的假值。

#### 处理器 ####

我们需要定义一个ClientHandler，它将发送请求并从服务器接收响应：

public class ClientHandler extends ChannelInboundHandlerAdapter {
    	@Override
    	public void channelActive(ChannelHandlerContext ctx) throws Exception {
    		RequestData msg = new RequestData();
    		msg.setIntValue(123);
    		msg.setStringValue("算算两倍是多少?");
    		ctx.writeAndFlush(msg);
    	}
    
    	@Override
    	public void channelRead(ChannelHandlerContext ctx, Object msg)
    			throws Exception {
    		System.out.println("计算结果："+msg);
    		ctx.close();
    	}
    }

#### 主程序 ####

现在写一个主程序启动我们的客户端：

public class NettyClient {
    	public static void main(String[] args) throws Exception {
    		String host = "localhost";
    		int port = 8080;
    		EventLoopGroup workerGroup = new NioEventLoopGroup();
    
    		try {
    			Bootstrap b = new Bootstrap();
    			b.group(workerGroup);
    			b.channel(NioSocketChannel.class);
    			b.option(ChannelOption.SO_KEEPALIVE, true);
    			b.handler(new ChannelInitializer<SocketChannel>() {
    				@Override
    				public void initChannel(SocketChannel ch) throws Exception {
    					ch.pipeline().addLast(new RequestDataEncoder(),
    							new ResponseDataDecoder(), new ClientHandler());
    				}
    			});
    
    			ChannelFuture f = b.connect(host, port).sync();
    
    			f.channel().closeFuture().sync();
    		} finally {
    			workerGroup.shutdownGracefully();
    		}
    	}
    }

### 服务端 ###

#### 请求数据解码 ####

在服务端应执行反向编码和解码。

public class RequestDataDecoder extends ReplayingDecoder<RequestData> {
    
    	private final Charset charset = Charset.forName("UTF-8");
    
    	@Override
    	protected void decode(ChannelHandlerContext ctx, ByteBuf in,
    			List<Object> out) throws Exception {
    		RequestData data = new RequestData();
    		data.setIntValue(in.readInt());
    		int strLen = in.readInt();
    		data.setStringValue(in.readCharSequence(strLen, charset).toString());
    		out.add(data);
    	}
    }

#### 处理器 ####

在服务端处理数据：

public class ProcessingHandler extends ChannelInboundHandlerAdapter {
    
    	@Override
    	public void channelRead(ChannelHandlerContext ctx, Object msg)
    			throws Exception {
    		RequestData requestData = (RequestData) msg;
    		ResponseData responseData = new ResponseData();
    		responseData.setIntValue(requestData.getIntValue() * 2);
    		ChannelFuture future = ctx.writeAndFlush(responseData);
    		future.addListener(ChannelFutureListener.CLOSE);
    		System.out.println(requestData.toString());
    	}
    }

#### 主程序 ####

服务端的启动程序：

public class NettyServer {
    
    	private int port;
    
    	private NettyServer(int port) {
    		this.port = port;
    	}
    
    	public static void main(String[] args) throws Exception {
    		int port;
    		if (args.length > 0) {
    			port = Integer.parseInt(args[0]);
    		} else {
    			port = 8080;
    		}
    		new NettyServer(port).run();
    	}
    
    	private void run() throws Exception {
    		EventLoopGroup bossGroup = new NioEventLoopGroup();
    		EventLoopGroup workerGroup = new NioEventLoopGroup();
    		try {
    			ServerBootstrap b = new ServerBootstrap();
    			b.group(bossGroup, workerGroup)
    					.channel(NioServerSocketChannel.class)
    					.childHandler(new ChannelInitializer<SocketChannel>() {
    						@Override
    						public void initChannel(SocketChannel ch)
    								throws Exception {
    							ch.pipeline().addLast(new RequestDataDecoder(),
    									new ResponseDataEncoder(),
    									new ProcessingHandler());
    						}
    					}).option(ChannelOption.SO_BACKLOG, 128)
    					.childOption(ChannelOption.SO_KEEPALIVE, true);
    
    			ChannelFuture f = b.bind(port).sync();
    			f.channel().closeFuture().sync();
    		} finally {
    			workerGroup.shutdownGracefully();
    			bossGroup.shutdownGracefully();
    		}
    	}
    }

### 运行结果 ###

客户端会得到预期的结果。

计算结果：ResponseData{intValue=246}

## 总结 ##

本文介绍了Netty中的一些基本概念，并通过一个简单的示例工程加以说明。完成工程代码可在[这里][Link 1]找到。

[Link 1]: https://github.com/peterwanghao/java-samples/tree/master/netty