Netty实战之构建Echo服务器和客户端

Posted 2021-04-30 Coding夜未眠

本节，我们将实现一个简单的Echo服务器和客户端。

Echo客户端和服务器

连接到服务器的多个并发的客户端

图中显示了连接到服务器的多个并发的客户端。在理论上，客户端可以支持的连接数只受限于使用的 JDK 版本中的制约。

echo（回声）客户端和服务器之间的交互是很简单的：

先启动客户端，然后建立一个连接并发送一个或多个消息发送到服务器，其中每相呼应消息返回给客户端。显然，这个应用程序的用处没有很大。但进行这项工作就是为了能够更好的理解请求 - 响应交互本身，这是一个基本的模式的客户端/服务器系统。

编写Echo服务器

Netty 实现的 echo服务器都需要下面这些条件：

• 一个服务器 handler：这个组件实现了服务器对从客户端接收的数据的处理，即它的业务逻辑，决定了连接创建后和接收到信息后该如何处理；

• Bootstrapping：配置服务器的启动，它会将服务器绑定到它要监听连接请求的端口上，用来监听连接请求。

Echo Server 将会将接受到的数据的拷贝发送给客户端。因此，我们需要实现 ChannelInboundHandler 接口，用来定义处理入站事件的方法。由于我们的应用很简单，只需要继承 ChannelInboundHandlerAdapter 就行了。这个类提供了默认 ChannelInboundHandler的实现，所以只需要覆盖下面的方法：

• channelRead() - 每个信息入站都会调用

• channelReadComplete() - 通知处理器最后的 channelread() 是当前批处理中的最后一条消息时调用

• exceptionCaught()- 读操作时捕获到异常时调用

EchoServerHandler代码如下：

//标识这类的实例之间可以在 channel 里面共享

@Sharable

public class EchoServerHandler extends ChannelInboundHandlerAdapter{

@Override

public void channelRead(ChannelHandlerContext ctx, Object msg) {

ByteBuf in = (ByteBuf) msg;

//将消息输出到控制台

System.out.println("Server received: " + in.toString(CharsetUtil.UTF_8));

ctx.write(in);

}

@Override

public void channelReadComplete(ChannelHandlerContext ctx) {

//将接收到的消息写给发送者，而不冲刷出站消息

ctx.writeAndFlush(Unpooled.EMPTY_BUFFER)

//将未决消息冲刷到远程节点，并且关闭该Channel

.addListener(ChannelFutureListener.CLOSE);

}

@Override

public void exceptionCaught(ChannelHandlerContext ctx,Throwable cause) {

//打印异常栈跟踪

cause.printStackTrace();

//关闭该Channel

ctx.close();

}

}

这种使用 ChannelHandler 的方式体现了关注点分离的设计原则，并简化业务逻辑的迭代开发的要求。处理程序很简单;它的每一个方法可以覆盖到“hook（钩子）”在活动周期适当的点。很显然，我们覆盖channelRead因为我们需要处理所有接收到的数据。

覆盖 exceptionCaught 使我们能够应对任何Throwable的子类型。在这种情况下我们记录，并关闭所有可能处于未知状态的连接。它通常是难以从连接错误中恢复，所以干脆关闭远程连接。当然，也有可能的情况是可以从错误中恢复的，所以可以用一个更复杂的措施来尝试识别和处理这样的情况。

如果异常没有被捕获，会发生什么？

每个 Channel 都拥有一个与之相关联的 ChannelPipeline，其持有一个 ChannelHandler的实例链。在默认的情况下，ChannelHandler 会把对它的方法的调用转发给链中的下一个 ChannelHandler。因此，如果exceptionCaught()方法没有被该链中的某处实现，那么所接收的异常将会被传递到 ChannelPipeline的尾端并被记录。 为此，应用程序应该提供至少有一个实现了exceptionCaught()方法的 ChannelHandler。

• 针对不同类型的事件来调用 ChannelHandler；

• 应用程序通过实现或者扩展 ChannelHandler 来挂钩到事件的生命周期，并且提供自 定义的应用程序逻辑；

• 在架构上，ChannelHandler 有助于保持业务逻辑与网络处理代码的分离。这简化了开 发过程，因为代码必须不断地演化以响应不断变化的需求。

下面，我们来编写引导服务器。

public class EchoServer {

private final int port;

public EchoServer(int port) {

this.port = port;

}

public static void main(String[] args) throws Exception {

int port = 8080;

new EchoServer(port).start();

}

public void start() throws Exception {

final EchoServerHandler serverHandler = new EchoServerHandler();

//创建EventLoopGroup

EventLoopGroup group = new NioEventLoopGroup();

try {

//创建ServerBootstrap

ServerBootstrap b = new ServerBootstrap();

b.group(group)

//指定所使用的 NIO传输 Channel

.channel(NioserverSocketChannel.class)

.localAddress(new InetSocketAddress(port))

//添加一个EchoServerHandler到子Channel的ChannelPipeline

//当一个新的连接被接受时，一个新的子Channel将会被创建，而 ChannelInitializer将会把一个你的EchoServerHandler的实例添加到该 Channel的 ChannelPipeline中。

.childHandler(new ChannelInitializer<SocketChannel>(){

@Override

public void initChannel(SocketChannel ch)

throws Exception {

//EchoServerHandler 被标注为@Shareable，所以我们可以总是使用同样的实例

ch.pipeline().addLast(serverHandler);

}

});

//异步地绑定服务器；调用 sync()方法阻塞等待直到绑定完成

ChannelFuture f = b.bind().sync();

//获取 Channel 的CloseFuture，并且阻塞当前线程直到它完成

f.channel().closeFuture().sync();

}finally {

//关闭 EventLoopGroup，释放所有的资源

group.shutdownGracefully().sync();

}

}

}

