Netty5用户手册之四：使用netty实现Timer客户端和服务端程序

谭仰岳

2023-12-01

服务端

这个协议使用来实现time协议。它与前一个例子不同的是它会发送一个包含32字节大小的int消息，不接受任何请求并且一旦发送消息马上关闭连接。这个例子中，将会学会怎么去构建并发送一个消息，并且完成发送后关闭连接。

由于我们将会忽略接收到的数据而是去发送一个消息当连接创建的时候，我们不能用channelRead方法了。替代它的我们需要覆盖channelActive方法，下面Wie实现代码：

package com.zzj.nio.netty;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

/**
 * @author zzj
 * @date Oct 19, 2016 5:00:50 PM
 */
public class TimeServerHandler extends ChannelInboundHandlerAdapter {

    @Override
	public void channelActive(final ChannelHandlerContext ctx) throws Exception {
    	final ByteBuf time = ctx.alloc().buffer(4);
    	time.writeInt((int)(System.currentTimeMillis()/1000l+2208988800l));
    	
    	final ChannelFuture future = ctx.writeAndFlush(time);
    	
    	//监听写入连接事件
    	future.addListener(new ChannelFutureListener() {
			
			@Override
			public void operationComplete(ChannelFuture f) throws Exception {
				  assert future == f;
	              ctx.close();
			}
		});
    	
	}

	@Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}

1.channelActive方法会被调用当产生了一个连接并且准备进行传输时。我们将会写一个32字节的int类型信息作为现在时间在这个方法中。

2.为了发送一个新消息，我们需要分配一个新buffer缓冲区来存放消息。这里通过ChannelHandlerContext.alloc()得到一个当前的ByteBufAllocator，然后分配一个新的缓冲。

3.像往常一样，我们需要构建消息。

3.1 在nio中我们通常调用java.nio.ByteBuffer对象的flip方法在发送消息之前。但是这里ByteBuf没有这个方法，这是因为它有两个点：一个是用于读操作，另一个是用于写操作。writer index会增加当写数据到bytebuf时，但是reader index索引却不会变化。reader index索引和writerindex索引代表了消息的开始和结束.

3.2 相反的，nio缓冲对象并没有提供一种简介的方法来计算出消息的开始和结束，除非你调用flip方法。但是，当你忘记调用flip方法而引起没有数据或者错误数据被发送时，你将很麻烦。这样的错误不应该发生在netty中，因为我们对于不同的操作类型有不同的指针。你会发现使用这种方式会很简单，因为你已经习惯一种没有使用flip的方式。

3.3 另外一个需要指出的是，ChannelHandlerContext.write（或writeAndFlush）方法执行后会返回一个ChannelFuture。一个ChannelFuture代表了一个还未发生的一个io操作。意思是，任何请求操作可能还没有被执行，因为所有的操作在netty中都是一步的。例如，下面的代码可能在发送消息之间就已经关闭：

Channel ch = ...;
ch.writeAndFlush(message);
ch.close();

因此，我们需要在ChannelFuture完成之后调用close方法，ChannelFuture是调用ChannelHandlerContext.write方法返回的，并且它会通知他的监听器写操作已经完成了。需要注意的是，close方法可能不会马上关闭connction，并且他也会返回一个channelFuture对象。

4.当一个写请求已经完成是如何通知到我们？这个只需要简单地在返回的ChannelFuture上增加一个ChannelFutureListener。这里我们构建了一个匿名的ChannelFutureListener类用来在操作完成时关闭Channel。或者，你可以使用简单的预定义监听器代码。

下面为启动main方法的类，类似Echo服务端，下面再粘贴出来：

package com.zzj.nio.netty;

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;

/**
 * 抛弃消息服务端
 * @author zzj
 * @date Oct 19, 2016 1:24:05 PM
 */
public class TimeServer {

	/**端口号**/
	private int port;
	
	/**
	 * 构造函数
	 * @param port
	 */
	public TimeServer(int port){
		this.port = port;
	}
	
