Netty编程

一、概念

• 一个非阻塞的客户端—服务端网络通讯框架
• 基于异步事件驱动模型
• 简化Java的TCP和UDP编程
• 支持HTTP/2,SSL等多种协议
• 支持多种数据格式，如JSON等
• 关键技术
  • 通道Channel(数据交换)
    • ServerSocketChannel/NioServerSocketChannel/….
    • SocketChannel/NioSocketChannel
  • 事件EventLoop
    • 为每个通道定义一个EventLoop,处理所有I/O事件
    • EventLoop注册事件
    • EventLoop将事件派发给ChannelHandler
    • EventLoop安排进一步操作
  • 事件
    • 事件按照数据流向进行分类
    • 入站事件：连接激活/数据读取。。。。
    • 出站事件：打开到远程连接/写数据。。。。
  • 事件处理ChannelHandler
    • Channel通道发生数据或状态改变
    • EventLoop会将事件分类，并调用ChannelHandler的回调函数
    • 程序员需要实现ChannelHandler内的回调函数
    • ChannelInboundHandler/ChannelOutboundHandler
  • ChannelHandler工作模式：责任链
    • 责任链模式
      • 将请求的接受者连成一条链
      • 在链上传递请求，直到有一个接收者处理该请求
      • 避免请求者和接收者耦合
    • ChannelHandler 可以有多个，依次进行调用
    • ChannelPipline作为容器，承载多个ChannelHandler
  • ByteBuf
    • 强大的字节容器，提供丰富API进行操作

二、实战

服务端-main

package netty1;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import java.net.InetSocketAddress;
public class EchoServer {
    public static void main(String[] args) throws InterruptedException {
        int port = 8080;
        final EchoServerHandler serverHandler = new EchoServerHandler();
        EventLoopGroup group = new NioEventLoopGroup();
        try{
//            ServerBootstrap是Netty中的一个服务器引导类
            ServerBootstrap b = new ServerBootstrap();
            b.group(group)
                    .channel(NioServerSocketChannel.class)//设置通道类型
                    .localAddress(new InetSocketAddress(port))//设置监听端口
                    .childHandler(new ChannelInitializer<SocketChannel>() {//
                        @Override
                        public void initChannel(SocketChannel ch)
                        {
                            ch.pipeline().addLast(serverHandler);//添加处理类
                        }
            });
            ChannelFuture f = b.bind().sync();//开启监听
            if (f.isSuccess())
            {
                System.out.println(EchoServer.class.getName()+
                        " started and listening for connections on "+f.channel().localAddress());
            }
            f.channel().closeFuture().sync();
        }finally {
            group.shutdownGracefully().sync();
        }
    }
}

服务端-handler

package netty1;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;
public class EchoServerHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx,Object msg)
    {
        ByteBuf in = (ByteBuf) msg;
        String content = in.toString(CharsetUtil.UTF_8);
        System.out.println("Server recevied:"+content);
        ByteBuf out = ctx.alloc().buffer(1024);
        out.writeBytes((content+" 666").getBytes());
        ctx.write(out);
    }
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx)
    {
        ctx.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
    }
}

客户端-main

package netty1;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
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.net.InetSocketAddress;
public class EchoClient {
    public static void main(String[] args) throws InterruptedException {
        String host = "localhost";
        int port = 8080;
        EventLoopGroup group = new NioEventLoopGroup();
        try{
            Bootstrap b = new Bootstrap();
            b.group(group)
                    .channel(NioSocketChannel.class)
                    .remoteAddress(new InetSocketAddress(host,port))
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel ch)
                        {
                            ch.pipeline().addLast(new EchoClientHandler());
                        }
            });
            ChannelFuture f = b.connect().sync();
            f.channel().closeFuture().sync();
        }finally {
            group.shutdownGracefully().sync();
        }
    }
}

客户端-handler

package netty1;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.CharsetUtil;
public class EchoClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
    @Override
    public void channelActive(ChannelHandlerContext ctx)
    {
        ctx.writeAndFlush(Unpooled.copiedBuffer("Netty rocks!", CharsetUtil.UTF_8));
    }
    @Override
    public void channelRead0(ChannelHandlerContext ctx,ByteBuf in)
    {
        System.out.println("Client received:"+in.toString(CharsetUtil.UTF_8));
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx,Throwable cause)
    {
        cause.printStackTrace();
        ctx.close();
    }
}

执行结果

先执行服务端，服务端打印
netty1.EchoServer started and listening for connections on /0:0:0:0:0:0:0:0:8080
再执行客户端，客户端打印
Client received:Netty rocks! 666
服务端打印
Server recevied:Netty rocks!