跟大家聊聊 Netty 和 session

开发过 Web 应用的同学应该都会使用 session。由于 HTTP 协议本身是无状态的，所以一个客户端多次访问这个 web 应用的多个页面，服务器无法判断多次访问的客户端是否是同一个客户端。有了 session 就可以设置一些和客户端相关的属性，用于保持这种连接状态。例如用户登录系统后，设置 session 标记这个客户端已登录，那么访问别的页面时就不用再次登录了。

不过本文的内容不是 Web 应用的 session，而是 TCP 连接的 session，实际上二者还是有很大区别的。Web 应用的 session 实现方式并不是基于同一个 TCP 连接，而是通过 cookie 实现，这里不再详细展开。上面讲到 Web 应用的 session 只是让大家理解 session 的概念。

在同步阻塞的网络编程中，代码都是按照 TCP 操作顺序编写的，即创建连接、多次读写、关闭连接，这样很容易判断这一系列操作是否是同一个连接。而在事件驱动的异步网络编程框架中，IO 操作都会触发一个事件调用相应的事件函数，例如接收到客户端的新数据，会调用 channelRead，同一个 TCP 连接的多次请求和多个客户端请求都是一样的。

那么如何判断多次请求到底是不是同一个 TCP 连接，如何保存连接相关的信息？针对这个问题，Netty 提供了相应的解决方案。

下面用 Netty 实现一个请求次数计数器，用于记录同一个连接多次请求的请求次数。

Netty 中分为两种情况，一种是针对每个 TCP 连接创建一个新的 ChannelHandler 实例，另一种是所有 TCP 连接共用一个 ChannelHandler 实例。这两种方式的区别在于 ChannelPipeline 的 addLast 方法中添加的是否是新的 ChannelHandler 实例。

针对每个TCP连接创建一个新的ChannelHandler实例

针对每个 TCP 连接创建一个新的 ChannelHandler 实例是最常用的一种方式。这种情况非常简单，直接在 ChannelHandler 的实现类中加入一个成员变量即可保存连接相关的信息。

public class TcpServer {

 public static void main(String[] args) throws InterruptedException {
   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 {
             ChannelPipeline pipeline = ch.pipeline();
             pipeline.addLast(new LineBasedFrameDecoder(80));
             pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
             pipeline.addLast(new TcpServerHandler()); // 针对每个TCP连接创建一个新的ChannelHandler实例
           }
         });
     ChannelFuture f = b.bind(8080).sync();
     f.channel().closeFuture().sync();
   } finally {
     workerGroup.shutdownGracefully();
     bossGroup.shutdownGracefully();
   }
 }

}

class TcpServerHandler extends ChannelInboundHandlerAdapter {

 // 连接相关的信息直接保存在TcpServerHandler的成员变量中
 private int counter = 0;

 @Override
 public void channelRead(ChannelHandlerContext ctx, Object msg) {

   counter++;

   String line = (String) msg;
   System.out.println("第" + counter + "次请求:" + line);
 }

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

所有TCP连接共用一个ChannelHandler实例

在这种情况下，就不能把连接相关的信息放在 ChannelHandler 实现类的成员变量中了，否则这些信息会被其他连接共用。这里就要使用到 ChannelHandlerContext 的 Attribute 了。

public class TcpServer {

 public static void main(String[] args) throws InterruptedException {
   EventLoopGroup bossGroup = new NioEventLoopGroup();
   EventLoopGroup workerGroup = new NioEventLoopGroup();
   try {
     ServerBootstrap b = new ServerBootstrap();
     b.group(bossGroup, workerGroup)
         .channel(NioServerSocketChannel.class)
         .childHandler(new ChannelInitializer<SocketChannel>() {

           private TcpServerHandler tcpServerHandler = new TcpServerHandler();

           @Override
           public void initChannel(SocketChannel ch) throws Exception {
             ChannelPipeline pipeline = ch.pipeline();
             pipeline.addLast(new LineBasedFrameDecoder(80));
             pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
             pipeline.addLast(tcpServerHandler); // 多个连接使用同一个ChannelHandler实例
           }
         });
     ChannelFuture f = b.bind(8080).sync();
     f.channel().closeFuture().sync();
   } finally {
     workerGroup.shutdownGracefully();
     bossGroup.shutdownGracefully();
   }
 }

}

@Sharable // 多个连接使用同一个ChannelHandler，要加上@Sharable注解
class TcpServerHandler extends ChannelInboundHandlerAdapter {

 private AttributeKey<Integer> attributeKey = AttributeKey.valueOf("counter");

 @Override
 public void channelRead(ChannelHandlerContext ctx, Object msg) {

   Attribute<Integer> attribute = ctx.attr(attributeKey);

   int counter = 1;

   if(attribute.get() == null) {
     attribute.set(1);
   } else {
     counter = attribute.get();
     counter++;
     attribute.set(counter);
   }

   String line = (String) msg;
   System.out.println("第" + counter + "次请求:" + line);
 }

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

下面是一个 Java 实现的客户端，代码中发起了 3 次 TCP 连接，在每个连接中发送两次请求数据到服务器：

public class TcpClient {

 public static void main(String[] args) throws IOException, InterruptedException {

   // 3次TCP连接，每个连接发送2个请求数据
   for(int i = 0; i < 3; i++) {


     Socket socket = null;
     OutputStream out = null;

     try {

       socket = new Socket("localhost", 8080);
       out = socket.getOutputStream();

       // 第一次请求服务器
       String lines1 = "Hello\r\n";
       byte[] outputBytes1 = lines1.getBytes("UTF-8");
       out.write(outputBytes1);
       out.flush();

       // 第二次请求服务器
       String lines2 = "World\r\n";
       byte[] outputBytes2 = lines2.getBytes("UTF-8");
       out.write(outputBytes2);
       out.flush();

     } finally {
       // 关闭连接
       out.close();
       socket.close();
     }

     Thread.sleep(1000);
   }
 }
}

输出结果是：

第1次请求:Hello
第2次请求:World
第1次请求:Hello
第2次请求:World
第1次请求:Hello
第2次请求:World