Netty核心之Channel-netty学习笔记(5)-20210804

//操作一：打开 FileChannel通道
RandomAccessFile aFile     = new RandomAccessFile("test.txt"，"rw");
FileChannel      inChannel =aFile.getChannel();

//操作二：读取数据
ByteBuffer buf = ByteBuffer.allocate(48);
int bytesRead =inChannel.read(buf);

//操作三：写入数据
String newData = "New String to write to file..." +System.currentTimeMillis();
ByteBuffer buf = ByteBuffer.allocate(48);
buf.clear();
buf.put(newData.getBytes());
buf.flip();
while(buf.hasRemaining())
{
    channel.write(buf);
}

//操作四：关闭
channel.close();

//当我们对 FileChannel 的操作完成后，必须将其关闭。

//操作五：强制刷新磁盘
channel.force(true);

//操作一：创建
SocketChannel socketChannel =SocketChannel.open();
socketChannel.connect(new InetSocketAddress("127.0.0.1"，80));

//这个是客户端的创建。当一个服务器端的ServerSocketChannel 接受到连接请求时，也会返回一个 SocketChannel 对象。

//操作二：读取
ByteBuffer buf = ByteBuffer.allocate(48);
int bytesRead =socketChannel.read(buf);
//如果 read()返回 -1，那么表示连接中断了.

//操作三：写入数据
String newData = "New String to write to file..." +System.currentTimeMillis();
ByteBuffer buf = ByteBuffer.allocate(48);
buf.clear();
buf.put(newData.getBytes());
buf.flip();
while(buf.hasRemaining()) {
channel.write(buf);
}

//操作四：关闭
socketChannel.close();
//在非阻塞模式，我们可以设置 SocketChannel 为异步模式，这样我们的 connect，read，write 都是异步的了.
//操作一：连接
socketChannel.configureBlocking(false);
socketChannel.connect(new InetSocketAddress("127.0.0.1"，80));
while(!socketChannel.finishConnect() ){
    //wait，or do something else...
}
//在异步模式中，或许连接还没有建立，socketChannel.connect 方法就返回了，因此我们不断的自旋，检查当前是否是连接到了主机。

//操作二：非阻塞读写
//在异步模式下，读写的方式是一样的.
//在读取时，因为是异步的，因此我们必须检查 read 的返回值，来判断当前是否读取到了数据.
ServerSocketChannel
ServerSocketChannel 顾名思义，是用在服务器为端的，可以监听客户端的 TCP 连接，例如:
ServerSocketChannel serverSocketChannel =ServerSocketChannel.open();
serverSocketChannel.socket().bind(new InetSocketAddress(9999));
while(true){
SocketChannel socketChannel =serverSocketChannel.accept();
    //do something with socketChannel...
}

//操作四：关闭
ServerSocketChannel serverSocketChannel =ServerSocketChannel.open();
serverSocketChannel.close();

while(true){

SocketChannel socketChannel =serverSocketChannel.accept();

//do something with socketChannel...
}

ServerSocketChannel serverSocketChannel =ServerSocketChannel.open();

serverSocketChannel.socket().bind(new InetSocketAddress(9999));

serverSocketChannel.configureBlocking(false);

while(true){

    SocketChannel socketChannel =serverSocketChannel.accept();

    if(socketChannel != null){

        //do something with socketChannel...
}

}

操作一：打开

DatagramChannel channel =DatagramChannel.open();

channel.socket().bind(new InetSocketAddress(9999));

操作二：读取数据

ByteBuffer buf = ByteBuffer.allocate(48);

buf.clear();

channel.receive(buf);
操作三：发送数据

String newData = "New String to write to file..."

                    +System.currentTimeMillis();

ByteBuffer buf = ByteBuffer.allocate(48);

buf.clear();

buf.put(newData.getBytes());

buf.flip();

int bytesSent = channel.send(buf，new InetSocketAddress("example.com"，80));

因为 UDP 是非连接的，因此这个的 connect 并不是向 TCP 一样真正意义上的连接，因此我们仅仅可以从指定的地址中读取或写入数据.

channel.connect(new InetSocketAddress("example.com"，80));

