Netty源码之写入数据

Posted 2023-04-27

参考技术A 写入数据，是从缓冲区写入到通道中。读取数据和写入数据是相对概念，从通道读取到缓冲区，从缓冲区写入到通道。
Pipeline结构

出站：TailContext---->StringEncoder------>HeadContext
写入到通道有两种方式：
1、channelHandlerContext.channel().writeAndFlush("");
通过NiosocketChannel调用writeAndFlush()方法，这个从pipeline的tail尾节点往前执行过滤
出站：TailContext---->StringEncoder------>HeadContext
2、channelHandlerContext.writeAndFlush("");
通过ChannelHandlerContext类执行，从当前节点往前执行
出站：StringEncoder------>HeadContext

ServerHandler类是自定义的入站类，在调用channelRead0方法读取数据后，经服务端转发到非当前客户端。实现简单聊天室功能。

1、writeAndFlush
io.netty.channel.AbstractChannel#writeAndFlush(java.lang.Object)

3、writeAndFlush
io.netty.channel.AbstractChannelHandlerContext#writeAndFlush(java.lang.Object)

④、最终启动线程后，会调用StringEncoder.invokeWrite(m, promise);方法

1、invokeWriteAndFlush
io.netty.channel.AbstractChannelHandlerContext#invokeWriteAndFlush

2、write
调用到Head节点的invokeWrite方法

4、invokeWrite0
io.netty.channel.AbstractChannelHandlerContext#invokeWrite0

io.netty.channel.ChannelOutboundBuffer#addMessage

1、invokeWriteAndFlush
io.netty.channel.AbstractChannelHandlerContext#invokeWriteAndFlush

2、flush
io.netty.channel.ChannelOutboundHandlerAdapter#flush

4、invokeFlush
io.netty.channel.AbstractChannelHandlerContext#invokeFlush

②、addFlush方法把没有刷新的unflushedEntry指针数据添加到刷新的flushedEntry指针

Netty的写入数据是把ByteBuf写到到Channel通道的过程。
ByteBuf数据缓冲区经历了Channel通道所有Pipeline的出站实现类，这个过程是开启了NioEventLoop的事件线程，用来执行出站任务。
写到通道分两个步骤：首先，调用write方法把所有缓冲区数据封装成Entry加入到ChannelOutboundBuffer。
然后调用刷新方法写入到Channel通道

netty源码之读取数据

目录

前话

1、读数据的技巧

2、简约流程

3、读数据的本质

流程

1、读取数据

---

前话

1、读数据的技巧

 

 

2、简约流程

 

3、读数据的本质

 

流程

读数据的轮询和接收连接是一样的 ，都是在NioEventLoop对象的run方法里，但是在最终读的时候，调用了另外一个对象，AbstractNioByteChannel.read 方法

1、读取数据

//AbstractNioByteChannel 
public final void read() 
    final ChannelConfig config = config();
    final ChannelPipeline pipeline = pipeline();
    final ByteBufAllocator allocator = config.getAllocator();
    final RecvByteBufAllocator.Handle allocHandle = recvBufAllocHandle();
    allocHandle.reset(config);
​
    ByteBuf byteBuf = null;
    boolean close = false;
    try 
        // 这是一个循环的读取数据
        do 
            byteBuf = allocHandle.allocate(allocator);
            allocHandle.lastBytesRead(doReadBytes(byteBuf));
            if (allocHandle.lastBytesRead() <= 0) 
                // nothing was read. release the buffer.
                byteBuf.release();
                byteBuf = null;
                close = allocHandle.lastBytesRead() < 0;
                break;
            
​
            allocHandle.incMessagesRead(1);
            readPending = false;
            //处理每次收到的数据
            pipeline.fireChannelRead(byteBuf);
            byteBuf = null;
         while (allocHandle.continueReading());
​
        allocHandle.readComplete();
        //本次事件处理完毕
        pipeline.fireChannelReadComplete();
​
        if (close) 
            closeOnRead(pipeline);
        
     catch (Throwable t) 
        handleReadException(pipeline, byteBuf, t, close, allocHandle);
     finally 
        // Check if there is a readPending which was not processed yet.
        // This could be for two reasons:
        // * The user called Channel.read() or ChannelHandlerContext.read() in channelRead(...) method
        // * The user called Channel.read() or ChannelHandlerContext.read() in channelReadComplete(...) method
        //
        // See https://github.com/netty/netty/issues/2254
        if (!readPending && !config.isAutoRead()) 
            removeReadOp();
        
    

​
 protected int doReadBytes(ByteBuf byteBuf) throws Exception 
        final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
        allocHandle.attemptedBytesRead(byteBuf.writableBytes());
        return byteBuf.writeBytes(javaChannel(), allocHandle.attemptedBytesRead());