Java NIO框架Netty简单使用的示例

之前写了一篇文章：Java 网络IO编程总结（BIO、NIO、AIO均含完整实例代码），介绍了如何使用Java原生IO支持进行网络编程，本文介绍一种更为简单的方式，即Java NIO框架。

Netty是业界最流行的NIO框架之一，具有良好的健壮性、功能、性能、可定制性和可扩展性。同时，它提供的十分简单的API，大大简化了我们的网络编程。

同Java IO介绍的文章一样，本文所展示的例子，实现了一个相同的功能。

1、服务端

Server：

package com.anxpp.io.calculator.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;
public class Server {
 private int port;
 public Server(int port) {
   this.port = port;
 }
 public void run() throws Exception {
   EventLoopGroup bossGroup = new NioEventLoopGroup();
   EventLoopGroup workerGroup = new NioEventLoopGroup();
   try {
     ServerBootstrap b = new ServerBootstrap();
     b.group(bossGroup, workerGroup)
      .channel(NioServerSocketChannel.class)
      .option(ChannelOption.SO_BACKLOG, 1024)
      .childOption(ChannelOption.SO_KEEPALIVE, true)
      .childHandler(new ChannelInitializer<SocketChannel>() {
        @Override
        public void initChannel(SocketChannel ch) throws Exception {
          ch.pipeline().addLast(new ServerHandler());
        }
      });
     ChannelFuture f = b.bind(port).sync();
     System.out.println("服务器开启："+port);
     f.channel().closeFuture().sync();
   } finally {
     workerGroup.shutdownGracefully();
     bossGroup.shutdownGracefully();
   }
 }
 public static void main(String[] args) throws Exception {
   int port;
   if (args.length > 0) {
     port = Integer.parseInt(args[0]);
   } else {
     port = 9090;
   }
   new Server(port).run();
 }
}

ServerHandler：

package com.anxpp.io.calculator.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import java.io.UnsupportedEncodingException;
import com.anxpp.io.utils.Calculator;
public class ServerHandler extends ChannelInboundHandlerAdapter {
 @Override
 public void channelRead(ChannelHandlerContext ctx, Object msg) throws UnsupportedEncodingException {
   ByteBuf in = (ByteBuf) msg;
   byte[] req = new byte[in.readableBytes()];
   in.readBytes(req);
   String body = new String(req,"utf-8");
   System.out.println("收到客户端消息:"+body);
   String calrResult = null;
   try{
     calrResult = Calculator.Instance.cal(body).toString();
   }catch(Exception e){
     calrResult = "错误的表达式：" + e.getMessage();
   }
   ctx.write(Unpooled.copiedBuffer(calrResult.getBytes()));
 }
 @Override
 public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
   ctx.flush();
 }
 /**
  * 异常处理
  */
 @Override
 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
   cause.printStackTrace();
   ctx.close();
 }
}
package com.anxpp.io.calculator.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import java.io.UnsupportedEncodingException;
import com.anxpp.io.utils.Calculator;
public class ServerHandler extends ChannelInboundHandlerAdapter {
 @Override
 public void channelRead(ChannelHandlerContext ctx, Object msg) throws UnsupportedEncodingException {
   ByteBuf in = (ByteBuf) msg;
   byte[] req = new byte[in.readableBytes()];
   in.readBytes(req);
   String body = new String(req,"utf-8");
   System.out.println("收到客户端消息:"+body);
   String calrResult = null;
   try{
     calrResult = Calculator.Instance.cal(body).toString();
   }catch(Exception e){
     calrResult = "错误的表达式：" + e.getMessage();
   }
   ctx.write(Unpooled.copiedBuffer(calrResult.getBytes()));
 }
 @Override
 public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
   ctx.flush();
 }
 /**
  * 异常处理
  */
 @Override
 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
   cause.printStackTrace();
   ctx.close();
 }
}

2、客户端

Client：

