Code Ease Code Ease
  • 个人博客网站 (opens new window)
  • 好用的工具网站 (opens new window)
  • Java核心基础
  • 框架的艺术
  • 分布式与微服务
  • 开发经验大全
  • 设计模式
  • 版本新特性
数据库系列
大数据+AI
  • xxl-job
运维与Linux
  • 基于SpringBoot和BootStrap的论坛网址
  • 基于VuePress的个人博客网站
  • 基于SpringBoot开发的小功能
  • 做一个自己的IDEA插件
程序人生
关于我
  • 分类
  • 标签
  • 归档

神秘的鱼仔

你会累是因为你在走上坡路
  • 个人博客网站 (opens new window)
  • 好用的工具网站 (opens new window)
  • Java核心基础
  • 框架的艺术
  • 分布式与微服务
  • 开发经验大全
  • 设计模式
  • 版本新特性
数据库系列
大数据+AI
  • xxl-job
运维与Linux
  • 基于SpringBoot和BootStrap的论坛网址
  • 基于VuePress的个人博客网站
  • 基于SpringBoot开发的小功能
  • 做一个自己的IDEA插件
程序人生
关于我
  • 分类
  • 标签
  • 归档
服务器
  • Java核心基础

  • 框架的艺术

    • Spring

    • Mybatis

    • SpringBoot

    • MQ

    • Zookeeper

    • netty

      • 网络开发的最强大框架:Netty快速入门
        • 什么是netty
        • 第一个 Netty 入门程序
        • 理解Netty中的组件
          • EventLoop
          • EventLoopGroup
        • Bootstrap
          • Channel
          • ChannelHandler
          • ByteBuf
        • 一、EventLoop
        • 二、Channel
          • Channel 的连接
          • Channel 的关闭
        • 三、ChannelHandler
        • 四、ByteBuf
          • 1、自动扩容
          • 2、直接内存和堆内存
          • 3、池化技术
        • 总结
      • 粘包和半包有了解过吗?netty是如何解决这个问题的
      • 如何使用Netty实现Http服务器
  • 分布式与微服务

  • 开发经验大全

  • 版本新特性

  • Java
  • 框架的艺术
  • netty
CodeEase
2023-10-09
目录

网络开发的最强大框架:Netty快速入门

作者:鱼仔
博客首页: codeease.top (opens new window)
公众号:Java鱼仔

# 什么是netty

Netty 是一个异步的,基于事件驱动的网络应用框架,用于快速开发可维护、高性能的网络服务器和客户端。Netty 的应用十分广泛,可以说主流的框架中,如果有网络方面的需求,一般用的都是 Netty 框架。比如 Dubbo、ES、**Zookeeper **中都用到了 Netty。因此即使在平常工作中没有 Netty 的使用场景,Netty 还是十分值得我们去学习的。

Netty 底层基于 NIO 开发,其实大部分的 Java 程序员对于网络方面的开发能力是比较弱的,因此如果有网络相关的开发业务,如果自己通过 BIO 或者 NIO 实现,会产生很多问题。而通过 Netty 可以快速开发网络应用,因此也有人把 Netty 称为网络开发框架中的 Spring。

关于 NIO 和 BIO 的区别,我之前在博客中也讲到过,BIO 每次通信都要新建一个线程去处理,NIO 通过多路复用的方式去处理请求。下图就是 NIO 的处理流程。

1-1.jpeg

# 第一个 Netty 入门程序

既然基于 NIO 开发,Netty 的入门程序和我们当时写的 NIO 入门程序比较像,首先开发一个服务器端,Netty 的开发流程可以遵循一套规范:

  1. 通过 ServerBootstrap 启动,组装 Netty 组件
  2. 组装 EventLoopGroup
  3. 组装 Channel
  4. 通过 Handler 处理连接、读写请求
