`
Donald_Draper
  • 浏览: 978832 次
社区版块
存档分类
最新评论

netty 抽象nio通道解析

阅读更多
netty 通道接口定义:http://donald-draper.iteye.com/blog/2392740
netty 抽象通道初始化:http://donald-draper.iteye.com/blog/2392801
netty 抽象Unsafe定义:http://donald-draper.iteye.com/blog/2393053
netty 通道Outbound缓冲区:http://donald-draper.iteye.com/blog/2393098
netty 抽象通道后续:http://donald-draper.iteye.com/blog/2393166
引言:
上一篇文章我们看了抽象通道触发OutboundInvoker相关事件方法,先来回顾一下:
    通道的绑定操作、连接,写消息,读操作,刷新操作,反注册、断开连接,关闭通道等操作事件实际调用通道的Channel管道的相关方法,即触发通道相关事件,这些方法是重写了通道OutboundInvoker的相关方法。在抽象Unsafe那篇文章中,我们看到其内部也有绑定、注册,读操作,写操作和关闭操作,这些是通道的实际操作方法。
今天我们来看抽象nio通道AbstractNioChannel
package io.netty.channel.nio;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.ByteBufUtil;
import io.netty.buffer.Unpooled;
import io.netty.channel.AbstractChannel;
import io.netty.channel.Channel;
import io.netty.channel.ChannelException;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelPromise;
import io.netty.channel.ConnectTimeoutException;
import io.netty.channel.EventLoop;
import io.netty.util.ReferenceCountUtil;
import io.netty.util.ReferenceCounted;
import io.netty.util.internal.ThrowableUtil;
import io.netty.util.internal.logging.InternalLogger;
import io.netty.util.internal.logging.InternalLoggerFactory;

import java.io.IOException;
import java.net.SocketAddress;
import java.nio.channels.CancelledKeyException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.ConnectionPendingException;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

/**
 * Abstract base class for {@link Channel} implementations which use a Selector based approach.
 */
public abstract class AbstractNioChannel extends AbstractChannel {

    private static final InternalLogger logger =
            InternalLoggerFactory.getInstance(AbstractNioChannel.class);
    //关闭通道异常
    private static final ClosedChannelException DO_CLOSE_CLOSED_CHANNEL_EXCEPTION = ThrowableUtil.unknownStackTrace(
            new ClosedChannelException(), AbstractNioChannel.class, "doClose()");

    private final SelectableChannel ch;//关联通道
    protected final int readInterestOp;//读操作事件
    volatile SelectionKey selectionKey;//关联选择key
    boolean readPending;//是否开始读操作
    private final Runnable clearReadPendingRunnable = new Runnable() {
        @Override
        public void run() {
            clearReadPending0();
        }
    };

    /**
     * The future of the current connection attempt.  If not null, subsequent
     * connection attempts will fail.
     */
    private ChannelPromise connectPromise;//异步可写的连接任务
    private ScheduledFuture<?> connectTimeoutFuture;
    private SocketAddress requestedRemoteAddress;//远端socket地址
}

从上面可以看出,抽象nio通道内部关联一个可选择通道(SelectableChannel)和一个选择key(selectionKey)


来看构造:
/**
 * Create a new instance
 *
 * @param parent            the parent {@link Channel} by which this instance was created. May be {@code null}
 所属通道
 * @param ch                the underlying {@link SelectableChannel} on which it operates
 底层选择通道
 * @param readInterestOp    the ops to set to receive data from the {@link SelectableChannel}
 读操作事件
 */
protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
    super(parent);
    this.ch = ch;
    this.readInterestOp = readInterestOp;
    try {
        //初始化通道,并配置为非阻塞模式
        ch.configureBlocking(false);
    } catch (IOException e) {
        try {
            ch.close();
        } catch (IOException e2) {
            if (logger.isWarnEnabled()) {
                logger.warn(
                        "Failed to close a partially initialized socket.", e2);
            }
        }

        throw new ChannelException("Failed to enter non-blocking mode.", e);
    }
}

从上面可以看出抽象Nio通道构造,主要是初始化通道并配置为非阻塞模式。
下面几个方法,很简单,看一下就行
//通道是否打开
@Override
public boolean isOpen() {
    return ch.isOpen();
}
//获取底层选择通道
protected SelectableChannel javaChannel() {
    return ch;
}
//获取通道所在的事件循环
@Override
public NioEventLoop eventLoop() {
    return (NioEventLoop) super.eventLoop();
}


来看抽象通道几个实际操作方法do*,
先从注册方法开始
 @Override
protected void doRegister() throws Exception {
    boolean selected = false;
    for (;;) {
        try {
	   //注册可选择通道到通道所在事件循环的选择器中
            selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
            return;
        } catch (CancelledKeyException e) {
            if (!selected) {
                // Force the Selector to select now as the "canceled" SelectionKey may still be
                // cached and not removed because no Select.select(..) operation was called yet.
		//如果注册异常,并且没有选择,则执行选择操作,将选择key从选择器的取消key集合中移除
                eventLoop().selectNow();
                selected = true;
            } else {
                // We forced a select operation on the selector before but the SelectionKey is still cached
                // for whatever reason. JDK bug ?
                throw e;
            }
        }
    }
}

从上面来看注册工作主要是,注册可选择通道到通道所在事件循环的选择器中。

//再来看反注册
@Override
protected void doDeregister() throws Exception {
    //委托给事件循环,取消选择key,即从事件循环关联选择器的选择key集合中移除当前选择key
    eventLoop().cancel(selectionKey());
}


//读操作
@Override
protected void doBeginRead() throws Exception {
    // Channel.read() or ChannelHandlerContext.read() was called
    final SelectionKey selectionKey = this.selectionKey;
    if (!selectionKey.isValid()) {
        //选择key无效,直接返回
        return;
    }

    readPending = true;
    final int interestOps = selectionKey.interestOps();
    if ((interestOps & readInterestOp) == 0) {
        //将读操作事件,添加选择key的兴趣事件集
        selectionKey.interestOps(interestOps | readInterestOp);
    }
}

从上面可以看出,开始读操作,实际工作为将读操作事件,添加选择key的兴趣事件集

再来看实际关闭任务操作:
@Override
protected void doClose() throws Exception {
    ChannelPromise promise = connectPromise;
    if (promise != null) {
        // Use tryFailure() instead of setFailure() to avoid the race against cancel().
        promise.tryFailure(DO_CLOSE_CLOSED_CHANNEL_EXCEPTION);
        connectPromise = null;
    }
    //获取连接超时任务
    ScheduledFuture<?> future = connectTimeoutFuture;
    if (future != null) {
        //取消任务
        future.cancel(false);
        connectTimeoutFuture = null;
    }
}

来看其他方法:
//获取通道Unsafe
@Override
public NioUnsafe unsafe() {
    return (NioUnsafe) super.unsafe();
}
/**
 * Return the current {@link SelectionKey}
 获取通道选择key
 */
protected SelectionKey selectionKey() {
    assert selectionKey != null;
    return selectionKey;
}

/**
 * @deprecated No longer supported.
 * No longer supported.
 是否正在进行读操作
 */
@Deprecated
protected boolean isReadPending() {
    return readPending;
}

/**
 * @deprecated Use {@link #clearReadPending()} if appropriate instead.
 * No longer supported.
 设置正在读标志
 */
@Deprecated
protected void setReadPending(final boolean readPending) {
    if (isRegistered()) {
        EventLoop eventLoop = eventLoop();
        if (eventLoop.inEventLoop()) {
	    //在当前事件循环,完成实际设置读标志
            setReadPending0(readPending);
        } else {
            eventLoop.execute(new Runnable() {
                @Override
                public void run() {
                    setReadPending0(readPending);
                }
            });
        }
    } else {
        // Best effort if we are not registered yet clear readPending.
        // NB: We only set the boolean field instead of calling clearReadPending0(), because the SelectionKey is
        // not set yet so it would produce an assertion failure.
        this.readPending = readPending;
    }
}
//完成实际设置读标志
private void setReadPending0(boolean readPending) {
    this.readPending = readPending;
    if (!readPending) {
         //如果为不在读数据,委托给Unsafe,从选择key兴趣集中移除读操作事件
        ((AbstractNioUnsafe) unsafe()).removeReadOp();
    }
}


/**
 * Set read pending to {@code false}.
 清除读操作标志,即设置标志为false
 */
protected final void clearReadPending() {
    if (isRegistered()) {
        EventLoop eventLoop = eventLoop();
        if (eventLoop.inEventLoop()) {
	   //在当前事件循环,完成实际清除
            clearReadPending0();
        } else {
            eventLoop.execute(clearReadPendingRunnable);
        }
    } else {
        // Best effort if we are not registered yet clear readPending. This happens during channel initialization.
        // NB: We only set the boolean field instead of calling clearReadPending0(), because the SelectionKey is
        // not set yet so it would produce an assertion failure.
        readPending = false;
    }
}
//完成实际清除
private void clearReadPending0() {
    readPending = false;
    //委托给Unsafe,从选择key兴趣集中移除读操作事件
    ((AbstractNioUnsafe) unsafe()).removeReadOp();
}


从下面一个方法:
//获取通道Unsafe
@Override
public NioUnsafe unsafe() {
    return (NioUnsafe) super.unsafe();
}

可以看出抽象nio通道的内部Unsafe为NioUnsafe

/**
 * Special {@link Unsafe} sub-type which allows to access the underlying {@link SelectableChannel}
 特殊的Unsafe,允许访问底层的选择通道
 */
public interface NioUnsafe extends Unsafe {
    /**
     * Return underlying {@link SelectableChannel}
     返回底层选择通道
     */
    SelectableChannel ch();

    /**
     * Finish connect
     完成连接
     */
    void finishConnect();

    /**
     * Read from underlying {@link SelectableChannel}
     从底层选择操作,读取数据
     */
    void read();
    //强制刷新
    void forceFlush();
}


来看NioUnsafe的抽象实现AbstractNioUnsafe,为抽象nio通道的内部类,
 protected abstract class AbstractNioUnsafe extends AbstractUnsafe implements NioUnsafe {
        //移除读兴趣事件
        protected final void removeReadOp() {
	    //获取选择key
            SelectionKey key = selectionKey();
            // Check first if the key is still valid as it may be canceled as part of the deregistration
            // from the EventLoop
            // See https://github.com/netty/netty/issues/2104
            if (!key.isValid()) {
	        //选择key无效,直接返回
                return;
            }
            int interestOps = key.interestOps();
            if ((interestOps & readInterestOp) != 0) {
                // only remove readInterestOp if needed
		//从选择key兴趣事件集中,移除读操作事件
                key.interestOps(interestOps & ~readInterestOp);
            }
        }
        //获取选择通道,实际返回的抽象nio通道内部的底层可选择通道
        @Override
        public final SelectableChannel ch() {
            return javaChannel();
        }
}

从上面可以看出,抽象nioUnsafe为特殊的Unsafe,允许访问底层的选择通道。
选择通道方法返回的实际为抽象nio通道内部的底层可选择通道。
移除读兴趣事件removeReadOp,即从选择key兴趣事件集中,移除读操作事件。

来看抽象NioUnsafe的其他方法
//连接远端Socket地址,如果需要绑定本地socket地址,连接完成通知异步可写任务promise
@Override
public final void connect(
        final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) {
    if (!promise.setUncancellable() || !ensureOpen(promise)) {
        //确保任务没有取消,通道打开
        return;
    }

    try {
        if (connectPromise != null) {
	    //已连接
            // Already a connect in process.
            throw new ConnectionPendingException();
        }

        boolean wasActive = isActive();
        if (doConnect(remoteAddress, localAddress)) {
	   //如果连接成功,则更新任务结果,如果需要,则触发通道的激活事件fireChannelActive
            fulfillConnectPromise(promise, wasActive);
        } else {
	    //连接失败,则添加异步连接任务
            connectPromise = promise;
            requestedRemoteAddress = remoteAddress;

            // Schedule connect timeout.连接超时配置
            int connectTimeoutMillis = config().getConnectTimeoutMillis();
	    //更新连接异步任务结果为连接超时,并将任务交个事件循环区调度
            if (connectTimeoutMillis > 0) {
                connectTimeoutFuture = eventLoop().schedule(new Runnable() {
                    @Override
                    public void run() {
                        ChannelPromise connectPromise = AbstractNioChannel.this.connectPromise;
                        ConnectTimeoutException cause =
                                new ConnectTimeoutException("connection timed out: " + remoteAddress);
                        if (connectPromise != null && connectPromise.tryFailure(cause)) {
                            close(voidPromise());
                        }
                    }
                }, connectTimeoutMillis, TimeUnit.MILLISECONDS);
            }
            //添加任务监听器
            promise.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    if (future.isCancelled()) {
                        if (connectTimeoutFuture != null) {
                            connectTimeoutFuture.cancel(false);
                        }
                        connectPromise = null;
			//连接任务取消,则关闭任务
                        close(voidPromise());
                    }
                }
            });
        }
    } catch (Throwable t) {
       //异常,则更新任务失败,如果需要则关闭通道
        promise.tryFailure(annotateConnectException(t, remoteAddress));
        closeIfClosed();
    }
}

连接方法,我们需要关注为以下片段:

if (doConnect(remoteAddress, localAddress)) {
   //如果连接成功,则更新任务结果,如果需要,则触发通道的激活事件fireChannelActive
    fulfillConnectPromise(promise, wasActive);
} 


//AbstractNioChannel
/**
 * Connect to the remote peer,待子类实现
 */
protected abstract boolean doConnect(SocketAddress remoteAddress, SocketAddress localAddress) throws Exception;



//更新任务结果,触发通道的激活事件fireChannelActive
private void fulfillConnectPromise(ChannelPromise promise, boolean wasActive) {
    if (promise == null) {
        // Closed via cancellation and the promise has been notified already.
	通道已经关闭,直接返回
        return;
    }

    // Get the state as trySuccess() may trigger an ChannelFutureListener that will close the Channel.
    // We still need to ensure we call fireChannelActive() in this case.
    boolean active = isActive();

    // trySuccess() will return false if a user cancelled the connection attempt.
    //更新连接任务成功完成
    boolean promiseSet = promise.trySuccess();

    // Regardless if the connection attempt was cancelled, channelActive() event should be triggered,
    // because what happened is what happened.
    if (!wasActive && active) {
        //触发通道的激活事件fireChannelActive
        pipeline().fireChannelActive();
    }
    // If a user cancelled the connection attempt, close the channel, which is followed by channelInactive().
    if (!promiseSet) {
        //关闭任务
        close(voidPromise());
    }
}

从上面可以看出,连接操作,将实际连接操作委托给doConnect,待子类实现,如果连接成功
,则通知异步任务连接成功,如果是第一次连接,则触发通道的激活事件fireChannelActive。

再来看完成连接方法
@Override
public final void finishConnect() {
    // Note this method is invoked by the event loop only if the connection attempt was
    // neither cancelled nor timed out.

    assert eventLoop().inEventLoop();

    try {
        boolean wasActive = isActive();
	//实际完成连接
        doFinishConnect();
	//更新任务结果,触发通道的激活事件fireChannelActive
        fulfillConnectPromise(connectPromise, wasActive);
    } catch (Throwable t) {
         //这个包装异常的方法annotateConnectException,在上一篇已看过,即将远端socket地址,添加的异常信息中
        fulfillConnectPromise(connectPromise, annotateConnectException(t, requestedRemoteAddress));
    } finally {
        // Check for null as the connectTimeoutFuture is only created if a connectTimeoutMillis > 0 is used
        // See https://github.com/netty/netty/issues/1770
        if (connectTimeoutFuture != null) {
            connectTimeoutFuture.cancel(false);
        }
        connectPromise = null;
    }
}

//AbstractNioChannel
/**
 * Finish the connect,待子类实现
 */
protected abstract void doFinishConnect() throws Exception;



//更新连接任务为异常失败
private void fulfillConnectPromise(ChannelPromise promise, Throwable cause) {
    if (promise == null) {
        // Closed via cancellation and the promise has been notified already.
        return;
    }
    // Use tryFailure() instead of setFailure() to avoid the race against cancel().
    promise.tryFailure(cause);
    closeIfClosed();
}

从上面可以看出,完成连接操作,实际工作委托给抽象Nio通道的doFinishConnect方法,待子类实现,完成后更新任务结果,触发通道的激活事件fireChannelActive,如果出现异常,则更新连接任务为异常失败。

再来看刷新操作
@Override
protected final void flush0() {
    // Flush immediately only when there's no pending flush.
    // If there's a pending flush operation, event loop will call forceFlush() later,
    // and thus there's no need to call it now.
    if (isFlushPending()) {
        return;
    }
    //委托给父类
    super.flush0();
}


@Override
public final void forceFlush() {
    // directly call super.flush0() to force a flush now
     //委托给父类
    super.flush0();
}

private boolean isFlushPending() {
    SelectionKey selectionKey = selectionKey();
    //写操作事件,存在选择key兴趣事件集中
    return selectionKey.isValid() && (selectionKey.interestOps() & SelectionKey.OP_WRITE) != 0;
}


回到抽象nio通道,再来看抽象nio通道的其他方法:
//判断事件循环为通道兼容,即判断事件循环是否为Nio事件循环
@Override
protected boolean isCompatible(EventLoop loop) {
    return loop instanceof NioEventLoop;
}


/**
 * Returns an off-heap copy of the specified {@link ByteBuf}, and releases the original one.
 * Note that this method does not create an off-heap copy if the allocation / deallocation cost is too high,
 * but just returns the original {@link ByteBuf}..
 包装原始buf为direct buf,成功则释放原始buf,如果保证成本较高,则返回原始buf
 */
protected final ByteBuf newDirectBuffer(ByteBuf buf) {
    //获取buf可读字节数
    final int readableBytes = buf.readableBytes();
    if (readableBytes == 0) {
        //释放buf
        ReferenceCountUtil.safeRelease(buf);
        return Unpooled.EMPTY_BUFFER;
    }
    //获取字节buf分配器
    final ByteBufAllocator alloc = alloc();
    if (alloc.isDirectBufferPooled()) {
        //如果分配器为Direct池Buffer类型,则分配字节Direct类型字节buf
        ByteBuf directBuf = alloc.directBuffer(readableBytes);
	//将原始buf中的数据,写到新的Direct buf中
        directBuf.writeBytes(buf, buf.readerIndex(), readableBytes);
	//释放原始buf
        ReferenceCountUtil.safeRelease(buf);
        return directBuf;
    }
    //否则,获取线程本地的direct buf
    final ByteBuf directBuf = ByteBufUtil.threadLocalDirectBuffer();
    if (directBuf != null) {
       //将原始buf中的数据,写到新的Direct buf中
        directBuf.writeBytes(buf, buf.readerIndex(), readableBytes);
        ReferenceCountUtil.safeRelease(buf);
        return directBuf;
    }
    // Allocating and deallocating an unpooled direct buffer is very expensive; give up.
   //如果分配和回收一个非池类的Direct buf代价比较高,则直接返回原始buf。
    return buf;
}


/**
 * Returns an off-heap copy of the specified {@link ByteBuf}, and releases the specified holder.
 * The caller must ensure that the holder releases the original {@link ByteBuf} when the holder is released by
 * this method.  Note that this method does not create an off-heap copy if the allocation / deallocation cost is
 * too high, but just returns the original {@link ByteBuf}..
 此方法与上面方法不同的是,释放的是buf的Holder,主要是保证原始buf能够释放
 */
protected final ByteBuf newDirectBuffer(ReferenceCounted holder, ByteBuf buf) {
    final int readableBytes = buf.readableBytes();
    if (readableBytes == 0) {
        //释放的是buf的Holder
        ReferenceCountUtil.safeRelease(holder);
        return Unpooled.EMPTY_BUFFER;
    }

    final ByteBufAllocator alloc = alloc();
    if (alloc.isDirectBufferPooled()) {
        ByteBuf directBuf = alloc.directBuffer(readableBytes);
        directBuf.writeBytes(buf, buf.readerIndex(), readableBytes);
	//释放的是buf的Holder
        ReferenceCountUtil.safeRelease(holder);
        return directBuf;
    }

    final ByteBuf directBuf = ByteBufUtil.threadLocalDirectBuffer();
    if (directBuf != null) {
        directBuf.writeBytes(buf, buf.readerIndex(), readableBytes);
	//释放的是buf的Holder
        ReferenceCountUtil.safeRelease(holder);
        return directBuf;
    }

    // Allocating and deallocating an unpooled direct buffer is very expensive; give up.
    if (holder != buf) {
        // Ensure to call holder.release() to give the holder a chance to release other resources than its content.
        buf.retain();//buf引用计数器自增1
        ReferenceCountUtil.safeRelease(holder);
    }
    return buf;
}


我们来看这句:
//否则,获取线程本地的direct buf
final ByteBuf directBuf = ByteBufUtil.threadLocalDirectBuffer();


//ByteBufUtil
**
 * Returns a cached thread-local direct buffer, if available.
 *
 * @return a cached thread-local direct buffer, if available.  {@code null} otherwise.
 */
public static ByteBuf threadLocalDirectBuffer() {
    if (THREAD_LOCAL_BUFFER_SIZE <= 0) {
       //线程本地buffer size小于0,则直接返回
        return null;
    }
    if (PlatformDependent.hasUnsafe()) {
        return ThreadLocalUnsafeDirectByteBuf.newInstance();
    } else {
        return ThreadLocalDirectByteBuf.newInstance();
    }
}

//PlatformDependent
public final class PlatformDependent {
private static final boolean HAS_UNSAFE = hasUnsafe0();
 /**
     * Return {@code true} if {@code sun.misc.Unsafe} was found on the classpath and can be used for accelerated
     * direct memory access.
     */
    public static boolean hasUnsafe() {
        return HAS_UNSAFE;
    }
    private static boolean hasUnsafe0() {
        if (isAndroid()) {
            logger.debug("sun.misc.Unsafe: unavailable (Android)");
            return false;
        }
        if (PlatformDependent0.isExplicitNoUnsafe()) {
            return false;
        }
        try {
            boolean hasUnsafe = PlatformDependent0.hasUnsafe();
            logger.debug("sun.misc.Unsafe: {}", hasUnsafe ? "available" : "unavailable");
            return hasUnsafe;
        } catch (Throwable ignored) {
            // Probably failed to initialize PlatformDependent0.
            return false;
        }
    }
...
}

//PlatformDependent0
/**
 * The {@link PlatformDependent} operations which requires access to {@code sun.misc.*}.
 */
final class PlatformDependent0 {
    private static final InternalLogger logger = InternalLoggerFactory.getInstance(PlatformDependent0.class);
    private static final long ADDRESS_FIELD_OFFSET;
    private static final long BYTE_ARRAY_BASE_OFFSET;
    private static final Constructor<?> DIRECT_BUFFER_CONSTRUCTOR;
    private static final boolean IS_EXPLICIT_NO_UNSAFE = explicitNoUnsafe0();
    private static final Method ALLOCATE_ARRAY_METHOD;
    private static final int JAVA_VERSION = javaVersion0();
    private static final boolean IS_ANDROID = isAndroid0();
    private static final Object INTERNAL_UNSAFE;
    static final Unsafe UNSAFE;
    static boolean hasUnsafe() {
        return UNSAFE != null;
    }
    ...
}

下面两个字节部分都是ByteBufUtil的内部类
//ThreadLocalDirectByteBuf
static final class ThreadLocalDirectByteBuf extends UnpooledDirectByteBuf {
    
        private static final Recycler<ThreadLocalDirectByteBuf> RECYCLER = new Recycler<ThreadLocalDirectByteBuf>() {
            @Override
            protected ThreadLocalDirectByteBuf newObject(Handle<ThreadLocalDirectByteBuf> handle) {
                return new ThreadLocalDirectByteBuf(handle);
            }
        };
    
        static ThreadLocalDirectByteBuf newInstance() {
            ThreadLocalDirectByteBuf buf = RECYCLER.get();
            buf.setRefCnt(1);
            return buf;
        }
	...
}


//ThreadLocalUnsafeDirectByteBuf
static final class ThreadLocalUnsafeDirectByteBuf extends UnpooledUnsafeDirectByteBuf {
	private static final Recycler<ThreadLocalUnsafeDirectByteBuf> RECYCLER =
		new Recycler<ThreadLocalUnsafeDirectByteBuf>() {
		    @Override
		    protected ThreadLocalUnsafeDirectByteBuf newObject(Handle<ThreadLocalUnsafeDirectByteBuf> handle) {
			return new ThreadLocalUnsafeDirectByteBuf(handle);
		    }
		};

	static ThreadLocalUnsafeDirectByteBuf newInstance() {
	    ThreadLocalUnsafeDirectByteBuf buf = RECYCLER.get();
	    buf.setRefCnt(1);
	    return buf;
	}
	...
}

总结:

抽象nio通道AbstractNioChannel内部关联一个可选择通道(SelectableChannel)和一个选择key(selectionKey)。抽象Nio通道构造,主要是初始化通道并配置为非阻塞模式。

注册doRegister工作主要是,注册可选择通道到通道所在事件循环的选择器中。反注册doDeregister,委托给事件循环,取消选择key,即从事件循环关联选择器的选择key集合中移除当前选择key。开始读操作doBeginRead,实际工作为将读操作事件,添加选择key的兴趣事件集

抽象nioUnsafe为特殊的Unsafe,允许访问底层的选择通道。选择通道方法返回的实际为抽象nio通道内部的底层可选择通道。移除读兴趣事件removeReadOp,即从选择key兴趣事件集中,移除读操作事件。连接操作,将实际连接操作委托给doConnect,待子类实现,如果连接成功,则通知异步任务连接成功,如果是第一次连接,则触发通道的激活事件fireChannelActive。完成连接操作,实际工作委托给抽象Nio通道的doFinishConnect方法,待子类实现,完成后更新任务结果,触发通道的激活事件fireChannelActive,如果出现异常,则更新连接任务为异常失败。



附:
/**
 * A collection of utility methods that is related with handling {@link ByteBuf},
 * such as the generation of hex dump and swapping an integer's byte order.
 */
public final class ByteBufUtil {

    private static final InternalLogger logger = InternalLoggerFactory.getInstance(ByteBufUtil.class);
    private static final FastThreadLocal<CharBuffer> CHAR_BUFFERS = new FastThreadLocal<CharBuffer>() {
        @Override
        protected CharBuffer initialValue() throws Exception {
            return CharBuffer.allocate(1024);
        }
    };

    //ThreadLocalDirectByteBuf
    static final class ThreadLocalDirectByteBuf extends UnpooledDirectByteBuf {
    
        private static final Recycler<ThreadLocalDirectByteBuf> RECYCLER = new Recycler<ThreadLocalDirectByteBuf>() {
            @Override
            protected ThreadLocalDirectByteBuf newObject(Handle<ThreadLocalDirectByteBuf> handle) {
                return new ThreadLocalDirectByteBuf(handle);
            }
        };
    
        static ThreadLocalDirectByteBuf newInstance() {
            ThreadLocalDirectByteBuf buf = RECYCLER.get();
            buf.setRefCnt(1);
            return buf;
        }
    
        private final Handle<ThreadLocalDirectByteBuf> handle;
    
        private ThreadLocalDirectByteBuf(Handle<ThreadLocalDirectByteBuf> handle) {
            super(UnpooledByteBufAllocator.DEFAULT, 256, Integer.MAX_VALUE);
            this.handle = handle;
        }
    
        @Override
        protected void deallocate() {
            if (capacity() > THREAD_LOCAL_BUFFER_SIZE) {
                super.deallocate();
            } else {
                clear();
                handle.recycle(this);
            }
        }
    }
    //ThreadLocalUnsafeDirectByteBuf
    static final class ThreadLocalUnsafeDirectByteBuf extends UnpooledUnsafeDirectByteBuf {

        private static final Recycler<ThreadLocalUnsafeDirectByteBuf> RECYCLER =
                new Recycler<ThreadLocalUnsafeDirectByteBuf>() {
                    @Override
                    protected ThreadLocalUnsafeDirectByteBuf newObject(Handle<ThreadLocalUnsafeDirectByteBuf> handle) {
                        return new ThreadLocalUnsafeDirectByteBuf(handle);
                    }
                };

        static ThreadLocalUnsafeDirectByteBuf newInstance() {
            ThreadLocalUnsafeDirectByteBuf buf = RECYCLER.get();
            buf.setRefCnt(1);
            return buf;
        }

        private final Handle<ThreadLocalUnsafeDirectByteBuf> handle;

        private ThreadLocalUnsafeDirectByteBuf(Handle<ThreadLocalUnsafeDirectByteBuf> handle) {
            super(UnpooledByteBufAllocator.DEFAULT, 256, Integer.MAX_VALUE);
            this.handle = handle;
        }

        @Override
        protected void deallocate() {
            if (capacity() > THREAD_LOCAL_BUFFER_SIZE) {
                super.deallocate();
            } else {
                clear();
                handle.recycle(this);
            }
        }
    }
}
0
0
分享到:
评论

相关推荐

    aiohttp-3.7.3-cp36-cp36m-win_amd64.whl.rar

    python whl离线安装包 pip安装失败可以尝试使用whl离线安装包安装 第一步 下载whl文件,注意需要与python版本配套 python版本号、32位64位、arm或amd64均有区别 第二步 使用pip install XXXXX.whl 命令安装,如果whl路径不在cmd窗口当前目录下,需要带上路径 WHL文件是以Wheel格式保存的Python安装包, Wheel是Python发行版的标准内置包格式。 在本质上是一个压缩包,WHL文件中包含了Python安装的py文件和元数据,以及经过编译的pyd文件, 这样就使得它可以在不具备编译环境的条件下,安装适合自己python版本的库文件。 如果要查看WHL文件的内容,可以把.whl后缀名改成.zip,使用解压软件(如WinRAR、WinZIP)解压打开即可查看。 为什么会用到whl文件来安装python库文件呢? 在python的使用过程中,我们免不了要经常通过pip来安装自己所需要的包, 大部分的包基本都能正常安装,但是总会遇到有那么一些包因为各种各样的问题导致安装不了的。 这时我们就可以通过尝试去Python安装包大全中(whl包下载)下载whl包来安装解决问题。

    基于Java中的swing类的图形化飞机游戏的开发练习.zip

    基于Java中的Swing类开发的图形化飞机游戏练习包,为初学者和进阶学习者提供了实践Java GUI编程的绝佳机会。通过本资源,开发者可以利用Java语言和Swing库构建一个用户交互式的2D游戏,深入理解图形用户界面(GUI)编程和事件处理机制。该游戏的核心包括玩家飞机的控制、敌机的生成与移动、子弹发射与碰撞检测以及游戏胜负判定等逻辑。玩家通过鼠标移动控制己方飞机,实现平滑的移动和连续的子弹发射;而敌方飞机则按照一定算法无规律出现,随着游戏进程难度逐渐增加。游戏中还引入了特殊NPC,增加了额外的挑战和乐趣。为了提高游戏体验,游戏还包含了开始背景、结束背景以及背景音乐等元素。当玩家击毁敌机时,会有相应的得分计算和展示;若被敌机击中,则游戏结束并显示最终得分。此外,游戏还提供了查看历史前十记录、帮助和退出等选项,方便玩家进行游戏设置和了解游戏玩法。本资源适用于计算机科学与技术、软件工程、信息管理及相关专业的课程设计、毕业设计等环节,为学生提供实践操作的机会,帮助他们巩固Java编程知识,提高动手能力和发散思维。同时,也为希望学习不同技术领域的学习者提供了一个优秀的入门项目。

    SQLite:SQLite数据库创建与管理.docx

    SQLite:SQLite数据库创建与管理

    【完整源码+数据库】SpringBoot 集成 Spring Security短信验证码登录

    Spring Security 默认是账号和密码登录,现在是对 Spring Security 进行扩展,来实现短信验证码方式登录。 SpringBoot 集成 Spring Security短信验证码登录【完整源码+数据库】

    去年和朋友一起做的java小游戏.游戏具体界面在readme中,游戏设计的uml图在design.pdf中.zip

    本资源是一个Java小游戏项目,由我和我的朋友在去年共同开发。这个项目不仅包含了完整的游戏代码,还有详细的设计文档和UML图,适合作为学习和参考的素材。游戏的界面设计简洁明了,玩法有趣且富有挑战性,能够让玩家在游戏中体验到乐趣。在readme文件中,你可以找到游戏的具体界面展示,让你对游戏的外观有一个直观的了解。而design.pdf中则包含了游戏的UML图,详细展示了游戏的设计结构和各个模块之间的关系,对于理解游戏的整体架构非常有帮助。这个Java小游戏项目是一个非常好的学习资源,无论是对于初学者还是有一定经验的开发者来说,都可以通过这个项目来提升自己的编程技能和游戏设计能力。通过阅读代码和设计文档,你可以了解到如何构建一个功能完整的游戏,并且可以根据自己的需要进行修改和扩展。总之,这个Java小游戏项目是一个值得学习和探索的资源,希望对你有所帮助!

    ad3-2.2.1-cp34-cp34m-win_amd64.whl.rar

    python whl离线安装包 pip安装失败可以尝试使用whl离线安装包安装 第一步 下载whl文件,注意需要与python版本配套 python版本号、32位64位、arm或amd64均有区别 第二步 使用pip install XXXXX.whl 命令安装,如果whl路径不在cmd窗口当前目录下,需要带上路径 WHL文件是以Wheel格式保存的Python安装包, Wheel是Python发行版的标准内置包格式。 在本质上是一个压缩包,WHL文件中包含了Python安装的py文件和元数据,以及经过编译的pyd文件, 这样就使得它可以在不具备编译环境的条件下,安装适合自己python版本的库文件。 如果要查看WHL文件的内容,可以把.whl后缀名改成.zip,使用解压软件(如WinRAR、WinZIP)解压打开即可查看。 为什么会用到whl文件来安装python库文件呢? 在python的使用过程中,我们免不了要经常通过pip来安装自己所需要的包, 大部分的包基本都能正常安装,但是总会遇到有那么一些包因为各种各样的问题导致安装不了的。 这时我们就可以通过尝试去Python安装包大全中(whl包下载)下载whl包来安装解决问题。

    arctic-1.67.1-cp36-cp36m-win32.whl.rar

    python whl离线安装包 pip安装失败可以尝试使用whl离线安装包安装 第一步 下载whl文件,注意需要与python版本配套 python版本号、32位64位、arm或amd64均有区别 第二步 使用pip install XXXXX.whl 命令安装,如果whl路径不在cmd窗口当前目录下,需要带上路径 WHL文件是以Wheel格式保存的Python安装包, Wheel是Python发行版的标准内置包格式。 在本质上是一个压缩包,WHL文件中包含了Python安装的py文件和元数据,以及经过编译的pyd文件, 这样就使得它可以在不具备编译环境的条件下,安装适合自己python版本的库文件。 如果要查看WHL文件的内容,可以把.whl后缀名改成.zip,使用解压软件(如WinRAR、WinZIP)解压打开即可查看。 为什么会用到whl文件来安装python库文件呢? 在python的使用过程中,我们免不了要经常通过pip来安装自己所需要的包, 大部分的包基本都能正常安装,但是总会遇到有那么一些包因为各种各样的问题导致安装不了的。 这时我们就可以通过尝试去Python安装包大全中(whl包下载)下载whl包来安装解决问题。

    基于Java实现的黄金矿工小游戏.zip

    本资源是一个基于Java实现的黄金矿工小游戏项目,旨在帮助初学者通过实践巩固Java编程知识。游戏包含多个功能模块,如窗口绘制、图片绘制、红线摇摆及抓取判定等,并采用双缓存技术解决画面闪动问题。此外,还实现了金块和石块的随机生成与抓取机制、积分设置、关卡设置以及商店购物等功能。本项目适合刚入门或有一定基础的Java学习者,通过完成这个项目,可以提升面向对象编程的理解和应用能力,同时增强对Java基础知识的掌握。

    课设毕设基于SpringBoot+Vue的大学生心理咨询平台源码可运行.zip

    本压缩包资源说明,你现在往下拉可以看到压缩包内容目录 我是批量上传的基于SpringBoot+Vue的项目,所以描述都一样;有源码有数据库脚本,系统都是测试过可运行的,看文件名即可区分项目~ |Java|SpringBoot|Vue|前后端分离| 开发语言:Java 框架:SpringBoot,Vue JDK版本:JDK1.8 数据库:MySQL 5.7+(推荐5.7,8.0也可以) 数据库工具:Navicat 开发软件: idea/eclipse(推荐idea) Maven包:Maven3.3.9+ 系统环境:Windows/Mac

    网络直播带货查询系统 SSM毕业设计 附带论文.zip

    网络直播带货查询系统 SSM毕业设计 附带论文 启动教程:https://www.bilibili.com/video/BV1GK1iYyE2B

    Assimulo-3.1-cp35-cp35m-win_amd64.whl.rar

    python whl离线安装包 pip安装失败可以尝试使用whl离线安装包安装 第一步 下载whl文件,注意需要与python版本配套 python版本号、32位64位、arm或amd64均有区别 第二步 使用pip install XXXXX.whl 命令安装,如果whl路径不在cmd窗口当前目录下,需要带上路径 WHL文件是以Wheel格式保存的Python安装包, Wheel是Python发行版的标准内置包格式。 在本质上是一个压缩包,WHL文件中包含了Python安装的py文件和元数据,以及经过编译的pyd文件, 这样就使得它可以在不具备编译环境的条件下,安装适合自己python版本的库文件。 如果要查看WHL文件的内容,可以把.whl后缀名改成.zip,使用解压软件(如WinRAR、WinZIP)解压打开即可查看。 为什么会用到whl文件来安装python库文件呢? 在python的使用过程中,我们免不了要经常通过pip来安装自己所需要的包, 大部分的包基本都能正常安装,但是总会遇到有那么一些包因为各种各样的问题导致安装不了的。 这时我们就可以通过尝试去Python安装包大全中(whl包下载)下载whl包来安装解决问题。

    abcview-1.0.8-py2-none-any.whl.rar

    python whl离线安装包 pip安装失败可以尝试使用whl离线安装包安装 第一步 下载whl文件,注意需要与python版本配套 python版本号、32位64位、arm或amd64均有区别 第二步 使用pip install XXXXX.whl 命令安装,如果whl路径不在cmd窗口当前目录下,需要带上路径 WHL文件是以Wheel格式保存的Python安装包, Wheel是Python发行版的标准内置包格式。 在本质上是一个压缩包,WHL文件中包含了Python安装的py文件和元数据,以及经过编译的pyd文件, 这样就使得它可以在不具备编译环境的条件下,安装适合自己python版本的库文件。 如果要查看WHL文件的内容,可以把.whl后缀名改成.zip,使用解压软件(如WinRAR、WinZIP)解压打开即可查看。 为什么会用到whl文件来安装python库文件呢? 在python的使用过程中,我们免不了要经常通过pip来安装自己所需要的包, 大部分的包基本都能正常安装,但是总会遇到有那么一些包因为各种各样的问题导致安装不了的。 这时我们就可以通过尝试去Python安装包大全中(whl包下载)下载whl包来安装解决问题。

    Teradata:TeradataSQL语言入门.docx

    Teradata:TeradataSQL语言入门.docx

    winlibs-x86-64-win32-seh-gcc-14.2.0-llvm-19.1.3-mingw-w64.zip

    winlibs-x86-64-win32-seh-gcc-14.2.0-llvm-19.1.3-mingw-w64.zip

    aicspylibczi-3.0.5-cp39-cp39-win_amd64.whl.rar

    python whl离线安装包 pip安装失败可以尝试使用whl离线安装包安装 第一步 下载whl文件,注意需要与python版本配套 python版本号、32位64位、arm或amd64均有区别 第二步 使用pip install XXXXX.whl 命令安装,如果whl路径不在cmd窗口当前目录下,需要带上路径 WHL文件是以Wheel格式保存的Python安装包, Wheel是Python发行版的标准内置包格式。 在本质上是一个压缩包,WHL文件中包含了Python安装的py文件和元数据,以及经过编译的pyd文件, 这样就使得它可以在不具备编译环境的条件下,安装适合自己python版本的库文件。 如果要查看WHL文件的内容,可以把.whl后缀名改成.zip,使用解压软件(如WinRAR、WinZIP)解压打开即可查看。 为什么会用到whl文件来安装python库文件呢? 在python的使用过程中,我们免不了要经常通过pip来安装自己所需要的包, 大部分的包基本都能正常安装,但是总会遇到有那么一些包因为各种各样的问题导致安装不了的。 这时我们就可以通过尝试去Python安装包大全中(whl包下载)下载whl包来安装解决问题。

    #-ssm-068-mysql-学生智能选课系统-.zip

    管理员用户: 1.管理员详情: 1.1查看个人信息; 1.2添加新的管理员;管理员的详细信息2.学生详情: 2.1 添加学生;学号,密码(与学号一样),姓名,性别,班级,联系电话,身份证号 2.2 查询所有学生;要有一个筛选的地方可以筛选学号,姓名,性别,班级,评论,(筛选的地方后面做一个查询的按钮); 下面做一个查询的页面,展示学号,姓名,性别,班级,密码,身份证号,成绩,后添加一个按钮(操作,可以删除该条学生记录,可以修改学生信息); 3.课程功能; 3.1 添加课程; 3.2 查询课程; 4.老师功能; 4.1添加老师; 4.2查询所有老师; 老师用户; 1.个人信息;2.打分功能; 做一个筛选(根据班级筛选,根据成绩排序)可以看到选了自己的课的学生信息, 3.任课信息; 3.学生功能; 1.查看个人信息 2.选课,展示跟自己专业相关课程的所有信息,最后做一个操作按钮 选课,也可以取消选课; 3.查看选课信息及成绩, 可以看到课程所有信息,和任课老师的姓名,电话, 以及打的平时成绩,考试成绩,最终成绩。。

    LabVIEW练习39,程序开始运行后要求用户输入密码

    程序开始运行后要求用户输入密码,密码正确时字符串显示控件显示 “欢迎进入”, 否则显示字符串“密码错误”,同时退出程序。

    使用java实现的简单飞机大战游戏.zip

    本资源提供了使用java实现的简单飞机大战游戏,是一款经典的2D射击游戏。玩家在游戏中控制一架飞机,通过键盘操作移动和发射子弹,击落敌机获得积分。当达到一定积分时,会出现Boss,增加游戏难度。游戏还包含碰撞检测、得分系统以及游戏状态管理等功能。该资源旨在帮助学习者掌握Java编程的各个方面,包括面向对象设计、图形界面编程等。通过这个项目,学习者可以深入理解如何使用Java语言实现一个具备基本功能的小游戏,并学会如何处理游戏中的交互、碰撞检测和动画效果等技术细节。本资源适合Java编程初学者及对游戏开发感兴趣的开发者学习和参考,是提升编程技能和项目经验的绝佳实践材料。

    基于java多线程的一款小游戏.zip

    本资源是一款基于Java多线程开发的小游戏,旨在通过实战项目帮助学习者深入理解多线程编程的概念和应用。游戏中涉及多个并发任务,如角色移动、碰撞检测和动画更新等,每个任务都由独立的线程处理,以实现更流畅的游戏体验。游戏设计简洁但不失趣味性,包括一个主窗体、游戏面板以及控制面板。玩家可以通过控制面板选择角色并开始游戏,角色将在游戏面板中进行奔跑或其他活动。游戏过程中,各线程协同工作,确保游戏的实时响应和高效运行。此外,该资源还提供了详细的代码注释和文档说明,方便学习者理解每一部分的功能和实现方式。通过本项目的学习,不仅可以掌握Java多线程编程的基本技能,还能提升对游戏开发流程的理解。本资源完全基于学习和研究目的,请勿用于商业用途。

    17年国赛,基于 python 实现的图像识别(彩图找黑点和绿点,单纯找黑点,优先绿点后黑点)

    【作品名称】:17年国赛,基于 python 实现的图像识别(彩图找黑点和绿点,单纯找黑点,优先绿点后黑点) 【适用人群】:适用于希望学习不同技术领域的小白或进阶学习者。可作为毕设项目、课程设计、大作业、工程实训或初期项目立项。 【项目介绍】: 17年国赛,基于 python 实现的图像识别(彩图找黑点和绿点,单纯找黑点,优先绿点后黑点) 17年国赛,基于 python 实现的图像识别(彩图找黑点和绿点,单纯找黑点,优先绿点后黑点) 17年国赛,基于 python 实现的图像识别(彩图找黑点和绿点,单纯找黑点,优先绿点后黑点) 检测黑点与绿点 色块检测 【资源声明】:本资源作为“参考资料”而不是“定制需求”,代码只能作为参考,不能完全复制照搬。需要有一定的基础看懂代码,自行调试代码并解决报错,能自行添加功能修改代码。

Global site tag (gtag.js) - Google Analytics