Netty源码分析思维导图模板_ProcessOn思维导图、流程图

服务端启动

创建ServerSocketChannel

new NioServerSocketChannel()

newSocket(PROVIDER) -> ServerSocketChannel

this.readInterestOp = SelectionKey.OP_ACCEPT 保存关心的事件为Accept

初始化ServerSocketChannel

添加之后的结构

注册selector

AbstractChannel.this.eventLoop = eventLoop;
将分配到的EventLoop保存至chanel内部

将Netty-Channel内部的jdk-Channel注册至该EventLoop绑定的selector上，此时并不关心事件
selectionKey = javaChannel().register(eventLoop().selector, 0, this);

获取注册后的selectionKey作为Netty-Channel的成员变量

服务端口绑定

jdk-Channel绑定端口
javaChannel().bind(localAddress, config.getBacklog());

绑定完毕，触发channelActive事件
HeadContext ctx.fireChannelActive();

触发完毕，调用HeadContext.readIfIsAutoRead();

传播至HeadContext.read()
-> unsafe.beginRead();
-> doBeginRead() 修改SelectionKey感兴趣的事件为创建时保存的兴趣事件

创建时保存感兴趣事件，创建jdk-Channel并创建id, unsafe, pipeline

服务端初始化主要用于在pipeline添加ServerBootstrapAcceptor处理器
用于将监听到的客户端Channel注册至客户端Group中

注册主要目的用于将
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)

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

processSelectedKeysOptimized(selectedKeys.flip())

processSelectedKey(SelectionKey k, AbstractNioChannel ch)

任务执行

循环获取scheduledTask.deadlineNanos() <= nanoTime的定时任务
即将需要执行的定时任务合并至taskQueue

在deadline(ioRatio默认1:1)之内时循环执行taskQueue中的task

2 * cpu
AbstractBootstrap.doBind0() -> channel.eventLoop().execute -> channel.bind
当execute调用时，当分配到的EventLoop尚未启动时触发线程的启动执行
在绑定结束后，会触发pipeline的channelActive事件

实际阻塞的select操作未发生阻塞并超过阈值则重新构建selector

当外部线程调用eventLoop或channel方法时
将外部线程执行的任务封装成task丢至EventLoop顺序执行

新连接接入

新连接检测

processSelectedKey(SelectionKey k, AbstractNioChannel ch)

NioServerSocketChannel.doReadMessages()
调用accept方法获取jdk SocketChannel

buf.add(new NioSocketChannel(this, ch)) 包装成netty channel

pipeline.fireChannelRead(readBuf.get(i))
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())

检测到新连接时触发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

next().register(channel)
通过chooser选择一个NioEventLoop进行注册

AbstractUnsafe.register(EventLoop eventLoop, final ChannelPromise promise)
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)
传播结束回到头结点，此时实际设置感兴趣的事件

select检测到新连接

processSelectedKey 处理jdk-客户端channel
此时创建Netty-channel保存感兴趣的事件

触发服务端读事件至ServerBootstrapAcceptor.channelRead
调用childGroup.register(child)，通过choose分配EventLoop
Netty-Channel保存分配的EventLoop并将Channel保存的jdk-Channel注册至selector

注册完毕触发ChannelActive事件
由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)

注册的两个部分

register() 保存选择出来的eventLoop

register0() jdk-channel实际注册，未监听任何事件

beginRead() 监听感兴趣的事件

ChannelPipeline

pipeline初始化

new AbstractChannel(Channel parent)
-> 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;

pipeline默认结构

添加ChannelHandler

ChannelPipeline.addLast(ChannelHandler... handlers)

判断是否重复添加
checkMultiplicity(handler);

创建节点
newCtx = newContext(group, filterName(name, handler), handler);

添加至链表
addLast0(newCtx);

回调添加完成事件
callHandlerAdded0(newCtx);
->
ctx.handler().handlerAdded(ctx);
ctx.setAddComplete();

删除ChannelHandler

ChannelHandler分类

