Netty系列--简介

Dlimeng

发布于 2023-06-29 16:28:46

2100

文章被收录于专栏：开源心路开源心路

从编程语言层面

BIO | NIO | AIO 以Java的角度，理解如下：

BIO,同步阻塞式IO,简单理解：一个线程处理一个连接，发起和处理IO请求都是同步的。

NIO,同步非阻塞IO, 简单理解：一个线程处理多个连接，发起IO请求是非阻塞的但处理IO请求是同步的

AIO，异步非阻塞IO，简单理解：一个有效请求一个线程，发起和处理IO请求都是异步的

NIO的优势不在于速度

1、分散与聚集读取

2、文件锁定功能

3、网络异步IO

NIO示例

Server:

public static void main(String[] args) throws IOException {
    ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
    ServerSocket ss = serverSocketChannel.socket();
    ss.setReuseAddress(true);//地址重用
    ss.bind(new InetSocketAddress("localhost",9026));
    System.out.println("监控端口："+new InetSocketAddress("localhost",9026).toString());

    //分配一个新的字节缓冲区
    ByteBuffer allocate = ByteBuffer.allocate(4096);

    //读取数据
    while (true){
        SocketChannel channel = serverSocketChannel.accept();//接收数据
        System.out.println("accepted:"+channel);
        channel.configureBlocking(true);//设置阻塞，接不到就停
        int nread=0;
        while(nread !=-1){
            try {
                System.out.println("往缓冲区里读");
                nread = channel.read(allocate); //往缓冲区里读
                byte[] array=allocate.array();//数据转换array

                //打印
                String string = new String(array,0,allocate.position());
                System.out.println(string);
                allocate.clear();

            }catch (Exception e){
                e.printStackTrace();
                nread = -1;
            }
        }
    }
}

Client:

public static void main(String[] args) throws IOException {
    long start = System.currentTimeMillis();
    //打开socket的nio管道
    SocketChannel sc = SocketChannel.open();
    sc.connect(new InetSocketAddress("localhost",9026));//绑定
    sc.configureBlocking(true);//设置阻塞

    //将文件放到channel中
    FileChannel channel = new FileInputStream("/Users/limeng/Public/aa.txt").getChannel();

    //做好标记量
    long size=channel.size();
    int pos=0;
    int offset=4096;
    long curnset=0;
    long counts=0;

    //循环写
    while (pos < size){
        //把文件直接读取到socket chanel中，返回文件大小
        curnset=channel.transferTo(pos,4096,sc);

        pos+=offset;
        counts+=curnset;

    }
    //关闭
    channel.close();
    sc.close();

    //打印传输字节数
    System.out.println(counts);

    //打印时间
    System.out.println("bytes send--"+counts+"and totaltime--"+(System.currentTimeMillis() - start));

}

Netty

简介

Netty是基于Java NIO的网络应用框架.

Netty是一个NIO client-server(客户端服务器)框架，使用Netty可以快速开发网络应用，例如服务器和客户端协议。Netty提供了一种新的方式来使开发网络应用程序，这种新的方式使得它很容易使用和有很强的扩展性。Netty的内部实现时很复杂的，但是Netty提供了简单易用的api从网络处理代码中解耦业务逻辑。Netty是完全基于NIO实现的，所以整个Netty都是异步的。

网络应用程序通常需要有较高的可扩展性，无论是Netty还是其他的基于Java NIO的框架，都会提供可扩展性的解决方案。Netty中一个关键组成部分是它的异步特性.

优点：

API使用简单，开发门槛低;

功能强大，预置了多种编解码功能，支持多种主流协议；

定制能力强，可以通过ChannelHandler对通信框架进行灵活的扩展;

性能高，通过与其他业界主流的NIO框架对比，netty的综合性能最优；

稳定性好

社区活跃度高

大规模商用

Netty示例

Server:

配置服务器功能，如线程，端口。实现服务器处理程序，它包含业务逻辑，决定当有一个请求连接或接收数据时该做什么

private final int port;