服务器的主代码组件是

• EchoServerHandler 实现了的业务逻辑

• 在 main() 方法，引导了服务器

执行后者所需的步骤是：

• 创建 ServerBootstrap 实例来引导服务器并随后绑定

• 创建并分配一个 NioEventLoopGroup 实例来处理事件的处理，如接受新的连接和读/写数据。

• 指定本地 InetSocketAddress 给服务器绑定

• 通过 EchoServerHandler 实例给每一个新的 Channel 初始化

• 最后调用 ServerBootstrap.bind() 绑定服务器

这样服务器的初始化就完成了，并可以被使用。

编写Echo客户端

客户端的工作内容：

1. 连接到服务器；

2. 发送一个或者多个消息；

3. 对于每个消息，等待并接收从服务器发回的相同的消息；

4. 关闭连接。

首先，我们通过ChannelHandler实现客户端逻辑。

跟写服务器一样，我们用ChannelInboundHandler 来处理数据。下面例子，我们用 SimpleChannelInboundHandler 来处理所有的任务，需要覆盖三个方法：

• channelActive() - 服务器的连接被建立后调用

• channelRead0() - 数据后从服务器接收到调用

• exceptionCaught() - 捕获一个异常时调用

@Sharable

public class EchoClientHandler extends SimpleChannelInboundHandler<ByteBuf>{

@Override

public void channelActive(ChannelHandlerContext ctx) {

//当被通知 Channel是活跃的时候，发送一条消息

ctx.writeAndFlush(Unpooled.copiedBuffer("Netty HelloWorld!",CharsetUtil.UTF_8));

}

@Override

//记录已接收消息的转储

public void channelRead0(ChannelHandlerContext ctx, ByteBuf in) {

System.out.println("Client received: " + in.toString(CharsetUtil.UTF_8));

}

@Override

//在发生异常时，记录错误并关闭Channel

public void exceptionCaught(ChannelHandlerContext ctx,Throwable cause) {

cause.printStackTrace();

ctx.close();

}

}

SimpleChannelInboundHandler vs. ChannelInboundHandler

何时用这两个要看具体业务的需要。在客户端，当 channelRead0() 完成，我们已经拿到的入站的信息。当方法返回时，SimpleChannelInboundHandler 会小心的释放对 ByteBuf（保存信息） 的引用。而在 EchoServerHandler,我们需要将入站的信息返回给发送者，由于 write() 是异步的，在 channelRead() 返回时，可能还没有完成。所以，我们使用 ChannelInboundHandlerAdapter,无需释放信息。最后在 channelReadComplete() 我们调用 ctxWriteAndFlush() 来释放信息。

之后，我们编写EchoClient类引导客户端。

public class EchoClient {

private final String host;

private final int port;

public EchoClient(String host, int port) {

this.host = host;

this.port = port;

}

public void start() throws Exception {

EventLoopGroup group = new NioEventLoopGroup();

try {

//创建 Bootstrap

Bootstrap b = new Bootstrap();

b.group(group)

//指定 EventLoopGroup以处理客户端事件；需要适用于NIO的实现

.channel(NioSocketChannel.class)

//适用于 NIO传输的Channel类型

.remoteAddress(new InetSocketAddress(host, port))

//在创建Channel时，向 ChannelPipeline中添加一个 EchoClientHandler实例

.handler(new ChannelInitializer<SocketChannel>() {

@Override

public void initChannel(SocketChannel ch)throws Exception {

ch.pipeline().addLast(

new EchoClientHandler());

}

});

//连接到远程节点，阻塞等待直到连接完成

ChannelFuture f = b.connect().sync();

//阻塞，直到Channel关闭

f.channel().closeFuture().sync();

} finally {

//关闭线程池并且释放所有的资源

group.shutdownGracefully().sync();

}

}

public static void main(String[] args) throws Exception {

String host = "localhost";

int port = 8080;

new EchoClient(host, port).start();

}

}

回顾一下，我们的操作重点：

• 创建一个 Bootstrap 来初始化客户端

• 一个 NioEventLoopGroup 实例被分配给处理该事件的处理，这包括创建新的连接和处理入站和出站数据

• 创建一个 InetSocketAddress 以连接到服务器

• 连接好服务器之时，将安装一个 EchoClientHandler 在 pipeline

• 之后 Bootstrap.connect（）被调用连接到远程的 - 本例就是 echo(回声)服务器。