inBound事件的传播

顺序传播

outBound事件的传播

逆序传播

异常的传播

当前节点顺序传播

问题

根据class类型

根据传播规律添加

ctx当前节点
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

内存规格

MemoryRegionCache

每个节点为该种规格的RegionCache，通过内部的queue来存储这种规格的内存

PoolThreadCache结构

Thread - PoolThreadCache - Allocator 三者关系

Arenas默认大小为2 * cpu核心数
Arena用于开辟一块连续内存

PoolThreadCache - MemoryRegionCache 关系 memCache用于缓存一块连续内存

PoolArena结构

Chunk结构以及Page切分

缓存的分配流程

PooledByteBufAllocator.newDirectBuffer(int, int)
-> directArena.allocate
-> newByteBuf(maxCapacity); 从Recycler中获取一个纯净对象
-> allocate(cache, buf, reqCapacity); 给这个纯净对象分配内存

PoolThreadCache.allocateNormal(PoolArena<?>, PooledByteBuf<?>, int, int)
首先在cache上进行内存分配

PoolArena.allocateNormal(PooledByteBuf<T>, int, int)
cache无法内存分配时，由arena分配

命中缓存的分配流程

计算缓存节点，tiny通过除以16得出节点下标
拿到MemoryRegionCache

MemoryRegionCache.queue.poll(); 弹出一个Entry
initBuf(); 将弹出的entry所代表的内存分配给ByteBuf
-> buf.init() 完成初始化

entry.recycle(); 将弹出的entry丢回对象池(默认只回收1/8)

未命中缓存的分配流程

page级别内存分配
PoolArena.allocateNormal(PooledByteBuf<T>, int, int)

PoolChunkList.allocate(PooledByteBuf<T>, int, int)
第一次，此时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即可指向一块内存

subpage级别内存分配

ByteBuf的回收

拿到MemoryRegionCache节点，添加至队列

当缓存队列满后加入失败，则标记分配到的连续内存为未使用

通过recycle()回收至Recycler

总结

heap/direct safe/unsafe pooled/unpooled

Allocator持有Arena数组，Arena用于分配内存
通过PoolThreadCache将线程与Arena绑定，默认一个Nio线程持管理一个Arena

huge - 直接分配
normal - page
small/tiny - subpage

Netty解码

ByteToMessageDecoder

通过cumulation累加字节

decodeRemovalReentryProtection(ctx, in, out);
调用子类的docode方法进行解析

未解析数据则跳出循环
解析到数据至out时则循环传播解析到的list后clear

FixedLengthFrameDecoder

直到可读字节数达到一帧则读取ByteBuf至out

LineBasedFrameDecoder

以\r\n或\n作为分隔符读取一帧

当发现已经超出所设的最大长度时，则丢弃下一个分隔符前的所有字节

DelimiterBasedFrameDecoder

构建时当发现分割符为LineBase则初始化LineBasedFrameDecoder

逻辑同LineBasedFrameDecoder
当有多个分隔符时，每次取最小的帧，即以最近的分隔符为截止点

LengthFieldBasedFrameDecoder

frameLength += lengthAdjustment + lengthFieldEndOffset;
当frameLength大于最大帧限制时，直到将该帧字节全部丢弃完毕才会退出丢弃模式，逻辑同其余解码器

总结

通过一定规则累积ByteBuf，当满足一帧时向后传播

如上

Netty编码

writeAndFlush()

通过pipeline调用时从tail节点传播，否则从当前节点传播，见pipeline传播机制

acceptOutboundMessage(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);

msg = filterOutboundMessage(msg);
检测msg类型是否支持，将buf变为directBuf

outboundBuffer.addMessage(msg, size, promise);

添加至buffer

setUnwritable(invokeLater);
-> fireChannelWritabilityChanged(invokeLater);
当大于64 * 1024时，设置不可写状态

outboundBuffer.addFlush();
当总pending字节小于低水位时则设置为可写状态

状态

AbstractNioByteChannel.doWrite(ChannelOutboundBuffer in)