package com.anxpp.io.calculator.netty;
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;
import java.util.Scanner;
public class Client implements Runnable{
 static ClientHandler client = new ClientHandler();
 public static void main(String[] args) throws Exception {
   new Thread(new Client()).start();
   @SuppressWarnings("resource")
   Scanner scanner = new Scanner(System.in);
   while(client.sendMsg(scanner.nextLine()));
 }
 @Override
 public void run() {
   String host = "127.0.0.1";
   int port = 9090;
   EventLoopGroup workerGroup = new NioEventLoopGroup();
   try {
     Bootstrap b = new Bootstrap();
     b.group(workerGroup);
     b.channel(NioSocketChannel.class);
     b.option(ChannelOption.SO_KEEPALIVE, true);
     b.handler(new ChannelInitializer<SocketChannel>() {
       @Override
       public void initChannel(SocketChannel ch) throws Exception {
         ch.pipeline().addLast(client);
       }
     });
     ChannelFuture f = b.connect(host, port).sync();
     f.channel().closeFuture().sync();
   } catch (InterruptedException e) {
     e.printStackTrace();
   } finally {
     workerGroup.shutdownGracefully();
   }
 }
}

ClientHandler：

package com.anxpp.io.calculator.netty;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import java.io.UnsupportedEncodingException;
public class ClientHandler extends ChannelInboundHandlerAdapter {
 ChannelHandlerContext ctx;
 /**
  * tcp链路简历成功后调用
  */
 @Override
 public void channelActive(ChannelHandlerContext ctx) throws Exception {
   this.ctx = ctx;
 }
 public boolean sendMsg(String msg){
   System.out.println("客户端发送消息："+msg);
   byte[] req = msg.getBytes();
   ByteBuf m = Unpooled.buffer(req.length);
   m.writeBytes(req);
   ctx.writeAndFlush(m);
   return msg.equals("q")?false:true;
 }
 /**
  * 收到服务器消息后调用
  * @throws UnsupportedEncodingException
  */
 @Override
 public void channelRead(ChannelHandlerContext ctx, Object msg) throws UnsupportedEncodingException {
   ByteBuf buf = (ByteBuf) msg;
   byte[] req = new byte[buf.readableBytes()];
   buf.readBytes(req);
   String body = new String(req,"utf-8");
   System.out.println("服务器消息："+body);
 }
 /**
  * 发生异常时调用
  */
 @Override
 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
   cause.printStackTrace();
   ctx.close();
 }
}

3、用于计算的工具类

package com.anxpp.io.utils;
import javax.script.ScriptEngine;
import javax.script.ScriptEngineManager;
import javax.script.ScriptException;
public enum Calculator {
 Instance;
 private final static ScriptEngine jse = new ScriptEngineManager().getEngineByName("JavaScript");
 public Object cal(String expression) throws ScriptException{
   return jse.eval(expression);
 }
}

4、测试

分别启动服务端和客户端，然后再客户端控制台输入表达式：

1+5+5+5+5+5
客户端发送消息：1+5+5+5+5+5
服务器消息：26
156158*458918+125615
客户端发送消息：156158*458918+125615
服务器消息：7.1663842659E10
1895612+555+5+5+5+5+5+5+5-5*4/4
客户端发送消息：1895612+555+5+5+5+5+5+5+5-5*4/4
服务器消息：1896197

可以看到服务端返回的结果。

查看服务端控制台：

服务器开启：9090
收到客户端消息:1+5+5+5+5+5
收到客户端消息:156158*458918+125615
收到客户端消息:1895612+555+5+5+5+5+5+5+5-5*4/4

5、更多

相关文章：

 Java 网络IO编程总结（BIO、NIO、AIO均含完整实例代码）

本文例子以及Java BIO NIO AIO例子的源码Git地址：https://github.com/anxpp/Java-IO.git

后续会继续更新Netty相关内容，直到一个简陋的通讯服务器完成。

来源：http://blog.csdn.net/anxpp/article/details/52108238