写在文章开头
由于Netty
优秀的设计和封装,开发一个高性能网络程序就变得非常简单,本文从一个简单的服务端落地简单介绍一下Netty
中的几个核心组件,希望对你有帮助。
Hi,我是 sharkChili ,是个不断在硬核技术上作死的 java coder ,是 CSDN的博客专家 ,也是开源项目 Java Guide 的维护者之一,熟悉 Java 也会一点 Go ,偶尔也会在 C源码 边缘徘徊。写过很多有意思的技术博客,也还在研究并输出技术的路上,希望我的文章对你有帮助,非常欢迎你关注我的公众号: 写代码的SharkChili 。
因为近期收到很多读者的私信,所以也专门创建了一个交流群,感兴趣的读者可以通过上方的公众号获取笔者的联系方式完成好友添加,点击备注 “加群” 即可和笔者和笔者的朋友们进行深入交流。
快速落地一个服务端
我们希望通过Netty
快速落地一个简单的主从reactor
模型,由主reactor
对应的线程组接收连接交由acceptor
创建连接,与之建立的客户端的读写事件都会交由从reactor
对应的线程池处理:
基于此设计,我们通过Netty
写下如下代码,可以看到我们做了如下几件事:
- 声明一个服务端创建引导类
ServerBootstrap
,负责配置服务端及其启动工作。 - 声明主从
reactor
线程组,其中boss
可以看作监听端口接收新连接的线程组,而work
则是负责处理客户端数据读写的线程组。 - 基于上述线程池作为
group
的入参完成主从reactor
模型创建。 - 通过
channel
函数指定server channe
为NioServerSocketChannel
即采用NIO
模型,而NioServerSocketChannel
我们可以直接理解为serverSocket
的抽象表示。 - 通过
childHandler
方法给引导设置每一个连接数据读写的处理器handler
。
最后调用bind
启动服务端并通过addListener
对连接结果进行异步监听:
public static void main(String[] args) {
//1. 声明一个服务端创建引导类
ServerBootstrap serverBootstrap = new ServerBootstrap();
//2. 声明主从reactor线程组
NioEventLoopGroup boss = new NioEventLoopGroup();
NioEventLoopGroup worker = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors());
serverBootstrap.group(boss, worker)//3. 基于上述线程池创建主从reactor模型
.channel(NioServerSocketChannel.class)//server channel采用NIO模型
.childHandler(new ChannelInitializer<NioSocketChannel>() {//添加客户端读写请求处理器到subreactor中
@Override
protected void initChannel(NioSocketChannel ch) throws Exception {
// 对于ChannelInboundHandlerAdapter,收到消息后会按照顺序执行即 A -> B->ServerHandler
ch.pipeline().addLast(new InboundHandlerA())
.addLast(new InboundHandlerB())
.addLast(new ServerHandler());
// 处理写数据的逻辑,顺序是反着的 B -> A
ch.pipeline().addLast(new OutboundHandlerA())
.addLast(new OutboundHandlerB())
.addLast(new OutboundHandlerC());
ch.pipeline().addLast(new ExceptionHandler());
}
});
//绑定8080端口并设置回调监听结果
serverBootstrap.bind("127.0.0.1", 8080)
.addListener(f -> {
if (f.isSuccess()) {
System.out.println("连接成功");
}
});
}
对于客户端的发送的数据,我们都会通过ChannelInboundHandlerAdapter
添加顺序处理,就如代码所示我们的执行顺序为InboundHandlerA
->InboundHandlerB
->ServerHandler
,对此我们给出InboundHandlerA
的代码,InboundHandlerB
内容一样就不展示了:
public class InboundHandlerA extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("InBoundHandlerA : " + ((ByteBuf)msg).toString(StandardCharsets.UTF_8));
//将当前的处理过的msg转交给pipeline的下一个ChannelHandler
super.channelRead(ctx, msg);
}
}
而ServerHandler
的则是:
- 客户端与服务端建立连接,对应客户端channel被激活,触发
channelActive
方法。 ChannelHandlerContext
的Channel
已注册到其EventLoop
中,执行channelRegistered
。- 将
ChannelHandler
添加到实际上下文并准备好处理事件后调用。 - 解析客户端的数据然后回复
Hello Netty client
:
private static class ServerHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelActive(ChannelHandlerContext ctx) {
System.out.println("channel被激活,执行channelActive");
}
@Override
public void channelRegistered(ChannelHandlerContext ctx) {
System.out.println("执行channelRegistered");
}
@Override
public void handlerAdded(ChannelHandlerContext ctx) {
System.out.println("执行handlerAdded");
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf byteBuf = (ByteBuf) msg;
//打印读取到的数据
System.out.println(new Date() + ": 服务端读到数据 -> " + byteBuf.toString(StandardCharsets.UTF_8));
// 回复客户端数据
System.out.println(new Date() + ": 服务端写出数据");
//组装数据并发送
ByteBuf out = getByteBuf(ctx);
ctx.channel().writeAndFlush(out);
super.channelRead(ctx, msg);
}
private ByteBuf getByteBuf(ChannelHandlerContext ctx) {
ByteBuf buffer = ctx.alloc().buffer();
byte[] bytes = "Hello Netty client ".getBytes(StandardCharsets.UTF_8);
buffer.writeBytes(bytes);
return buffer;
}
//......
}
我们通过telnet
对应ip和端口后发现服务端输出如下内容,也很我们上文说明的一致:
执行handlerAdded
执行channelRegistered
端口绑定成功,channel被激活,执行channelActive
然后我们发送消息 1,可以看到触发所有inbound
的channelRead
方法:
InBoundHandlerA : 1
InBoundHandlerB: 1
Wed Jul 24 00:05:18 CST 2024: 服务端读到数据 -> 1
然后我们回复hello netty client
,按照添加的倒叙触发OutBoundHandler
:
Wed Jul 24 00:05:18 CST 2024: 服务端写出数据
OutBoundHandlerC: Hello Netty client
OutBoundHandlerB: Hello Netty client
OutBoundHandlerA: Hello Netty client
详解Netty中的核心组件
channel接口
channel是Netty对于底层class socket中的bind、connect、read、write等原语的封装,简化了我们网络编程的复杂度,同时Netty也提供的各种现成的channel,我们可以根据个人需要自行使用。
下面便是笔者比较常用的Tcp或者UDP中比较常用的几种channel。
NioServerSocketChannel
:基于NIO
选择器处理新连接。EpollServerSocketChannel
:使用linux EPOLL Edge
触发模式实现最大性能的实现。NioDatagramChannel
:发送和接收AddressedEnvelope
的 NIO 数据报通道。EpollDatagramChannel
:使用linux EPOLL Edge
触发模式实现最大性能的DatagramChannel
实现。
EventLoop接口
在Netty
中,所有channel都会注册到某个eventLoop
上, 每一个EventLoopGroup
中有一个或者多个EventLoop
,而每一个EventLoop
都绑定一个线程,负责处理一个或者多个channel
的事件:
这里我们也简单的给出NioEventLoop
中的run
方法,它继承自SingleThreadEventExecutor
,我们可以大概看到NioEventLoop
的核心逻辑本质就是轮询所有注册到NioEventLoop上的channel(socket的抽象)是否有其就绪的事件,然后
@Override
protected void run() {
for (;;) {
try {
//基于selectStrategy轮询查看是否有就绪事件
switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
case SelectStrategy.CONTINUE:
continue;
case SelectStrategy.SELECT:
select(wakenUp.getAndSet(false));
//......
if (wakenUp.get()) {
selector.wakeup();
}
default:
// fallthrough
}
cancelledKeys = 0;
needsToSelectAgain = false;
final int ioRatio = this.ioRatio;
//根据IO配比执行网络IO事件方法processSelectedKeys以及其他事件方法runAllTasks
if (ioRatio == 100) {
try {
processSelectedKeys();
} finally {
// Ensure we always run tasks.
runAllTasks();
}
} else {
final long ioStartTime = System.nanoTime();
try {
processSelectedKeys();
} finally {
// Ensure we always run tasks.
final long ioTime = System.nanoTime() - ioStartTime;
runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
}
}
} catch (Throwable t) {
handleLoopException(t);
}
//......
}
}
pipeline和channelHandler以channelHandlerContext
每一个channel
的事件都会交由channelHandler
处理,而负责同一个channel
的channelHandler
都会交由pipeline
一条逻辑链进行连接,这两者的的关系都会一一封装成channelHandlerContext
,channelHandlerContext
主要是负责当前channelHandler
和与其同一条channelpipeline
上的其他channelHandler
之间的交互。
举个例子,当我们收到客户端的写入数据时,这些数据就会交由pipeline
上的channelHandler
处理,如下图所示,从第一个channelHandler
处理完成之后,每个channelHandlerContext
就会将消息转交到当前pipeline
的下一个channelHandler
处理:
假设我们的channelHandler
执行完ChannelActive
后,如希望继续传播则会调用fireChannelActive
:
@Override
public void channelActive(ChannelHandlerContext ctx) {
System.out.println("端口绑定成功,channel被激活,执行channelActive");
ctx.fireChannelActive()
}
查看其内部逻辑即可知晓,它就是通过AbstractChannelHandlerContext
得到pipeline
的下一个ChannelHandler
并执行其channelActive
方法:
@Override
public ChannelHandlerContext fireChannelActive() {
final AbstractChannelHandlerContext next = findContextInbound();
invokeChannelActive(next);
return this;
}
回调的思想
我们可以说回调其实是一种设计思想,Netty
对于连接或者读写操作都是异步非阻塞的,所以我们希望在连接被建立进行一些响应的处理,那么Netty
就会在连接建立的时候方法暴露一个回调方法供用户实现个性逻辑。
例如我们的channel
连接被建立时,其底层就会调用invokeChannelActive
获取我们绑定的ChannelInboundHandler
并执行其channelActive
方法:
private void invokeChannelActive() {
if (invokeHandler()) {
try {
((ChannelInboundHandler) handler()).channelActive(this);
} catch (Throwable t) {
notifyHandlerException(t);
}
} else {
fireChannelActive();
}
}
于是就会调用到我们服务端ServerHandler
的channelActive
方法:
private static class ServerHandler extends ChannelInboundHandlerAdapter {
//......
@Override
public void channelActive(ChannelHandlerContext ctx) {
System.out.println("端口绑定成功,channel被激活,执行channelActive");
}
//......
}
Future异步监听
为保证网络服务器执行的效率,Netty大部分网络IO操作都采用异步的,以笔者建立连接设置的监听器为例,当前连接成功后,就会返回给监听器一个java.util.concurrent.Future
,我们就可以通过这个f获取连接的结果是否成功:
//绑定8080端口并设置回调监听结果
serverBootstrap.bind("127.0.0.1", 8080)
.addListener(f -> {
if (f.isSuccess()) {
System.out.println("连接成功");
}
});
我们步入DefaultPromise
的addListener
即可发现其内部就是添加监听后判断这个连接的异步任务Future是否完成,如果完成调用notifyListeners
回调我们的监听器的逻辑:
@Override
public Promise<V> addListener(GenericFutureListener<? extends Future<? super V>> listener) {
//......
//添加监听
synchronized (this) {
addListener0(listener);
}
//连接任务完成,通知监听器
if (isDone()) {
notifyListeners();
}
return this;
}
小结
以上便是笔者对于Netty服务端的基础使用和分析,希望对你有帮助。
我是 sharkchili ,CSDN Java 领域博客专家,开源项目—JavaGuide contributor,我想写一些有意思的东西,希望对你有帮助,如果你想实时收到我写的硬核的文章也欢迎你关注我的公众号: 写代码的SharkChili 。
因为近期收到很多读者的私信,所以也专门创建了一个交流群,感兴趣的读者可以通过上方的公众号获取笔者的联系方式完成好友添加,点击备注 “加群” 即可和笔者和笔者的朋友们进行深入交流。
参考
高性能网络编程之 Reactor 网络模型(彻底搞懂):https://blog.csdn.net/ldw201510803006/article/details/124365838
《Netty in action》
《跟闪电侠学Netty》