	public static void main(String[] args) {
		int port = 9999;
		try {
			//开启服务端接受消息
			new TimeServer(port).run();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
	
	/**
	 * 开启服务器
	 * @author zzj
	 * @throws InterruptedException 
	 */
	public void run() throws InterruptedException{
		EventLoopGroup bossGroup = new NioEventLoopGroup();//(1)
		EventLoopGroup workerGroup = new NioEventLoopGroup();
		try {
			ServerBootstrap bootstrap = new ServerBootstrap();//(2)
			bootstrap.group(bossGroup, workerGroup)
			.channel(NioServerSocketChannel.class)//(3)
			.childHandler(new ChannelInitializer<SocketChannel>() {//(4)

				@Override
				protected void initChannel(SocketChannel channel) throws Exception {
					channel.pipeline().addLast(new TimeServerHandler());
				}
			});
			bootstrap.option(ChannelOption.SO_BACKLOG, 128)//(5)
			.childOption(ChannelOption.SO_KEEPALIVE, true);//(6)
			
			//紧接着，绑定并且开始接受请求的连接
			ChannelFuture future = bootstrap.bind(port).sync();
			
			//等待直到服务器socket被关闭,在这里例子中，这个不会发生，但是你可以这样写
			future.channel().closeFuture().sync();
		}finally{
			
			//优雅的关闭
			bossGroup.shutdownGracefully();
			workerGroup.shutdownGracefully();
		}
	}
}

Time客户端

与discard和echo服务端不同，time程序中我们需要一个客户端协议，因为一个人不能手动的传输一个32字节的二进制日期数据。在这个部分，我们会讨论如何确保服务端正常的工作，并且怎么用netty写一个客户端。

服务端和客户端最大的不同是，Bootstrap和Channel实现的方式，下面为代码：

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;

/**
 * Time的客户端
 * @author zzj
 * @date Oct 19, 2016 5:45:12 PM
 */
public class TimeClient {
	
	public static void main(String[] args) throws InterruptedException {
		int port = 9999;
		
		EventLoopGroup workerGroup = new NioEventLoopGroup();
		try {
			
			Bootstrap bootstrap = new Bootstrap();
			bootstrap.group(workerGroup);
			bootstrap.channel(NioSocketChannel.class);
			bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
			bootstrap.handler(new ChannelInitializer<SocketChannel>() {
				@Override
				protected void initChannel(SocketChannel channel) throws Exception {
					
					//添加业务处理器
					channel.pipeline().addLast(new TimeClientHandler());
				}
			});
			
			//运行客户端
	    	ChannelFuture future = bootstrap.connect("localhost", port);
	    	
	    	//等待知道连接被关闭
	    	future.channel().closeFuture().sync();
		}finally{
			workerGroup.shutdownGracefully();
		}
	}
}

1.Bootstrap类似于ServerBootstrap，区别是它是一个非服务端channels而言，比如一个客户端或者非连接channel。

2.如果只声明了一个EventLoopGroup，那么这个对象将即作为boss group又作为worker group。但是在客户端boss group中没有用到。

3.替代NioServerSocketChannel，NioSocketChannel被用来创建一个客户端程序。

4.我们需要知道，我们不用childOption方法，不像我们处理ServerBooststrap一样。因为客户端SocketChannel没有一个父类。

5.我们需要调用connect方法而不是调用bind方法来运行客户端。
正如你看到的，他和服务端的代码是不一样的。ChannelHandler是如何实现的?他应该从服务端接受一个32位的整数消息，把他翻译成人们能读懂的格式，并打印翻译好的时间，最后关闭连接:

package com.zzj.nio.netty;

import java.util.Date;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

/**
 * Time客户端处理器类
 * @author zzj
 * @date Oct 19, 2016 5:48:56 PM
 */
public class TimeClientHandler extends ChannelInboundHandlerAdapter {

	 @Override
	    public void channelRead(ChannelHandlerContext ctx, Object msg) {
	        ByteBuf m = (ByteBuf) msg; // (1)
	        try {
	            long currentTimeMillis = (m.readUnsignedInt() - 2208988800L) * 1000L;
	            System.out.println(new Date(currentTimeMillis));
	            ctx.close();
	        } finally {
	            m.release();
	        }
	    }

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
	   cause.printStackTrace();
       ctx.close();
	}
}

在TCP/IP中，NETTY会把读到的数据放到ByteBuf的数据结构中。

这样看起来非常简单，并且和服务端的那个例子的代码也相差不多。然而，处理器有时候会因为抛出IndexOutOfBoundsException而拒绝工作。在下个部分我们会讨论为什么会发生这种情况。。。

Netty5用户手册之四：使用netty实现Timer客户端和服务端程序

服务端

Time客户端

相关阅读

相关文章

相关问答

相关文档