讯息写

2013-12-13

消息写处理流程正如JDK规范中指出一个Channel任意时刻只能执行单个线程的写操作。单个Nioworker可以顺序处

消息写
   处理流程
   正如JDK规范中指出一个Channel任意时刻只能执行单个线程的写操作。单个Nioworker可以顺序处理多个socketChannel的写操作，单个SocketChannel上的多次写操作会事先放入到写请求队列；结果由Nioworker调度执行。当Channel被调度时正常情况下消息队列会被出列处理之至为空；如果中途Channel内核缓冲区已满，未写入数据将保留在内存中则下次Channel被调度时优先处理上次剩余数据之后出列可能的消息。
   而且一旦内核缓存区满并尝试n次写之后失败，将挂起写操作之后该channel的写入全部保存在消息队里中，等待下一次调度。其实挂起写操作时，selectKey可能修改为支持OP_WRITE，下次select()被唤醒时解挂。在此之前客户端发起的写全部被保存到消息队列中；以下是单个channel的写操作流程。
写消息逻辑大部分在如下方法中

  //AbstractNioWorker   protected void write0(AbstractNioChannel<?> channel)

   写操作数据结构
之前所有的消息事件中包含消息数据，而这种数据类型由SocketSendBufferPool强制规定；由此可见消息数据必须是ChannelBuffer或FileRegion类型。acquire()方法将所需类型经过处理转变为SendBuffer。它的实现包括
     处理FileRegion的FileSendBuffer
    处理ChannelBuffer的UnpooledSendBuffer,GatheringSendBuffer
    处理ChannelBuffer的PooledSendBuffer
第1,2种类型比较简单，只是一个简单的代理；重点是第3种类型。单个nioworker中多个Channel会可能会使用多块DirectBuffer；关于DirectBuffer在读消息中详细说明过。
如果写入信息块ChannelBuffer小于等于1024,将返回PooledSendBuffer。当每个Channel消息队列正常处理完，则会立即释放引用计数器。不然PooledSendBuffer会占用，一般来讲接下来的轮询会处理完，直到同段的最后一个处理完，才标志该内存段的为PooledHeader ，假如有一个Channel总是无法写入则占用的内存段不会被释放，这也是“内存泄露”。

  SendBuffer acquire(Object message) {        if (message instanceof ChannelBuffer) {            return acquire((ChannelBuffer) message);        }        if (message instanceof FileRegion) {            return acquire((FileRegion) message);        }        throw new IllegalArgumentException(                "unsupported message type: " + message.getClass());    }  //SendBuffer   interface SendBuffer {        boolean finished();        long writtenBytes();        long totalBytes();        long transferTo(WritableByteChannel ch) throws IOException;        long transferTo(DatagramChannel ch, SocketAddress raddr) throws IOException;        void release();    }   //PooledSendBuffer     final class PooledSendBuffer extends UnpooledSendBuffer {        private final Preallocation parent;        PooledSendBuffer(Preallocation parent, ByteBuffer buffer) {            super(buffer);            this.parent = parent;        }        @Override        public void release() {            final Preallocation parent = this.parent;            if (-- parent.refCnt == 0) {                parent.buffer.clear();                if (parent != current) {                    poolHead = new PreallocationRef(parent, poolHead);                }            }        }    }

PooledHead始终指向可用的链表头，详细的操作流程及描述如下图

热点排行

开源软件

讯息写