详述 Java NIO 以及 Socket 处理粘包和断包方法

private int mark = -1;
private int position = 0;
private int limit;
private int capacity;
long address;

// Step 1
RandomAccessFile file = new RandomAccessFile("C:\\Users\\niogeek\\Desktop\\testNIO.txt","rw");
FileChannel channel = file.getChannel();

// Step 2
ByteBuffer buffer = ByteBuffer.allocate(1024);
channel.read(buffer);

// Step 3
buffer.flip();
byte[] res = new byte[1024];
buffer.get(res, 0, buffer.limit());
System.out.println(new String(res));

// Step 4
channel.close();

public final FileChannel getChannel() {
    synchronized (this) {
        if (channel == null) {
            channel = FileChannelImpl.open(fd, path, true, rw, this);
        }
        return channel;
    }
}

public static FileChannel open(FileDescriptor var0, String var1, boolean var2, boolean var3, Object var4) {
    return new FileChannelImpl(var0, var1, var2, var3, false, var4);
}

public static ByteBuffer allocate(int capacity) {
    if (capacity < 0)
        throw new IllegalArgumentException();
    return new HeapByteBuffer(capacity, capacity);
}

HeapByteBuffer(int cap, int lim) {
    super(-1, 0, lim, cap, new byte[cap], 0);
}

    public int read(ByteBuffer var1) throws IOException {
        this.ensureOpen();
        if(!this.readable) {
            throw new NonReadableChannelException();
        } else {
            Object var2 = this.positionLock;
            synchronized(this.positionLock) {
                int var3 = 0;
                int var4 = -1;

                byte var5;
                try {
                    this.begin();
                    var4 = this.threads.add();
                    if(this.isOpen()) {
                        do {
                            var3 = IOUtil.read(this.fd, var1, -1L, this.nd);
                        } while(var3 == -3 && this.isOpen());

                        int var12 = IOStatus.normalize(var3);
                        return var12;
                    }

                    var5 = 0;
                } finally {
                    this.threads.remove(var4);
                    this.end(var3 > 0);

                    assert IOStatus.check(var3);

                }

                return var5;
            }
        }
    }

    static int read(FileDescriptor var0, ByteBuffer var1, long var2, NativeDispatcher var4) throws IOException {
        if(var1.isReadOnly()) {
            throw new IllegalArgumentException("Read-only buffer");
        } else if(var1 instanceof DirectBuffer) {
            return readIntoNativeBuffer(var0, var1, var2, var4);
        } else {
            ByteBuffer var5 = Util.getTemporaryDirectBuffer(var1.remaining());

            int var7;
            try {
                int var6 = readIntoNativeBuffer(var0, var5, var2, var4);
                var5.flip();
                if(var6 > 0) {
                    var1.put(var5);
                }

                var7 = var6;
            } finally {
                Util.offerFirstTemporaryDirectBuffer(var5);
            }

            return var7;
        }
    }

Selector selector = Selector.open();

//创建一个 TCP 套接字通道
SocketChannel channel = SocketChannel.open();
//调整通道为非阻塞模式
channel.configureBlocking(false);
//向选择器注册一个通道
SelectionKey key = channel.register(selector, SelectionKey.OP_READ);

int readySet = selectionKey.readyOps();

Set<SelectionKey> keys = selector.selectedKeys();

private void sendIntoChannel() {
 
        Runnable run = new Runnable() {
            @Override
            public void run() {
 
                try {
                    ByteArrayOutputStream bOut;
                    ObjectOutputStream out;
                    CBaseDataBean cbdb;
                    ByteBuffer bb = ByteBuffer.allocate(MemCache);
                    while (true) {
                        cbdb = CloudServer.cdsq.read();//Blocking Method
 
                        //处理自我命令:断开连接 退出线程
                        if (cbdb.getDataType() == CMsgTypeBean.MSG_TYPE_CUTDOWN) {
                            break;
                        }
 
                        bOut = new ByteArrayOutputStream();
                        out = new ObjectOutputStream(bOut);
                        out.writeObject(cbdb);
                        out.flush();
                        
                        //构造发送数据:整型数据头+有效数据段
                        byte[] arr = bOut.toByteArray();
                        final int ObjLength = arr.length;   //获取有效数据段长度                      
                        bb.clear();
                        bb.limit(IntLength + ObjLength);    //调整缓存池大小
                        bb.putInt(ObjLength);
                        bb.put(arr);
                        bb.position(0);                     //调整重置读写指针
 
                        SocketChannel channel = (SocketChannel) key.channel();
                        channel.write(bb);
 
                        out.close();
                        bOut.close();
                    }
                } catch (IOException ex) {
                }
            }
        };
        CloudServer.cstp.putNewThread(run);
    }

    /**
     * 开辟线程分发消息
     */
    private void Dispatcher() {
        Runnable run = new Runnable() {
 
            @Override
            public void run() {
                try {
                    while (true) {
                        selector.selectNow();
                        Thread.sleep(100);
                        Iterator<SelectionKey> itor = selector.selectedKeys().iterator();
                        while (itor.hasNext()) {
                            SelectionKey selKey = itor.next();
                            itor.remove();
 
                            if (selKey.isValid() && selKey.isAcceptable()) {
                                finshAccept(selKey);
                            }
 
                            if (selKey.isValid() && selKey.isReadable()) {
                                //消息分发
                                Processer();
                            }
                        }
                    }
                } catch (IOException | InterruptedException ex) {
                    System.out.println(ex.toString());
                }
            }
        };
        CloudServer.cstp.putNewThread(run);
    }
 
    /**
     * 消息处理器
     */
    private void Processer() {
        ByteBuffer bbInt = ByteBuffer.allocate(IntLength);    //读取INT头信息的缓存池
        ByteBuffer bbObj = ByteBuffer.allocate(MemCache);     //读取OBJ有效数据的缓存池
        SocketChannel channel = (SocketChannel)key.channel();
        ByteArrayInputStream bIn;
        ObjectInputStream in;
        CBaseDataBean cbdb;
        //有效数据长度
        int ObjLength;
        //从NIO信道中读出的数据长度
        int readObj;
        try {
            //读出INT数据头
            while (channel.read(bbInt) == IntLength) {
                //获取INT头中标示的有效数据长度信息并清空INT缓存池
                ObjLength = bbInt.getInt(0);
                bbInt.clear();
                
                //清空有效数据缓存池设置有效缓存池的大小
                bbObj.clear();
                bbObj.limit(ObjLength);
                
                //循环读满缓存池以保证数据完整性
                readObj = channel.read(bbObj);
                while (readObj != ObjLength) {
                    readObj += channel.read(bbObj);
                }
 
                bIn = new ByteArrayInputStream(bbObj.array());
                in = new ObjectInputStream(bIn);
                cbdb = (CBaseDataBean) in.readObject();
                switch (cbdb.getDataType()) {
                    case CMsgTypeBean.MSG_TYPE_COMMAND:
                        rcv_msg_command(cbdb);
                        break;
                    case CMsgTypeBean.MSG_TYPE_CUTDOWN:
                        rcv_msg_cutdown();
                        break;
                    case CMsgTypeBean.MSG_TYPE_VERIFYFILE:
                        rcv_msg_verifyfile(cbdb);
                        break;
                    case CMsgTypeBean.MSG_TYPE_SENDFILE:
                        rcv_msg_sendfile(cbdb);
                        break;
                    case CMsgTypeBean.MSG_TYPE_DISPATCHTASK:
                        rcv_msg_dispchtask(cbdb);
                        break;
                }
                in.close();
            
            }
        } catch (ClassNotFoundException | IOException ex) {
        }
    }