Netty框架初探

最近因为功能设计到TCP传输，所以准备研究一下Netty框架的使用。

初次使用，先写一个最简单的Netty通讯，后面会对一些具体的细节进行说明。

声明几个变量

private static final int BIZGROUPSIZE = Runtime.getRuntime().availableProcessors()*2;
  private static final int BIZTHREADSIZE = 100;
  private static final EventLoopGroup bossGroup = new NioEventLoopGroup(BIZGROUPSIZE);
  private static final EventLoopGroup workerGroup = new NioEventLoopGroup(BIZTHREADSIZE);

  //服务端
  ServerBootstrap serverBootstrap;
  //客户端
  Bootstrap clientBootstrap;
  //客户端通道
  Channel clientChannel;
  //服务端通道
  Channel serverChannel ;

服务端代码

  public class NettyServer{
   public void init(){
    serverBootstrap = new ServerBootstrap();
    serverBootstrap.group(bossGroup, workerGroup);
    serverBootstrap.channel(NioServerSocketChannel.class);
    //添加handler监听服务端的IO动作
    serverBootstrap.handler(new OutHandler());
    serverBootstrap.childHandler(new ChannelInitializer<Channel>(){

     @Override
     protected void initChannel(Channel arg0)
       throws Exception {
      // TODO Auto-generated method stub
      ChannelPipeline pipeline = arg0.pipeline();
      pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
      pipeline.addLast(new LengthFieldPrepender(4));
      pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
      pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
      //添加handler监听客户端Channel的状态变化
      pipeline.addLast(new TcpServerHandler());
     }
     
    });
    try {
     //服务端启动
     ChannelFuture cf = serverBootstrap.bind(getLocalHostIp(),5656).sync();
     Toast.makeText(getActivity(), "TCP服务器已启动", Toast.LENGTH_SHORT).show();
    } catch (InterruptedException e) {
     // TODO Auto-generated catch block
     e.printStackTrace();
    }
   }
  }

客户端代码

  public class NettyClient{
   EventLoopGroup group = new NioEventLoopGroup();
   public void init(){
    Bootstrap b = new Bootstrap();
    b.group(group);
    b.channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY, true);
    b.handler(new ChannelInitializer<SocketChannel>() {
         @Override
         protected void initChannel(SocketChannel ch) throws Exception {
             ChannelPipeline pipeline = ch.pipeline();
              pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
       pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
       pipeline.addLast("decoder", new StringDecoder(CharsetUtil.UTF_8));
       pipeline.addLast("encoder", new StringEncoder(CharsetUtil.UTF_8));
        //添加一个Hanlder用来处理各种Channel状态
       pipeline.addLast("handlerIn", new ClientHandler());
        //添加一个Handler用来接收监听IO操作的
       pipeline.addLast("handlerOut", new OutHandler());
         }
     });
    ChannelFuture f;
    try {
     //连接服务端
    f = b.connect(getLocalHostIp(), 5656).sync();
    serverChannel = f.channel();
    serverChannel.writeAndFlush("<<<<<<<<<<<<<<<<客户端请求连接>>>>>>>>>>>>>>>>");
    } catch (InterruptedException e) {
     // TODO Auto-generated catch block
     e.printStackTrace();
    }
   }
  }

服务端绑定的Handler

  public class TcpServerHandler extends ChannelInboundHandlerAdapter {
       @Override
       public void channelRead(ChannelHandlerContext ctx, Object msg)
               throws Exception {
           // TODO Auto-generated method stub
           System.out.println("<<<<<<<<<<<<<<<<收到客户端消息 :"+ msg);
           ctx.channel().writeAndFlush("<<<<<<<<<<服务端已经接收：" + msg);
       }
       @Override
       public void channelActive(ChannelHandlerContext ctx) throws Exception {
           // TODO Auto-generated method stub
           System.out.println("通道已经启用>>>>>>>>");
           clientChannel = ctx.channel();
       }
         @Override
      public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
           System.out.println("exception is general");
       }
  }

客户端绑定的Handler

  public class ClientHandler extends ChannelInboundHandlerAdapter {
      @Override
      public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
          System.out.println("<<<<<<<<<客户端收到消息:" + msg);
      }

      @Override
      public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
          System.out.println("client exception is general");
      }
  }

两端共用的Handler

  public class OutHandler extends ChannelOutboundHandlerAdapter{
   @Override
   public void connect(ChannelHandlerContext ctx,
     SocketAddress remoteAddress, SocketAddress localAddress,
     ChannelPromise promise) throws Exception {
    // TODO Auto-generated method stub
    super.connect(ctx, remoteAddress, localAddress, promise);
    System.out.println("<<<<<<<<<<<<<<< connect server success >>>>>>>>>>>>>>>>");
   }

   @Override
   public void bind(ChannelHandlerContext ctx,
     SocketAddress localAddress, ChannelPromise promise)
     throws Exception {
    // TODO Auto-generated method stub
    super.bind(ctx, localAddress, promise);
    System.out.println("<<<<<<<<<<<<<<< server bind success >>>>>>>>>>>>>>>>");
   }
  }

启动

//启动服务端
new NettyServer().init();
//启动客户端
new NettyClient().init();

发送消息

//向服务端发消息
serverChannel.writeAndFlush("<<<<<<<<<<<<<我是客户端>>>>>>>>>>>>>");
//向客户端发消息
clientChannel.writeAndFlush("<<<<<<<<<<<<<我是服务端>>>>>>>>>>>>>");

至此一个收发通道就连通了。