public EchoServer(int port) {
    this.port = port;
}
public void start() throws Exception{
    EventLoopGroup eventLoopGroup=null;
    try {
        //server端引导类
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        //连接池处理数据
        eventLoopGroup = new NioEventLoopGroup();
        //装配bootstrap
        serverBootstrap.group(eventLoopGroup)
                .channel(NioServerSocketChannel.class)//指定通道类型为NioServerSocketChannel，一种异步模式，OIO阻塞模式为OioServerSocketChannel
                .localAddress("localhost",port) //设置InetSocketAddress让服务器监听某个端口已等待客户端连接
                .childHandler(new ChannelInitializer<Channel>() {//设置childHandler执行所有的连接请求
                    @Override
                    protected void initChannel(Channel ch) throws Exception {
                        ch.pipeline().addLast(new EchoServerHandler());//注册handler
                    }
                });

        ChannelFuture channelFuture = serverBootstrap.bind().sync();
        System.out.println("开始监听，端口为：" + channelFuture.channel().localAddress());
        channelFuture.channel().closeFuture().sync();

    }finally {
        if (eventLoopGroup != null) {
            eventLoopGroup.shutdownGracefully().sync();
        }
    }
}

public static void main(String[] args) throws Exception {
    new EchoServer(20000).start();
}

ServerHandler:

public class EchoServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("server 读取数据......");
        //读取数据
        ByteBuf buf =(ByteBuf)msg;
        byte[] req=new byte[buf.readableBytes()];
        buf.readBytes(req);
        String body = new String(req, "UTF-8");
        System.out.println("接收客户端数据："+body);
        //向客户端写数据
        System.out.println("server向client发送数据");
        String currentTime = new Date(System.currentTimeMillis()).toString();
        ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
        ctx.write(resp);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        System.out.println("server 读取数据完毕..");
        ctx.flush();
    }

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

Client:

连接服务器写数据到服务器等待接受服务器返回相同的数据关闭连接

private final String host;
private final int port;

public EchoClient(String host, int port) {
    this.host = host;
    this.port = port;
}
public void start() throws Exception{
    EventLoopGroup eventLoopGroup=null;
    try {
        //客户端引导类
        Bootstrap bootstrap = new Bootstrap();
        //EventLoopGroup 可以理解为一个线程池，这个线程池用来处理数据，发送数据，接收数据
        eventLoopGroup=new NioEventLoopGroup();

        bootstrap.group(eventLoopGroup)//多线程处理
                .channel(NioSocketChannel.class)//指定通道类型为NioServerSocketChannel，一种异步模式，OIO阻塞模式为OioServerSocketChannel
                .remoteAddress(new InetSocketAddress(host,port))//地址
                .handler(new ChannelInitializer<SocketChannel>() {//业务处理类

                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new EchoClientHandler());//注册handler
                    }
                });

        // 链接服务器
        ChannelFuture channelFuture = bootstrap.connect().sync();
        channelFuture.channel().closeFuture().sync();
    }finally {
        if (eventLoopGroup != null) {
            eventLoopGroup.shutdownGracefully().sync();
        }
    }
}

public static void main(String[] args) throws Exception {
    new EchoClient("localhost", 20000).start();
}

ClientHandler:

public class EchoClientHandler extends SimpleChannelInboundHandler{
    //客户端连接服务请后被调用
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("客户端连接");
        String time=new Date().toString();
        String req2=time+" QUERY TIME ORDER";
        byte[] req=req2.getBytes();//消息
        ByteBuf buffer = Unpooled.buffer(req.length);
        buffer.writeBytes(req);//发送
        ctx.writeAndFlush(buffer);//flush
    }

    // 发生异常时被调用
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println("client exceptionCaught...");
        //释放资源
        ctx.close();
    }
    //从服务器接收到数据后调用
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("client 读取server数据..");
        //服务端返回消息后
        ByteBuf buf=(ByteBuf)msg;
        byte[] req = new byte[buf.readableBytes()];
        buf.readBytes(req);
        String body = new String(req, "UTF-8");
        System.out.println("服务端数据为："+body);

    }
}