interfaceUnsafe {
//把channel注册进EventLoop
voidregister(EventLoop eventLoop, ChannelPromise promise);
 
 //todo 给channel绑定一个 adress,
voidbind(SocketAddress localAddress, ChannelPromise promise);

//把channel注册进Selector
voidderegister(ChannelPromise promise);

//从channel中读取IO数据
voidbeginRead();

//往channe写入数据
void write(Object msg, ChannelPromise promise);

protectedAbstractChannel(Channel parent) {
    this.parent =parent;
    //todo channelId 代表Chanel唯一的身份标志
    id =newId();
    //todo 创建一个unsafe对象
    unsafe =newUnsafe();
    //todo 在这里初始化了每一个channel都会有的pipeline组件
    pipeline =newChannelPipeline();
}

protected abstract class AbstractUnsafe implementsUnsafe {

@Override
//todo 入参 eventLoop == SingleThreadEventLoop   promise == NioServerSocketChannel + Executor
public final void register(EventLoop eventLoop, finalChannelPromise promise) {
    if (eventLoop == null) {
        throw new NullPointerException("eventLoop");
    }
    if(isRegistered()) {
        promise.setFailure(new IllegalStateException("registered to an event loop already"));
        return;
    }
    if (!isCompatible(eventLoop)) {
        promise.setFailure(
                new IllegalStateException("incompatible event loop type: " +eventLoop.getClass().getName()));
        return;
    }
    //todo 赋值给自己的 事件循环, 把当前的eventLoop赋值给当前的Channel上  作用是标记后续的所有注册的操作都得交给我这个eventLoop处理, 正好对应着下面的判断
    //todo 保证了 即便是在多线程的环境下一条channel 也只能注册关联上唯一的eventLoop,唯一的线程
    AbstractChannel.this.eventLoop =eventLoop;

    //todo 下面的分支判断里面执行的代码是一样的!!, 为什么? 这是netty的重点, 它大量的使用线程, 线程之间就会产生同步和并发的问题
    //todo 下面的分支,目的就是把线程可能带来的问题降到最低限度
    //todo 进入inEventLoop() --> 判断当前执行这行代码的线程是否就是 SingleThreadEventExecutor里面维护的那条唯一的线程
    //todo 解释下面分支的必要性, 一个eventLoop可以注册多个channel, 但是channel的整个生命周期中所有的IO事件,仅仅和它关联上的thread有关系
    //todo 而且,一个eventLoop在他的整个生命周期中,只和唯一的线程进行绑定,
    //
    //todo 当我们注册channel的时候就得确保给他专属它的thread,
    //todo 如果是新的连接到了,
    if(eventLoop.inEventLoop()) {
        //todo 进入regist0()
register0(promise);
    } else{
        try{
            //todo 如果不是,它以一个任务的形式提交  事件循环 , 新的任务在新的线程开始,  规避了多线程的并发
            //todo 他是SimpleThreadEventExucutor中execute()实现的,把任务添加到执行队列执行
            eventLoop.execute(newRunnable() {
                @Override
                public voidrun() {
                    register0(promise);
                }
            });
        } catch(Throwable t) {
            logger.warn(
                    "Force-closing a channel whose registration task was not accepted by an event loop: {}",
                    AbstractChannel.this, t);
            closeForcibly();
            closeFuture.setClosed();
            safeSetFailure(promise, t);
        }
    }
}

private voidregister0(ChannelPromise promise) {
    try{
        //check if the channel is still open as it could be closed in the mean time when the register
        //call was outside of the eventLoop
        if (!promise.setUncancellable() || !ensureOpen(promise)) {
            return;
        }
        boolean firstRegistration =neverRegistered;
        //todo 进入这个方法doRegister()
        //todo 它把系统创建的ServerSocketChannel 注册进了选择器
doRegister();
        neverRegistered = false;
        registered = true;

        //Ensure we call handlerAdded(...) before we actually notify the promise. This is needed as the
        //user may already fire events through the pipeline in the ChannelFutureListener.
        //todo 确保在 notify the promise前调用 handlerAdded(...)
        //todo 这是必需的，因为用户可能已经通过ChannelFutureListener中的管道触发了事件。
        //todo 如果需要的话,执行HandlerAdded()方法
        //todo 正是这个方法, 回调了前面我们添加 Initializer 中添加 Accpter的重要方法
pipeline.invokeHandlerAddedIfNeeded();

        //todo  !!!!!!!  观察者模式!!!!!!  通知观察者,谁是观察者?  暂时理解ChannelHandler 是观察者
safeSetSuccess(promise);

        //todo 传播行为, 传播什么行为呢?   在head---> ServerBootStraptAccptor ---> tail传播事件ChannelRegistered  , 也就是挨个调用它们的ChannelRegisted函数
pipeline.fireChannelRegistered();
        //Only fire a channelActive if the channel has never been registered. This prevents firing
        //multiple channel actives if the channel is deregistered and re-registered.
        //todo 对于服务端:  javaChannel().socket().isBound(); 即  当Channel绑定上了端口   isActive()才会返回true
        //todo 对于客户端的连接 ch.isOpen() && ch.isConnected(); 返回true , 就是说, Channel是open的 打开状态的就是true
        if(isActive()) {
            if(firstRegistration) {
                //todo 在pipeline中传播ChannelActive的行为,跟进去
pipeline.fireChannelActive();
            } else if(config().isAutoRead()) {
                //This channel was registered before and autoRead() is set. This means we need to begin read
                //again so that we process inbound data.
                //
                //See https://github.com/netty/netty/issues/4805
                //todo 可以接受客户端的数据了
beginRead();
            }
        }
    } catch(Throwable t) {
        //Close the channel directly to avoid FD leak.
closeForcibly();
        closeFuture.setClosed();
        safeSetFailure(promise, t);
    }
}

@Override
public final void bind(final SocketAddress localAddress, finalChannelPromise promise) {
    assertEventLoop();

    if (!promise.setUncancellable() || !ensureOpen(promise)) {
        return;
    }

    //See: https://github.com/netty/netty/issues/576
    if (Boolean.TRUE.equals(config().getOption(ChannelOption.SO_BROADCAST)) &&localAddress instanceof InetSocketAddress &&
        !((InetSocketAddress) localAddress).getAddress().isAnyLocalAddress() &&
        !PlatformDependent.isWindows() && !PlatformDependent.maybeSuperUser()) {
        //Warn a user about the fact that a non-root user can't receive a
        //broadcast packet on *nix if the socket is bound on non-wildcard address.
logger.warn(
                "A non-root user can't receive a broadcast packet if the socket " +
                "is not bound to a wildcard address; binding to a non-wildcard " +
                "address (" + localAddress + ") anyway as requested.");
    }

    boolean wasActive =isActive();
    //todo 由于端口的绑定未完成，所以 wasActive是 false

    try{
        //todo 绑定端口, 进去就是NIO原生JDK绑定端口的代码
doBind(localAddress);
        //todo 端口绑定完成  isActive（）是true
    } catch(Throwable t) {
        safeSetFailure(promise, t);
        closeIfClosed();
        return;
    }
    //todo 根据上面的逻辑判断， 结果为 true
    if (!wasActive &&isActive()) {
        invokeLater(newRunnable() {
            //todo 来到这里很重要， 向下传递事件行为， 传播行为的时候， 从管道的第一个节点开始传播， 第一个节点被封装成 HeadContext的对象
           //todo 进入方法， 去 HeadContext里面查看做了哪些事情
            //todo 她会触发channel的read, 最终重新为 已经注册进selector 的 chanel, 二次注册添加上感性趣的accept事件
@Override
            public voidrun() {
                pipeline.fireChannelActive();
            }
        });
    }

    //todo 观察者模式, 设置改变状态, 通知观察者的方法回调
    safeSetSuccess(promise);

*/ //todo 无论是服务端的channel 还是客户端的channel都会使用这个方法进行初始化
// //TODO: 2019/6/23                null        ServerSocketChannel       accept
//todo  如果是在创建NioSocketChannel  parent==NioServerSocketChannel  ch == SocketChanel
protected AbstractNioChannel(Channel parent, SelectableChannel ch, intreadInterestOp) {
    super(parent);//todo  继续向上跟,创建基本的组件
    //todo 如果是创建NioSocketChannel   这就是在保存原生的jdkchannel
    //todo 如果是创建NioServerSocketChannel   这就是在保存ServerSocketChannel
    this.ch =ch;
    //todo 设置上感兴趣的事件
    this.readInterestOp =readInterestOp;
    try{
        //todo 作为服务端, ServerSocketChannel 设置为非阻塞的
        //todo 作为客户端   SocketChannel 设置为非阻塞的
        ch.configureBlocking(false);
    } catch (IOException e) {

private ChannelFuture doBind(finalSocketAddress localAddress) {
    //初始化及注册
    final ChannelFuture regFuture =initAndRegister();
    final Channel channel =regFuture.channel();
    if (regFuture.cause() != null) {
        returnregFuture;
    }

    if(regFuture.isDone()) {
        //At this point we know that the registration was complete and successful.
        ChannelPromise promise =channel.newPromise();
        //调用 doBind0
doBind0(regFuture, channel, localAddress, promise);
        returnpromise;
    } else{
        ....
    }
}

private static voiddoBind0(
        final ChannelFuture regFuture, finalChannel channel,
        final SocketAddress localAddress, finalChannelPromise promise) {

    channel.eventLoop().execute(newRunnable() {
        @Override
        public voidrun() {
            if(regFuture.isSuccess()) {
                //调用channel.bind接口
channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
            } else{
                promise.setFailure(regFuture.cause());
            }
        }
    });
}

@Override
protected void doRegister() throwsException {
boolean selected = false;
for(;;) {
try{
    //todo  javaChannel() -- 返回SelectableChanel 可选择的Channel,换句话说,可以和Selector搭配使用,他是channel体系的顶级抽象类, 实际的类型是 ServerSocketChannel
    //todo  eventLoop().unwrappedSelector(), -- >  获取选择器, 现在在AbstractNioChannel中 获取到的eventLoop是BossGroup里面的
    //todo  到目前看, 他是把ServerSocketChannel(系统创建的) 注册进了 EventLoop的选择器
    //todo 这里的 最后一个参数是  this是当前的channel , 意思是把当前的Channel当成是一个 attachment(附件) 绑定到selector上 作用???
    //todo  现在知道了attachment的作用了
     //todo 1. 当channel在这里注册进 selector中返回一个selectionKey, 这个key告诉selector 这个channel是自己的
     //todo 2. 当selector轮询到 有channel出现了自己的感兴趣的事件时, 需要从成百上千的channel精确的匹配出 出现Io事件的channel,
    //todo     于是seleor就在这里提前把channel存放入 attachment中, 后来使用
    //todo 最后一个 this 参数, 如果是服务启动时, 他就是NioServerSocketChannel   如果是客户端他就是 NioSocketChannel
    //todo 到目前为止, 虽然注册上了,但是它不关心任何事件
    selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
    return;
} catch (CancelledKeyException e) {

/*** Read from underlying {@linkSelectableChannel}
 */
//todo 两个实现类, NioByteUnsafe , 处理关于客户端发来的信息
//todo NioMessageUnsafe   处理客户端新进来的连接
voidread();


/*** Special {@linkUnsafe} sub-type which allows to access the underlying {@linkSelectableChannel}
*/
public interface NioUnsafe extendsUnsafe {
/*** Return underlying {@linkSelectableChannel}
 */SelectableChannel ch();

/*** Finish connect
 */
voidfinishConnect();


voidforceFlush();
}

protected AbstractNioMessageChannel(Channel parent, SelectableChannel ch, intreadInterestOp) {
    //todo 在进去
    //todo  null  ServerSocketChannel   accept
    super(parent, ch, readInterestOp);
}

//todo 用于处理新链接进来的内部类
private final class NioMessageUnsafe extendsAbstractNioUnsafe {

//todo 这个容器用于存放临时读到的连接
private final List<Object> readBuf = new ArrayList<Object>();

//todo 接受新链接的 read来到这里
@Override
public voidread() {
    ...
    doBeginRead(buf);
    ...
}

//todo 处理新的连接 是在 NioServerSocketChannel中实现的, 进入查看
protected abstract int doReadMessages(List<Object> buf) throws Exception;

@Override
protected int doReadMessages(List<Object> buf) throwsException {
    //todo java Nio底层在这里 创建jdk底层的 原生channel
    SocketChannel ch =SocketUtils.accept(javaChannel());

    try{
        if (ch != null) {
            //todo  把java原生的channel, 封装成 Netty自定义的封装的channel , 这里的buf是list集合对象,由上一层传递过来的
            //todo  this  --  NioServerSocketChannel
            //todo  ch --     SocketChnnel
            buf.add(new NioSocketChannel(this, ch));
            return 1;
        }

@Override
protected int doReadBytes(ByteBuf byteBuf) throwsException {
    final RecvByteBufAllocator.Handle allocHandle =unsafe().recvBufAllocHandle();
    allocHandle.attemptedBytesRead(byteBuf.writableBytes());
    returnbyteBuf.writeBytes(javaChannel(), allocHandle.attemptedBytesRead());
}