public class FirstServer {
    public static void main(String[] args) {
        // 创建两个NioEventLoopGroup实例,分别用于处理boss(接受连接)和worker(处理连接)任务
        NioEventLoopGroup workGroup = new NioEventLoopGroup();
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();

        try {
            // 创建一个新的ServerBootstrap实例,用于配置服务器
            ChannelFuture future = new ServerBootstrap()
                    // 指定bossGroup和workGroup,分别用于接受连接和处理连接
                    .group(bossGroup, workGroup)
                    // 指定使用的通道类型为NioServerSocketChannel,适用于NIO传输
                    .channel(NioServerSocketChannel.class)
                    // 设置子通道的处理器
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            // 获取通道的pipeline,用于添加处理器
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            // 添加一个字符串解码器,将ByteBuf转换为字符串
                            pipeline.addLast(new StringDecoder());
                            // 添加一个简单的通道处理器,处理接收到的字符串消息
                            pipeline.addLast(new SimpleChannelInboundHandler<String>() {
                                @Override
                                protected void channelRead0(ChannelHandlerContext channelHandlerContext, String msg) throws Exception {
                                    // 打印接收到的消息
                                    System.out.println(msg);
                                }
                            });
                        }
                    }).bind(8080).sync();
            // 等待服务器socket通道关闭
            future.channel().closeFuture().sync();
        } catch (Exception e) {
            // 处理可能的异常
            e.printStackTrace();
        } finally {
            // 优雅地关闭bossGroup和workGroup,释放资源
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

接着开发一个客户端,客户端的整体流程和服务器端十分类似:

1、通过 Bootstrap 启动,组装 Netty 组件

2、组装 EventLoopGroup

3、组装 Channel

4、添加 Handler 处理器

5、建立连接

6、发送数据到服务端


public class FirstClient {
    public static void main(String[] args) {
        NioEventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap()
                    .group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                                @Override
                                public void channelActive(ChannelHandlerContext ctx) throws Exception {
                                    System.out.println("client " + ctx);
                                    ctx.writeAndFlush(Unpooled.copiedBuffer("客户端发了一条消息", CharsetUtil.UTF_8));
                                }
                                
                                @Override
                                public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
                                    cause.printStackTrace();
                                    ctx.close();
                                }
                            });
                        }
                    });
            System.out.println("客户端 ok..");
            //启动客户端去连接服务器端
            //关于 ChannelFuture 要分析,涉及到netty的异步模型
            ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 8080).sync();
            //给关闭通道进行监听
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } finally {
            group.shutdownGracefully();
        }

    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

启动服务器端后再启动客户端,可以发现服务端接受到了客户端发过来的信息。看到这里觉得还是有点蒙没关系,下面会对每个组件进行讲解。

# 理解Netty中的组件

# EventLoop

EventLoop 其实是一个单线程的执行器,同时维护了一个 Selector,EventLoop 的主要任务是处理网络事件,包括 IO 事件、定时任务和用户自定义任务。

# EventLoopGroup

EventLoopGroup 是一组 EventLoop 的集合,管理一组 EventLoop 实例。它负责分配和管理这些事件循环线程。

Netty 服务端通常分为两组线程池(事件组)来处理不同类型的任务,BossEventLoopGroup 和 WorkerEventLoopGroup。

BossEventLoopGroup 负责监听 ServerSocketChannel 的连接事件,接收到新连接后,将其注册到 WorkerEventGroup。

WorkerEventLoopGroup 负责处理与已建立连接相关的所有 IO 事件(如读写操作)。

# Bootstrap

Bootstrap 的作用是串联整个 Netty 程序,同时他也是 Netty 程序的入口。ServerBootstrap 是 Netty 服务端的入口,Bootstrap 是 Netty 客户端的入口。

# Channel

Channel 是一个数据的传输流,Channel 可以理解为是通讯的载体。Channel可以提供下面一系列功能

网络I/O操作

  • 读取数据:从远程端接收数据。
  • 写入数据:将数据发送到远程端。

管理连接状态

  • 检测连接是否打开、是否激活等。
  • 提供关闭连接的操作。

关联事件循环

  • 每个 Channel 与一个 EventLoop 绑定,用于处理 I/O 操作和事件通知。

支持异步操作

  • Netty 的 I/O 操作(如写入或关闭连接)通常是异步的,Channel 提供了 ChannelFuture 接口,用于监听操作完成的结果。

常用的Channel

  • NioSocketChannel,异步的客户端 TCP Socket 连接。
  • NioServerSocketChannel,异步的服务器端 TCP Socket 连接。
  • NioDatagramChannel,异步的 UDP 连接。

# ChannelHandler

ChannelHandler 是用来处理 Channel 上的各种事件的,所有的 ChannelHandler 连起来就是 pipeline。简单来讲,Channel 是数据的传输通道,而 ChannelHandler 用来处理通道中的数据。

# ByteBuf

ByteBuf 是 netty 中数据的传输载体,网络数据的基本单位总是字节,ByteBuf 用来传输这些网络上的字节。

# 一、EventLoop

EventLoop可以处理多种任务,单独使用EventLoop可以通过下面几个步骤实现:

1、创建一个EventLoopGroup

2、从EventLoopGroup中获取EventLoop

3、通过EventLoop执行任务

通过代码这样表示:

public class TestEventLoop {
    public static void main(String[] args) {
        //1、创建事件循环组
        //NioEventLoopGroup可以处理IO事件、普通任务、定时任务
        EventLoopGroup group=new NioEventLoopGroup();
        //2、获取下一个事件循环对象
        EventLoop eventLoop = group.next();
        //3、执行普通任务
        eventLoop.execute(()->{
            System.out.println("普通任务");
        });
        //4、执行定时任务
        eventLoop.scheduleAtFixedRate(()->{
            System.out.println("定时任务");
        },0,1, TimeUnit.SECONDS);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

EventLoop最常用的就是执行IO任务了,我们在入门程序中写的group(new NioEventLoopGroup())就是把EventLoop用来处理IO任务。

在netty中,我们还会在绑定group时指定boss和work,boss用来处理连接,work用来处理收到读写请求后续的操作,有的时候我们还可以自定义EventLoopGroup处理其他任务,因此前面的FirstServer 可以写成下面这样:

public class NioServer {
    public static void main(String[] args) {
        //boss用来处理连接
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();
        //work用来处理读写请求
        NioEventLoopGroup workGroup = new NioEventLoopGroup();
        //otherGroup处理普通任务,比如打印一段内容
        EventLoopGroup otherGroup=new DefaultEventLoop();
        
        new ServerBootstrap()
                .group(bossGroup,workGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
                        nioSocketChannel.pipeline().addLast(new ChannelInboundHandlerAdapter(){
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                ByteBuf byteBuf = (ByteBuf) msg;
                                System.out.println(byteBuf.toString());
                                ctx.fireChannelRead(msg); //将msg传给下一个处理者
                            }
                        })
                        .addLast(otherGroup,"handler",new ChannelInboundHandlerAdapter(){
                            @Override
                            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                ByteBuf byteBuf = (ByteBuf) msg;
                                System.out.println(byteBuf.toString());
                            }
                        });
                    }
                }).bind(8080);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

# 二、Channel

Channel 中有几个常用的方法:

close() 关闭channel
pipeline() 添加处理器
write() 将数据写入到缓冲区
flush() 将数据刷出,也就是发给服务端
writeAndFlush() 将数据写入并刷出
1
2
3
4
5

我们通过入门案例的客户端代码讲解 Channel

# Channel 的连接

public class NioClient  {
    public static void main(String[] args) throws InterruptedException {
        ChannelFuture channelFuture = new Bootstrap()
                .group(new NioEventLoopGroup())
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
                        nioSocketChannel.pipeline().addLast(new StringEncoder());
                    }
                })
                //connect是一个异步调用的过程,因此必须要使用sync方法等待连接建立
                .connect(new InetSocketAddress("localhost", 8080));
        //1、使用sync方法阻塞线程直到连接建立
        channelFuture.sync();
        channelFuture.channel().writeAndFlush("hello,world");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

整个流程这里就不介绍了,主要介绍里面的一个方法 channelFuture.sync();
当调用 connect 方法建立连接时,这个 connect 方法其实是一个异步的方法,因此如果不加 channelFuture.sync() 方法等待连接建立,是无法获取到连接后的 channel 的,更别提写入数据了。

除了使用sync等待连接,还可以采用设置监听器的方式获取 channelFuture

public static void main(String[] args) throws InterruptedException {
    ChannelFuture channelFuture = new Bootstrap()
            .group(new NioEventLoopGroup())
            .channel(NioSocketChannel.class)
            .handler(new ChannelInitializer<NioSocketChannel>() {
                @Override
                protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
                    nioSocketChannel.pipeline().addLast(new StringEncoder());
                }
            })
            //connect是一个异步调用的过程,因此必须要使用sync方法等待连接建立
            .connect(new InetSocketAddress("localhost", 8080));
    //2、使用addListener方法异步处理结果
    channelFuture.addListener(new ChannelFutureListener() {
        //在nio连接建立完毕之后,调用operationComplete方法
        @Override
        public void operationComplete(ChannelFuture channelFuture) throws Exception {
            Channel channel = channelFuture.channel();
            channel.writeAndFlush("hello,world");
        }
    });
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

思路是一样的,等连接建立之后再处理对应的方法。

# Channel 的关闭

除了连接是异步方法之外,Channel 的关闭方法也是异步的,因此也需要通过

同步阻塞的方式等待关闭:
Channel channel = channelFuture.channel();
ChannelFuture closeFuture = channel.closeFuture();
System.out.println("等待关闭中");
//当其他线程关闭了channel,sync同步等待
closeFuture.sync();
System.out.println("连接已关闭");
1
2
3
4
5
6
7

同样也可以采用监听器回调的方式:

Channel channel = channelFuture.channel();
ChannelFuture closeFuture = channel.closeFuture();
closeFuture.addListener(new ChannelFutureListener() {
    @Override
    public void operationComplete(ChannelFuture channelFuture) throws Exception {
        System.out.println("连接已关闭");
    }
});
1
2
3
4
5
6
7
8