in.current()
拿到flushedEntry的msg

ByteBuf buf = (ByteBuf) msg;

in.remove();

当jdk底层无法写入时，之后可能的某个状态

问题

通过编码规则写入ByteBuf，通过ctx或pipeline传递至HeadContext节点

Netty性能优化工具类解析

FastThreadLocal

每次创建都有唯一ID

index = InternalThreadLocalMap.nextVariableIndex();
每次调用构造函数都分配唯一的index

get()实现

slowGet() - fastGet()

slowGet() -> ThreadLocal<InternalThreadLocalMap>
通过jdk线程变量存储该Map

fastGet((FastThreadLocalThread) thread);
直接拿到FastThread内部成员变量Map

每个Thread维护一个数组

Object[] array = new Object[32];
Arrays.fill(array, UNSET);
默认大小32

每个Thread持有一个InternalThreadLocalMap，为一个数组

每个FastThreadLocal持有一个index，
即可在该Thread内的数组中获取该线程变量

不同线程含有不同数组，即ThreadLocal在不同线程之间是隔离的

index为0是variablesToRemoveIndex，故实际有效下标从1开始

Object v = threadLocalMap.indexedVariable(index);
直接根据索引号从该线程所持有的数组中获取value

当获取的值为null时调用 -> initialValue();
随后将该值设入该线程所持有的数组中

addToVariablesToRemove(threadLocalMap, this);

set()实现

获取map

设置值后调用addToVariablesToRemove

remove时将该index位置设置为UNSET
当remove调的值不是UNSET时调用onRemoval(v)
removeFromVariablesToRemove同时将0位置的set中的该FastThreadLocal引用移除

Recycler

创建

FastThreadLocal<Stack<T>>
每个线程持有一个Stack

maxCapacity = 32k
ratioMask = 7 即只回收1/8的对象
maxDelayedQueues = 2 * cpu
avaliable = 32 / 2 = 16k

recycler.get()

获取线程变量Stack

stack.pop()
从Stack弹出一个handle

stack.pop()

scavengeSome()

boolean WeakOrderQueue.transfer(Stack<?> dst)
每次transfer转移一个Link块内的数据

当handle为空时，创建一个handle
并调用newObject()创建一个对象与handle绑定

回收对象
handle.recycle(this);
stack.push(this);

同线程回收对象
pushNow(item);

默认情况只回收1/8的未被回收过的对象

直接放入stack的【DefaultHandle<?>[] elements】中

结构

每一个link包含一个handles，默认大小为16
每次分配一个link，即批量分配可回收的handle空槽

绑定关系

每次创建WeakOrderQueue都插入head的头部
原始Stack就可以通过单向链表获得外部线程回收的对象

异线程回收对象
pushLater(item, currentThread);

FastThreadLocal<Map<Stack<?>, WeakOrderQueue>>
获取线程变量中存储的WeakOrderQueue

queue = WeakOrderQueue.allocate(this, thread))
获取为空时创建一个queue

当Link满时申请空间后创建一个link
tail.elements[writeIndex] = handle; 在link中存储该handle
handle.stack = null; handle存储于Link，此时已不属于原始Stack

总结

Netty设计模式应用

单例模式

ReadTimeoutException

MqttEncoder

策略模式

DefaultEventExecutorChooserFactory.newChooser(EventExecutor[])

PowerOfTowEventExecutorChooser

GenericEventExecutorChooser

装饰器模式

WrappedByteBuf及其子类

观察者模式

ChannelFuture为被观察者
addListener添加监听器即观察者

writeAndFlush()
Promise为被观察者
Future为观察者

迭代器模式

Bytebuf.foreach

责任链模式

Pipeline

责任处理器接口
ChannelHandler

责任链
ChannelPipeline

上下文
ChannelHandlerContext
通过ctx next/prev构成双向链表

责任终止机制
netty - fire
other - return false

总结

优化

单机调优

应用调优

耗时任务需要单独的线程池