Netty编写Echo服务器
使用Netty编写一个Echo服务器。Echo服务器用于将接收到的消息原样返回给客户端。
1.服务端
1.1代码流程
1.首先创建两个EventLoopGroup实例:bossGroup和workerGroup。bossGroup: 使用NioEventLoopGroup(1)构造函数创建,专门负责接受客户端连接。监听ServerSocketChannel的accept事件。
workerGroup:使用NioEventLoopGroup()默认构造函数创建,负责处理已建立连接的I/O操作。轮询处理多个SocketChannel的读写事件。
2.创建ServerBootstrap实例,这是一个服务器启动引导类。
通过连续调用其配置方法:首先调用group()方法设置两个事件循环组,然后调用channel()方法指定使用NioServerSocketChannel作为服务器通道类型,接着调用option()方法设置服务器选项,再调用childOption()方法设置子通道选项,最后调用childHandler()方法设置通道初始化器。
3.在ChannelInitializer的initChannel()方法中,构建处理管道Pipeline,依次添加StringDecoder解码器、StringEncoder编码器和EchoServerHandler业务处理器。 其中 EchoServerHandler入站事件是继承了ChannelInboundHandlerAdapter 类。重写channelRead方法用于接收数据。
4.配置完成后,调用ServerBootstrap的bind(port)方法绑定端口,然后调用sync()方法同步等待绑定完成。绑定成功后,调用Channel的closeFuture()方法获取关闭未来对象,再调用sync()方法阻塞等待服务器通道关闭。
1.2服务端执行流程
1.当客户端发起连接时,bossGroup的事件循环线程检测到连接事件,调用NioServerSocketChannel.accept()方法接受连接,创建新的NioSocketChannel实例,并将其注册到workerGroup的某个EventLoop中。
2.连接建立后,触发EchoServerHandler.channelActive()方法被调用,该方法记录客户端连接信息。
3.客户端发送消息时,workerGroup的EventLoop线程读取网络数据,触发处理管道Pipeline的处理流程:首先StringDecoder.channelRead()方法被调用,将字节数据解码为字符串;
4.然后EchoServerHandler.channelRead()方法被调用,处理业务逻辑,该方法内部调用ChannelHandlerContext.writeAndFlush()将接收到的消息原样写回客户端;在写回过程中,StringEncoder.encode()方法被调用,将字符串编码为字节数据发送到网络。
5.每次读取操作完成后,EchoServerHandler.channelReadComplete()方法被调用,确保所有缓冲数据都被刷新发送。
1.3服务端主程序代码实现
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.CharsetUtil;public class EchoServer {private final int port;public EchoServer(int port) {this.port = port;}public void run() throws Exception {// 创建事件循环组EventLoopGroup bossGroup = new NioEventLoopGroup(1); // 接收连接EventLoopGroup workerGroup = new NioEventLoopGroup(); // 处理连接try {ServerBootstrap b = new ServerBootstrap();b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class) // 使用NIO传输通道.childHandler(new ChannelInitializer<SocketChannel>() {@Overridepublic void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();// 添加字符串编解码器pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));// 添加自定义处理器pipeline.addLast(new EchoServerHandler());}}).option(ChannelOption.SO_BACKLOG, 128) // 连接队列大小.childOption(ChannelOption.SO_KEEPALIVE, true); // 保持连接// 绑定端口并启动服务器ChannelFuture f = b.bind(port).sync();System.out.println("Echo服务器启动,监听端口: " + port);// 等待服务器通道关闭f.channel().closeFuture().sync();} finally {// 优雅关闭workerGroup.shutdownGracefully();bossGroup.shutdownGracefully();}}public static void main(String[] args) throws Exception {int port = 8080;if (args.length > 0) {port = Integer.parseInt(args[0]);}new EchoServer(port).run();}
}
1.4EchoServerHandler代码实现
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;public class EchoServerHandler extends ChannelInboundHandlerAdapter {@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) {// 接收消息并原样返回String received = (String) msg;System.out.println("收到消息: " + received);// 将消息原样返回给客户端ctx.writeAndFlush(received);}@Overridepublic void channelReadComplete(ChannelHandlerContext ctx) {// 刷新缓冲区ctx.flush();}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {// 异常处理cause.printStackTrace();ctx.close();}@Overridepublic void channelActive(ChannelHandlerContext ctx) {System.out.println("客户端连接: " + ctx.channel().remoteAddress());}@Overridepublic void channelInactive(ChannelHandlerContext ctx) {System.out.println("客户端断开: " + ctx.channel().remoteAddress());}
}
2.客户端
2.1客户端代码流程
1.首先创建EventLoopGroup实例,然后创建Bootstrap客户端引导实例。通过连续调用其配置方法:调用group()方法设置事件循环组,调用channel()方法指定使用NioSocketChannel作为客户端通道类型,调用handler()方法设置通道初始化器。
2.在ChannelInitializer的initChannel()方法中,构建处理管道Pipeline,依次添加StringDecoder解码器、StringEncoder编码器和EchoClientHandler自定义业务处理器。EchoClientHandler继承了SimpleChannelInboundHandler类,重写channelRead0方法。
3.配置完成后,调用Bootstrap的connect(host, port)方法连接服务器,然后连续调用sync()方法同步等待连接完成。
2.2客户端执行流程
1.连接建立后,触发EchoClientHandler.channelActive()方法被调用,通知用户连接已建立。
2.客户端启动控制台输入循环,通过BufferedReader.readLine()方法读取用户输入。当用户输入消息后,调用Channel的writeAndFlush()方法发送消息到服务器。在发送过程中,StringEncoder.encode()方法被调用,将字符串编码为字节数据。
3.当服务器返回响应时,客户端EventLoop读取网络数据,触发处理管道Pipeline的处理流程:首先StringDecoder.channelRead()方法被调用,将字节数据解码为字符串;然后EchoClientHandler.channelRead0()方法被调用,显示服务器返回的消息。
2.3客户端主程序代码实现
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;import java.io.BufferedReader;
import java.io.InputStreamReader;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 b = new Bootstrap();b.group(group).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {@Overridepublic void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();pipeline.addLast(new StringDecoder());pipeline.addLast(new StringEncoder());pipeline.addLast(new EchoClientHandler());}});// 连接服务器Channel channel = b.connect(host, port).sync().channel();System.out.println("连接到Echo服务器,输入消息进行测试:");// 从控制台读取输入BufferedReader in = new BufferedReader(new InputStreamReader(System.in));while (true) {String input = in.readLine();if (input == null || "quit".equalsIgnoreCase(input)) {break;}channel.writeAndFlush(input);}} finally {group.shutdownGracefully();}}public static void main(String[] args) throws Exception {new EchoClient("localhost", 8080).start();}
}
2.4EchoClientHandler实现
class EchoClientHandler extends SimpleChannelInboundHandler<String> {@Overrideprotected void channelRead0(ChannelHandlerContext ctx, String msg) {System.out.println("服务器返回: " + msg);}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {cause.printStackTrace();ctx.close();}
}
参考资料
《Netty in Action》