# 三、ChannelHandler

ChannelHandler 是用来处理 Channel 上的各种事件的,handler 分为 **inbound **和 **outbount **两种,所有的ChannelHandler 连起来就是 pipeline。

**ChannelInboundHandlerAdapter **的子类主要用来读取客户端数据,写回结果。

**ChannelOutboundHandlerAdapter **的字类主要对写回结果进行加工。

关于 handler 和 pipeline 的代码在前面的例子中都有写

public static void main(String[] args) {
    new ServerBootstrap()
            .group(new NioEventLoopGroup())
            .channel(NioServerSocketChannel.class)
            .childHandler(new ChannelInitializer<NioSocketChannel>() {
                @Override
                protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
                     //1、从channel中获取到pipeline
                     ChannelPipeline pipeline = nioSocketChannel.pipeline();
                     //2、添加handler处理器到pipeline
                    pipeline.addLast(new ChannelInboundHandlerAdapter(){
                        @Override
                        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                            ByteBuf byteBuf = (ByteBuf) msg;
                            System.out.println(byteBuf.toString());
                            ctx.fireChannelRead(msg); //将msg传给下一个处理者
                        }
                    });
                    //3、添加多个表示依次执行
                    pipeline.addLast(new ChannelInboundHandlerAdapter(){
                        @Override
                        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                            ByteBuf byteBuf = (ByteBuf) msg;
                            System.out.println(byteBuf.toString());
                        }
                    });
                }
            }).bind(8080);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

# 四、ByteBuf

netty中的ByteBuf比JDK自带的ByteBuffer对字节数据的操作更加友好,也更加强大。ByteBuf的主要优势有几下几点:

1、支持自动扩容

2、支持池化技术,可以重用实例,节约内存

3、读写指针分离

4、很多方法体现了零拷贝,比如slice、duplicate等

接下来通过一些操作带你来了解ByteBuf。

# 1、自动扩容

创建一个默认的ByteBuf,初始容量是256,写入一系列数据之后,这个容量会随着数据的增大自动扩容。

public static void main(String[] args) {
    ByteBuf buf= ByteBufAllocator.DEFAULT.buffer();
    System.out.println(buf);
    StringBuilder stringBuilder=new StringBuilder();
    for (int i = 0; i < 500; i++) {
        stringBuilder.append("1");
    }
    buf.writeBytes(stringBuilder.toString().getBytes());
    System.out.println(buf);
}
1
2
3
4
5
6
7
8
9
10

结果:

PooledUnsafeDirectByteBuf(ridx: 0, widx: 0, cap: 256)
PooledUnsafeDirectByteBuf(ridx: 0, widx: 500, cap: 512)
1
2

ByteBuf的扩容规则如下:

如果数据大小没有超过512,每次扩容到16的整数倍

如果数据大小超过512,则扩容到下一个2^n次

扩容不能超过max capacity

# 2、直接内存和堆内存

ByteBuf支持创建基于直接内存的ByteBuf,也支持创建基于堆内存的ByteBuf。两者的差距在于:

堆内存的分配效率较高,但是读写性能相对比较低。

直接内存的分配效率比较低,但是读写性能较高(少一次内存复制)

netty默认使用直接内存作为创建ByteBuf的方式

ByteBufAllocator.DEFAULT.heapBuffer();
ByteBufAllocator.DEFAULT.directBuffer();
1
2

# 3、池化技术

ByteBuf支持池化技术,所谓池化指的是ByteBuf创建出来后可以重用,节约内存。通过JVM参数开启或关闭netty的池化,默认开启状态:

-Dio.netty.allocator.type={unpooled|pooled}
1

# 总结

Netty是个很强大的框架,但是网络开发本就是一件比较复杂的事情,接下来我会用Netty做一些简单的应用出来,通过这些应用会让Netty更加容易理解一些。我是鱼仔,我们下期再见!

上次更新: 2025/04/29, 17:22:06
Zookeeper实现分布式锁的原理是什么?
粘包和半包有了解过吗?netty是如何解决这个问题的

← Zookeeper实现分布式锁的原理是什么? 粘包和半包有了解过吗?netty是如何解决这个问题的→

最近更新
01
AI大模型部署指南
02-18
02
半个月了,DeepSeek为什么还是服务不可用
02-13
03
Python3.9及3.10安装文档
01-23
更多文章>
Theme by Vdoing | Copyright © 2023-2025 备案图标 浙公网安备33021202002405 | 浙ICP备2023040452号
  • 跟随系统
  • 浅色模式
  • 深色模式
  • 阅读模式