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相关内容，直到一个简陋的通讯服务器完成。

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持我们。