Netty源码分析
2023-12-27 15:02:50 33 举报AI智能生成
Netty源码分析
作者其他创作
大纲/内容
服务端启动
创建ServerSocketChannel
new NioServerSocketChannel()
newSocket(PROVIDER) -> ServerSocketChannel
this.readInterestOp = SelectionKey.OP_ACCEPT 保存关心的事件为Accept
初始化ServerSocketChannel
添加之后的结构
注册selector
AbstractChannel.this.eventLoop = eventLoop;
将分配到的EventLoop保存至chanel内部
将分配到的EventLoop保存至chanel内部
将Netty-Channel内部的jdk-Channel注册至该EventLoop绑定的selector上，此时并不关心事件
selectionKey = javaChannel().register(eventLoop().selector, 0, this);
selectionKey = javaChannel().register(eventLoop().selector, 0, this);
获取注册后的selectionKey作为Netty-Channel的成员变量
服务端口绑定
jdk-Channel绑定端口
javaChannel().bind(localAddress, config.getBacklog());
javaChannel().bind(localAddress, config.getBacklog());
绑定完毕，触发channelActive事件
HeadContext ctx.fireChannelActive();
HeadContext ctx.fireChannelActive();
触发完毕，调用HeadContext.readIfIsAutoRead();
传播至HeadContext.read()
-> unsafe.beginRead();
-> doBeginRead() 修改SelectionKey感兴趣的事件为创建时保存的兴趣事件
-> unsafe.beginRead();
-> doBeginRead() 修改SelectionKey感兴趣的事件为创建时保存的兴趣事件
创建时保存感兴趣事件，创建jdk-Channel并创建id, unsafe, pipeline
服务端初始化主要用于在pipeline添加ServerBootstrapAcceptor处理器
用于将监听到的客户端Channel注册至客户端Group中
用于将监听到的客户端Channel注册至客户端Group中
注册主要目的用于将
register() -> Netty-Channel注册至EventLoop (表现行为为保存EventLoop引用)
register0() -> jdk-Channel注册至selector上 (通过jdk的方式注册)
register() -> Netty-Channel注册至EventLoop (表现行为为保存EventLoop引用)
register0() -> jdk-Channel注册至selector上 (通过jdk的方式注册)
全部由HeadContext节点处理
NioEventLoop
创建
executor = new ThreadPerTaskExecutor(newDefaultThreadFactory())
factory -> new FastThreadLocalThread(threadGroup, r, name)
execute -> threadFactory.newThread(command).start()
this.executor = ObjectUtil.checkNotNull(executor, "executor");
taskQueue = newTaskQueue(this.maxPendingTasks);
selector = openSelector();
chooser = chooserFactory.newChooser(children);
EventExecutorChooser.next()轮询选择EventLoop
启动
SingleThreadEventExecutor.execute(task)
this.thread为null -> 非nio线程 -> doStartThread()
executor.execute()创建FastThreadLocalThread
thread = Thread.currentThread();
保存由executor分配的线程(FastThreadLocalThread)
保存由executor分配的线程(FastThreadLocalThread)
SingleThreadEventExecutor.this.run()，在该线程中正式启动EventLoop
addTask(task) -> taskQueue.offer(task) 添加此任务由eventloop执行
执行逻辑
检测IO事件和任务队列
获取最新要执行的定时任务的deadline作为这次select的deadline
hasTasks()为穿插任务，当出现穿插任务时，跳出这次select
selector.select(timeoutMillis) 进行阻塞式select
当检测到未实际阻塞并且超过selectCnt阈值512
触发空轮训bug -> rebuildSelector();
newSelector = openSelector();
Register all channels to the new Selector.
处理IO事件
在默认优化的前提下，
SelectorImpl的selectedKeys和publicSelectedKeys已被反射替换为new SelectedSelectionKeySet()，
该实现通过数组的方式优化了HashSet
SelectorImpl的selectedKeys和publicSelectedKeys已被反射替换为new SelectedSelectionKeySet()，
该实现通过数组的方式优化了HashSet
processSelectedKeysOptimized(selectedKeys.flip())
processSelectedKey(SelectionKey k, AbstractNioChannel ch)
任务执行
循环获取scheduledTask.deadlineNanos() <= nanoTime的定时任务
即将需要执行的定时任务合并至taskQueue
即将需要执行的定时任务合并至taskQueue
在deadline(ioRatio默认1:1)之内时循环执行taskQueue中的task
2 * cpu
AbstractBootstrap.doBind0() -> channel.eventLoop().execute -> channel.bind
当execute调用时，当分配到的EventLoop尚未启动时触发线程的启动执行
在绑定结束后，会触发pipeline的channelActive事件
AbstractBootstrap.doBind0() -> channel.eventLoop().execute -> channel.bind
当execute调用时，当分配到的EventLoop尚未启动时触发线程的启动执行
在绑定结束后，会触发pipeline的channelActive事件
实际阻塞的select操作未发生阻塞并超过阈值则重新构建selector
当外部线程调用eventLoop或channel方法时
将外部线程执行的任务封装成task丢至EventLoop顺序执行
将外部线程执行的任务封装成task丢至EventLoop顺序执行
新连接接入
新连接检测
processSelectedKey(SelectionKey k, AbstractNioChannel ch)
NioServerSocketChannel.doReadMessages()
调用accept方法获取jdk SocketChannel
调用accept方法获取jdk SocketChannel
buf.add(new NioSocketChannel(this, ch)) 包装成netty channel
pipeline.fireChannelRead(readBuf.get(i))
allocHandle用来控制accept速率
allocHandle用来控制accept速率
NioSocketChannel的创建
this.readInterestOp = SelectionKey.OP_READ 保存关心的事件为read
ch.configureBlocking(false);
javaSocket.setTcpNoDelay(true);
Netty中Channel的分类
Channel 层级关系
ChannelConfig 层级关系
新连接分配NioEventLoop并注册Selector
服务端channel初始化时触发Acceptor的添加
ServerBootstrap.init(Channel channel) -> pipeline.addLast(new ServerBootstrapAcceptor())
ServerBootstrap.init(Channel channel) -> pipeline.addLast(new ServerBootstrapAcceptor())
检测到新连接时触发NioMessageUnsafe.read() -> pipeline.fireChannelRead(readBuf.get(i));
触发ServerBootstrapAcceptor.channelRead(ChannelHandlerContext ctx, Object msg)
Acceptor的channelRead逻辑
child.pipeline().addLast(childHandler);
对新连接设置options和attrs
childGroup.register(channel)
新连接注册至workerGroup
新连接注册至workerGroup
next().register(channel)
通过chooser选择一个NioEventLoop进行注册
通过chooser选择一个NioEventLoop进行注册
AbstractUnsafe.register(EventLoop eventLoop, final ChannelPromise promise)
doRegister() -> selectionKey = javaChannel().register(eventLoop().selector, 0, this)
此时注册至selector但并不关心事件
doRegister() -> selectionKey = javaChannel().register(eventLoop().selector, 0, this)
此时注册至selector但并不关心事件
总结：在boss中的NioEventLoop检测到新连接，注册至worker中的NioEventLoop
NioSocketChannel读事件注册
入口：pipeline.fireChannelActive();
HeadContext.channelActive(ChannelHandlerContext ctx)
HeadContext.readIfIsAutoRead()
HeadContext.unsafe.beginRead();
AbstractNioChannel.doBeginRead()
selectionKey.interestOps(this.readInterestOp)
传播结束回到头结点，此时实际设置感兴趣的事件
selectionKey.interestOps(this.readInterestOp)
传播结束回到头结点，此时实际设置感兴趣的事件
select检测到新连接
processSelectedKey 处理jdk-客户端channel
此时创建Netty-channel保存感兴趣的事件
此时创建Netty-channel保存感兴趣的事件
触发服务端读事件至ServerBootstrapAcceptor.channelRead
调用childGroup.register(child)，通过choose分配EventLoop
Netty-Channel保存分配的EventLoop并将Channel保存的jdk-Channel注册至selector
调用childGroup.register(child)，通过choose分配EventLoop
Netty-Channel保存分配的EventLoop并将Channel保存的jdk-Channel注册至selector
注册完毕触发ChannelActive事件
由HeadContext触发beginRead，此时将开始监听实际感兴趣的事件
由HeadContext触发beginRead，此时将开始监听实际感兴趣的事件
eventloop -> processSelectedKeys();
NioMessageUnsafe.read()
-> doReadMessages(readBuf)
-> javaChannel().accept()
-> new NioSocketChannel(this, ch)
-> pipeline.fireChannelRead(readBuf.get(i));
-> ServerBootstrapAcceptor.channelRead()
-> childGroup.register(channel)
-> next().register(channel)
-> doReadMessages(readBuf)
-> javaChannel().accept()
-> new NioSocketChannel(this, ch)
-> pipeline.fireChannelRead(readBuf.get(i));
-> ServerBootstrapAcceptor.channelRead()
-> childGroup.register(channel)
-> next().register(channel)
注册的两个部分
register() 保存选择出来的eventLoop
register0() jdk-channel实际注册，未监听任何事件
beginRead() 监听感兴趣的事件
ChannelPipeline
pipeline初始化
new AbstractChannel(Channel parent)
-> pipeline = newChannelPipeline()
-> new DefaultChannelPipeline(Channel channel)
-> pipeline = newChannelPipeline()
-> new DefaultChannelPipeline(Channel channel)
this.channel = ObjectUtil.checkNotNull(channel, "channel");
tail = new TailContext(this);
head = new HeadContext(this);
head.next = tail;
tail.prev = head;
tail = new TailContext(this);
head = new HeadContext(this);
head.next = tail;
tail.prev = head;
pipeline默认结构
添加ChannelHandler
ChannelPipeline.addLast(ChannelHandler... handlers)
判断是否重复添加
checkMultiplicity(handler);
checkMultiplicity(handler);
创建节点
newCtx = newContext(group, filterName(name, handler), handler);
newCtx = newContext(group, filterName(name, handler), handler);
添加至链表
addLast0(newCtx);
addLast0(newCtx);
回调添加完成事件
callHandlerAdded0(newCtx);
->
ctx.handler().handlerAdded(ctx);
ctx.setAddComplete();
callHandlerAdded0(newCtx);
->
ctx.handler().handlerAdded(ctx);
ctx.setAddComplete();
删除ChannelHandler
ChannelHandler分类
inBound事件的传播
顺序传播
outBound事件的传播
逆序传播
异常的传播
当前节点顺序传播
问题
根据class类型
根据传播规律添加
ctx当前节点
pipeline头尾传播
pipeline头尾传播
ByteBuf
ByteBuf结构及重要API
结构
read，write改变指针，set，get不改变指针
mark用于记录当前指针，reset用来修改指针为mark所指的位置
ByteBuf分类
分类-1
分类-2
池化与非池化由子类实现
safe和unsafe自动检测
heap与direct由api提供
ByteBufAllocator内存分配器
UnpooledByteBufAllocator
heap通过创建数组[]分配内存
direct通过调用jdk-nio创建直接内存ByteBuffer
PooledByteBufAllocator
结构
PoolArena
- DirectArena
- HeapArena
Thread通过PoolThreadLocalCache创建的PoolThreadCache与某个Arean绑定
PooledByteBufAllocator每次创建时同时创建2种Arean
- DirectArena
- HeapArena
Thread通过PoolThreadLocalCache创建的PoolThreadCache与某个Arean绑定
PooledByteBufAllocator每次创建时同时创建2种Arean
内存规格
MemoryRegionCache
每个节点为该种规格的RegionCache，通过内部的queue来存储这种规格的内存
PoolThreadCache结构
Thread - PoolThreadCache - Allocator 三者关系
Arenas默认大小为2 * cpu核心数
Arena用于开辟一块连续内存
Arena用于开辟一块连续内存
PoolThreadCache - MemoryRegionCache 关系 memCache用于缓存一块连续内存
PoolArena结构
Chunk结构以及Page切分
缓存的分配流程
PooledByteBufAllocator.newDirectBuffer(int, int)
-> directArena.allocate
-> newByteBuf(maxCapacity); 从Recycler中获取一个纯净对象
-> allocate(cache, buf, reqCapacity); 给这个纯净对象分配内存
-> directArena.allocate
-> newByteBuf(maxCapacity); 从Recycler中获取一个纯净对象
-> allocate(cache, buf, reqCapacity); 给这个纯净对象分配内存
PoolThreadCache.allocateNormal(PoolArena<?>, PooledByteBuf<?>, int, int)
首先在cache上进行内存分配
首先在cache上进行内存分配
PoolArena.allocateNormal(PooledByteBuf<T>, int, int)
cache无法内存分配时，由arena分配
cache无法内存分配时，由arena分配
命中缓存的分配流程
计算缓存节点，tiny通过除以16得出节点下标
拿到MemoryRegionCache
拿到MemoryRegionCache
MemoryRegionCache.queue.poll(); 弹出一个Entry
initBuf(); 将弹出的entry所代表的内存分配给ByteBuf
-> buf.init() 完成初始化
initBuf(); 将弹出的entry所代表的内存分配给ByteBuf
-> buf.init() 完成初始化
entry.recycle(); 将弹出的entry丢回对象池(默认只回收1/8)
未命中缓存的分配流程
page级别内存分配
PoolArena.allocateNormal(PooledByteBuf<T>, int, int)
PoolArena.allocateNormal(PooledByteBuf<T>, int, int)
PoolChunkList.allocate(PooledByteBuf<T>, int, int)
第一次，此时List内部为空，即还没有chunk
第一次，此时List内部为空，即还没有chunk
Chunk通过一个平衡二叉树来保存内存分配情况
PoolChunk.allocateRun(int)
-> int d = maxOrder - (log2(normCapacity) - pageShifts); 计算需要在第几层分配
-> int id = allocateNode(d); id即表示树中的第几个节点，作为handle返回
-> updateParentsAlloc(id); 标记父节点内存被使用
PoolChunk.initBuf(PooledByteBuf<T>, long, int)
-> PooledByteBuf.init(PoolChunk<T>, long, int, int, int, PoolThreadCache)
分配完毕，保存chunk和handle即可指向一块内存
-> int d = maxOrder - (log2(normCapacity) - pageShifts); 计算需要在第几层分配
-> int id = allocateNode(d); id即表示树中的第几个节点，作为handle返回
-> updateParentsAlloc(id); 标记父节点内存被使用
PoolChunk.initBuf(PooledByteBuf<T>, long, int)
-> PooledByteBuf.init(PoolChunk<T>, long, int, int, int, PoolThreadCache)
分配完毕，保存chunk和handle即可指向一块内存
subpage级别内存分配
ByteBuf的回收
拿到MemoryRegionCache节点，添加至队列
当缓存队列满后加入失败，则标记分配到的连续内存为未使用
通过recycle()回收至Recycler
总结
heap/direct safe/unsafe pooled/unpooled
Allocator持有Arena数组，Arena用于分配内存
通过PoolThreadCache将线程与Arena绑定，默认一个Nio线程持管理一个Arena
通过PoolThreadCache将线程与Arena绑定，默认一个Nio线程持管理一个Arena
huge - 直接分配
normal - page
small/tiny - subpage
normal - page
small/tiny - subpage
Netty解码
ByteToMessageDecoder
通过cumulation累加字节
decodeRemovalReentryProtection(ctx, in, out);
调用子类的docode方法进行解析
调用子类的docode方法进行解析
未解析数据则跳出循环
解析到数据至out时则循环传播解析到的list后clear
解析到数据至out时则循环传播解析到的list后clear
FixedLengthFrameDecoder
直到可读字节数达到一帧则读取ByteBuf至out
LineBasedFrameDecoder
以\r\n或\n作为分隔符读取一帧
当发现已经超出所设的最大长度时，则丢弃下一个分隔符前的所有字节
DelimiterBasedFrameDecoder
构建时当发现分割符为LineBase则初始化LineBasedFrameDecoder
逻辑同LineBasedFrameDecoder
当有多个分隔符时，每次取最小的帧，即以最近的分隔符为截止点
当有多个分隔符时，每次取最小的帧，即以最近的分隔符为截止点
LengthFieldBasedFrameDecoder
frameLength += lengthAdjustment + lengthFieldEndOffset;
当frameLength大于最大帧限制时，直到将该帧字节全部丢弃完毕才会退出丢弃模式，逻辑同其余解码器
当frameLength大于最大帧限制时，直到将该帧字节全部丢弃完毕才会退出丢弃模式，逻辑同其余解码器
总结
通过一定规则累积ByteBuf，当满足一帧时向后传播
如上
Netty编码
writeAndFlush()
通过pipeline调用时从tail节点传播，否则从当前节点传播，见pipeline传播机制
acceptOutboundMessage(msg)
I cast = (I) msg;
匹配对象
I cast = (I) msg;
匹配对象
buf = allocateBuffer(ctx, cast, preferDirect);
分配内存
分配内存
encode(ctx, cast, buf);
编码实现，由子类实现
编码实现，由子类实现
ReferenceCountUtil.release(cast);
释放对象
释放对象
ctx.write(buf, promise);
传播数据
传播数据
buf.release();
释放内存
释放内存
HeadContext.write(ctx, msg, promise)
-> unsafe.write(msg, promise);
-> unsafe.write(msg, promise);
msg = filterOutboundMessage(msg);
检测msg类型是否支持，将buf变为directBuf
检测msg类型是否支持，将buf变为directBuf
outboundBuffer.addMessage(msg, size, promise);
添加至buffer
setUnwritable(invokeLater);
-> fireChannelWritabilityChanged(invokeLater);
当大于64 * 1024时，设置不可写状态
-> fireChannelWritabilityChanged(invokeLater);
当大于64 * 1024时，设置不可写状态
outboundBuffer.addFlush();
当总pending字节小于低水位时则设置为可写状态
当总pending字节小于低水位时则设置为可写状态
状态
AbstractNioByteChannel.doWrite(ChannelOutboundBuffer in)
in.current()
拿到flushedEntry的msg
拿到flushedEntry的msg
ByteBuf buf = (ByteBuf) msg;
in.remove();
当jdk底层无法写入时，之后可能的某个状态
问题
通过编码规则写入ByteBuf，通过ctx或pipeline传递至HeadContext节点
Netty性能优化工具类解析
FastThreadLocal
每次创建都有唯一ID
index = InternalThreadLocalMap.nextVariableIndex();
每次调用构造函数都分配唯一的index
每次调用构造函数都分配唯一的index
get()实现
slowGet() - fastGet()
slowGet() -> ThreadLocal<InternalThreadLocalMap>
通过jdk线程变量存储该Map
通过jdk线程变量存储该Map
fastGet((FastThreadLocalThread) thread);
直接拿到FastThread内部成员变量Map
直接拿到FastThread内部成员变量Map
每个Thread维护一个数组
Object[] array = new Object[32];
Arrays.fill(array, UNSET);
默认大小32
Arrays.fill(array, UNSET);
默认大小32
每个Thread持有一个InternalThreadLocalMap，为一个数组
每个FastThreadLocal持有一个index，
即可在该Thread内的数组中获取该线程变量
即可在该Thread内的数组中获取该线程变量
不同线程含有不同数组，即ThreadLocal在不同线程之间是隔离的
index为0是variablesToRemoveIndex，故实际有效下标从1开始
Object v = threadLocalMap.indexedVariable(index);
直接根据索引号从该线程所持有的数组中获取value
直接根据索引号从该线程所持有的数组中获取value
当获取的值为null时调用 -> initialValue();
随后将该值设入该线程所持有的数组中
随后将该值设入该线程所持有的数组中
addToVariablesToRemove(threadLocalMap, this);
set()实现
获取map
设置值后调用addToVariablesToRemove
remove时将该index位置设置为UNSET
当remove调的值不是UNSET时调用onRemoval(v)
removeFromVariablesToRemove同时将0位置的set中的该FastThreadLocal引用移除
当remove调的值不是UNSET时调用onRemoval(v)
removeFromVariablesToRemove同时将0位置的set中的该FastThreadLocal引用移除
Recycler
创建
FastThreadLocal<Stack<T>>
每个线程持有一个Stack
每个线程持有一个Stack
maxCapacity = 32k
ratioMask = 7 即只回收1/8的对象
maxDelayedQueues = 2 * cpu
avaliable = 32 / 2 = 16k
ratioMask = 7 即只回收1/8的对象
maxDelayedQueues = 2 * cpu
avaliable = 32 / 2 = 16k
recycler.get()
获取线程变量Stack
stack.pop()
从Stack弹出一个handle
从Stack弹出一个handle
stack.pop()
scavengeSome()
boolean WeakOrderQueue.transfer(Stack<?> dst)
每次transfer转移一个Link块内的数据
每次transfer转移一个Link块内的数据
当handle为空时，创建一个handle
并调用newObject()创建一个对象与handle绑定
并调用newObject()创建一个对象与handle绑定
回收对象
handle.recycle(this);
stack.push(this);
handle.recycle(this);
stack.push(this);
同线程回收对象
pushNow(item);
pushNow(item);
默认情况只回收1/8的未被回收过的对象
直接放入stack的【DefaultHandle<?>[] elements】中
结构
每一个link包含一个handles，默认大小为16
每次分配一个link，即批量分配可回收的handle空槽
每次分配一个link，即批量分配可回收的handle空槽
绑定关系
每次创建WeakOrderQueue都插入head的头部
原始Stack就可以通过单向链表获得外部线程回收的对象
原始Stack就可以通过单向链表获得外部线程回收的对象
异线程回收对象
pushLater(item, currentThread);
pushLater(item, currentThread);
FastThreadLocal<Map<Stack<?>, WeakOrderQueue>>
获取线程变量中存储的WeakOrderQueue
获取线程变量中存储的WeakOrderQueue
queue = WeakOrderQueue.allocate(this, thread))
获取为空时创建一个queue
获取为空时创建一个queue
当Link满时申请空间后创建一个link
tail.elements[writeIndex] = handle; 在link中存储该handle
handle.stack = null; handle存储于Link，此时已不属于原始Stack
tail.elements[writeIndex] = handle; 在link中存储该handle
handle.stack = null; handle存储于Link，此时已不属于原始Stack
总结
Netty设计模式应用
单例模式
ReadTimeoutException
MqttEncoder
策略模式
DefaultEventExecutorChooserFactory.newChooser(EventExecutor[])
PowerOfTowEventExecutorChooser
GenericEventExecutorChooser
装饰器模式
WrappedByteBuf及其子类
观察者模式
ChannelFuture为被观察者
addListener添加监听器即观察者
addListener添加监听器即观察者
writeAndFlush()
Promise为被观察者
Future为观察者
Promise为被观察者
Future为观察者
迭代器模式
Bytebuf.foreach
责任链模式
Pipeline
责任处理器接口
ChannelHandler
ChannelHandler
责任链
ChannelPipeline
ChannelPipeline
上下文
ChannelHandlerContext
通过ctx next/prev构成双向链表
ChannelHandlerContext
通过ctx next/prev构成双向链表
责任终止机制
netty - fire
other - return false
netty - fire
other - return false
总结
优化
单机调优
应用调优
耗时任务需要单独的线程池
0 条评论
下一页
