Echo客户端将会:
(1)连接到服务器;
(2)发送一个或者多个消息;
(3)对于每个消息,等待并接收从服务器发回的相同的消息;
(4)关闭连接。
1、通过ChannelHandler实现客户端逻辑
如同服务器,客户端将拥有一个用来处理数据的ChannelInboundHandler。在这个场景下,你将扩展SimpleChannelInboundHandler类以处理所有必须的任务,这就要求重写以下方法:
channelActive()——在到服务器的连接已经建立后将被调用;
channelRead()——当从服务器接收到一条消息时被调用;
exceptionCaught()——在处理过程中引发异常时被调用。
//标记该类的实例可以被多个Channel共享 @ChannelHandler.Sharable public class EchoClientHandler extends SimpleChannelInboundHandler<ByteBuf> { @Override public void channelActive(ChannelHandlerContext ctx){ //当被通知Channel是活跃的时候,发送一条消息 ctx.writeAndFlush(Unpooled.copiedBuffer("Netty rocks!", CharsetUtil.UTF_8)); } @Override protected void channelRead0(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf) throws Exception { //记录已接收消息的存储 System.out.println("Client received:" + byteBuf.toString(CharsetUtil.UTF_8)); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause){ //在发生异常时,记录错误并关闭Channel cause.printStackTrace(); ctx.close(); } }
(1)重写了channelActive()方法,其将在一个建立连接时被调用。这确保了数据将会被尽可能快地写入服务器,其在这个场景下是一个编码了字符串“Netty rocks!”的字节缓冲区。
(2)重写了channelRead()方法。每当接收数据时,都会调用这个方法。需要注意的是,由服务器接收的消息可能会被分块接收。也就是说,如果服务器发送了5字节,那么不能保证这5字节会被一次性接收。即使是对于这么少量的数据,channelRead()方法也可能会被调用两次,第一次使用一个持有3字节的ByteBuf(Netty的字节容器),第二次使用一个持有2字节的ByteBuf。作为一个面向流的协议,TCP保证了字节数组将会按照服务器发送他们的顺序被接收。
(3)重写了exceptionCaught()。关闭Channel,终止到服务器的连接,记录Throwable。
SimpleChannelInboundHandler与ChannelInboundHandler
为什么在客户端使用的是SimpleChannelInboundHandler,而在EchoServerHandler中所使用ChannelInboundHandlerAdapter?这和两个因素的相互作用有关:业务逻辑如何处理消息以及Netty如何管理资源。
在客户端,当channelRead()方法完成时,你已经有了消息,并且已经处理完它了。当该方法返回时,SimpleChannelInboundHandler负责释放指向保存该消息的ByteBuf的内存引用。
在EchoServerHandler中,你仍然需要将传入消息回送给发送者,而write()操作时异步地,直到channelRead()方法返回后可能仍然没有完成。为此,EchoServerHandler扩展了ChannelInboundHandlerAdapter,其在这个时间点上不会释放消息,
消息在EchoServerHandler的channelReadComplete()方法中,当writeAndFlush()方法被调用时被释放。
SimpleChannelInboundHandler与ChannelInboundHandler 为什么在客户端使用的是SimpleChannelInboundHandler,而在EchoServerHandler中所使用ChannelInboundHandlerAdapter?这和两个因素的相互作用有关:业务逻辑如何处理消息以及Netty如何管理资源。 在客户端,当channelRead()方法完成时,你已经有了消息,并且已经处理完它了。当该方法返回时,SimpleChannelInboundHandler负责释放指向保存该消息的ByteBuf的内存引用。 在EchoServerHandler中,你仍然需要将传入消息回送给发送者,而write()操作时异步地,直到channelRead()方法返回后可能仍然没有完成。为此,EchoServerHandler扩展了ChannelInboundHandlerAdapter,其在这个时间点上不会释放消息, 消息在EchoServerHandler的channelReadComplete()方法中,当writeAndFlush()方法被调用时被释放。
2.引导客户端
引导客户端类似于引导服务器,不同的是,客户端是使用主机和端口参数来连接远程地址,也就是这里的Echo服务器地址,而不是绑定要一个一直被监听的端口。
客户端的主类:
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(); //指定EventLoopGroup以处理客户端事件;需要使用于NIO的实现 b.group(group) //适用于NIO传输的Channel类型 .channel(NioSocketChannel.class) //设置服务器的InetSocketAddress .remoteAddress(new InetSocketAddress(host, port)) //在创建ChannelPipeline中添加一个EchoClientHandler实例 .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel channel) throws Exception { channel.pipeline().addLast(new EchoClientHandler()); } }); //连接到远程节点,阻塞等待直到连接完成 ChannelFuture future = b.connect().sync(); //阻塞,直到Channel关闭 future.channel().closeFuture().sync(); } finally { //关闭线程池并且释放所有的资源 group.shutdownGracefully().sync(); } } public static void main(String[] args) throws Exception { if(args.length != 2){ System.out.println("Usage:" + EchoClient.class.getSimpleName() + "<host> <port>"); return; } String host = args[0]; int port = Integer.parseInt(args[1]); new EchoClient(host, port).start(); } }
为初始化客户端,创建了一个Bootstrap实例;
为进行事件处理分配了一个NioEventLoopGroup实例,其中事件处理包括创建新的连接以及处理入站和出站数据;
为服务器连接创建了一个InetSocketAddress实例;
当连接被建立时,一个EchoClientHandler实例会被安装到(该Channel的)ChannelPipeline中;
在一切都设置完成后,调用BootStrap.connect()方法连接到远程节点。