本文共 8404 字，大约阅读时间需要 28 分钟。

前言

1. 粘包问题：一个请求里面带有多个响应，多个消息粘再一起给你发送回来；
2. 拆包问题：一个消息拆成多个请求发送回来；

粘包

private void pollSelectionKeys(Iterable
   
     selectionKeys,                                   boolean isImmediatelyConnected,                                   long currentTimeNanos) {
           //获取到所有key        Iterator
    
      iterator = selectionKeys.iterator();        //遍历所有的key        while (iterator.hasNext()) {
               SelectionKey key = iterator.next();            iterator.remove();            //根据key找到对应的KafkaChannel            KafkaChannel channel = channel(key);            // register all per-connection metrics at once            sensors.maybeRegisterConnectionMetrics(channel.id());            if (idleExpiryManager != null)                idleExpiryManager.update(channel.id(), currentTimeNanos);            try {
                   /* complete any connections that have finished their handshake (either normally or immediately) */                /**                 *                 * 我们代码第一次进来应该要走的是这儿分支，因为我们前面注册的是                 * SelectionKey key = socketChannel.register(nioSelector,                 * SelectionKey.OP_CONNECT);                 *                 */                if (isImmediatelyConnected || key.isConnectable()) {
                       //TODO 核心的代码来了                    //去最后完成网络的连接                    //如果我们之前初始化的时候，没有完成网络连接的话，这儿一定会帮你                    //完成网络的连接。                    if (channel.finishConnect()) {
                           //网络连接已经完成了以后，就把这个channel存储到                        this.connected.add(channel.id());                        this.sensors.connectionCreated.record();                        SocketChannel socketChannel = (SocketChannel) key.channel();                        log.debug("Created socket with SO_RCVBUF = {}, SO_SNDBUF = {}, SO_TIMEOUT = {} to node {}",                                socketChannel.socket().getReceiveBufferSize(),                                socketChannel.socket().getSendBufferSize(),                                socketChannel.socket().getSoTimeout(),                                channel.id());                    } else                        continue;                }                /* if channel is not ready finish prepare */                if (channel.isConnected() && !channel.ready())                    channel.prepare();                /* if channel is ready read from any connections that have readable data */                if (channel.ready() && key.isReadable() && !hasStagedReceive(channel)) {
                       NetworkReceive networkReceive;                    //接受服务端发送回来的响应（请求）                    //networkReceive 代表的就是一个服务端发送                    //回来的响应                    while ((networkReceive = channel.read()) != null)                        addToStagedReceives(channel, networkReceive);                }                /* if channel is ready write to any sockets that have space in their buffer and for which we have data */                //核心代码，处理发送请求的事件                //selector 注册了一个OP_WRITE                //selector 注册了一个OP_READ                if (channel.ready() && key.isWritable()) {
                       //获取到我们要发送的那个网络请求。                    //是这句代码就是要往服务端发送数据了。                    Send send = channel.write();                    if (send != null) {
                           this.completedSends.add(send);                        this.sensors.recordBytesSent(channel.id(), send.size());                    }                }                /* cancel any defunct sockets */                if (!key.isValid()) {
                       close(channel);                    this.disconnected.add(channel.id());                }            } catch (Exception e) {
                   String desc = channel.socketDescription();                if (e instanceof IOException)                    log.debug("Connection with {} disconnected", desc, e);                else                    log.warn("Unexpected error from {}; closing connection", desc, e);                close(channel);                this.disconnected.add(channel.id());            }        }    }
    
   

其中

if (channel.ready() && key.isReadable() && !hasStagedReceive(channel)) {
          NetworkReceive networkReceive;       //接受服务端发送回来的响应（请求）       //networkReceive 代表的就是一个服务端发送       //回来的响应       while ((networkReceive = channel.read()) != null)           addToStagedReceives(channel, networkReceive);   }

public NetworkReceive read() throws IOException {
           NetworkReceive result = null;        if (receive == null) {
               receive = new NetworkReceive(maxReceiveSize, id);        }        //一直在读取数据。        receive(receive);        //是否读完一个完整的响应消息        if (receive.complete()) {
               receive.payload().rewind();            result = receive;            receive = null;        }        return result;    }

public boolean complete() {
           //size 没有剩余空间（size读满） &&        return !size.hasRemaining() && !buffer.hasRemaining();    }

一直递归调用，最终处理粘包问题核心代码

public long readFromReadableChannel(ReadableByteChannel channel) throws IOException {
           int read = 0;        //size是一个4字节大小的内存空间        //如果size还有剩余的内存空间。        if (size.hasRemaining()) {
               //先读取4字节的数据，（代表的意思就是后面跟着的消息体的大小）            int bytesRead = channel.read(size);            if (bytesRead < 0)                throw new EOFException();            read += bytesRead;            //一直要读取到当这个size没有剩余空间            //说明已经读取到了一个4字节的int类型的数了。            if (!size.hasRemaining()) {
                   size.rewind();                //                int receiveSize = size.getInt();                if (receiveSize < 0)                    throw new InvalidReceiveException("Invalid receive (size = " + receiveSize + ")");                if (maxSize != UNLIMITED && receiveSize > maxSize)                    throw new InvalidReceiveException("Invalid receive (size = " + receiveSize + " larger than " + maxSize + ")");                //分配一个内存空间，这个内存空间的大小                //就是刚刚读出来的那个4字节的int的大小。                this.buffer = ByteBuffer.allocate(receiveSize);            }        }        if (buffer != null) {
               //去读取数据            int bytesRead = channel.read(buffer);            if (bytesRead < 0)                throw new EOFException();            read += bytesRead;        }        return read;    }

拆包

50生产者|60如何处理

1. 消息体的size
2. 消息体

1. 消息体的size

public long readFromReadableChannel(ReadableByteChannel channel) throws IOException {
           int read = 0;        //size是一个4字节大小的内存空间        //如果size还有剩余的内存空间。        if (size.hasRemaining()) {
               //先读取4字节的数据，（代表的意思就是后面跟着的消息体的大小）            int bytesRead = channel.read(size);            if (bytesRead < 0)                throw new EOFException();            read += bytesRead;            //一直要读取到当这个size没有剩余空间            //说明已经读取到了一个4字节的int类型的数了。            //这里如果size没有读满，就不会去读消息体的信息            if (!size.hasRemaining()) {
                   size.rewind();                //                int receiveSize = size.getInt();                if (receiveSize < 0)                    throw new InvalidReceiveException("Invalid receive (size = " + receiveSize + ")");                if (maxSize != UNLIMITED && receiveSize > maxSize)                    throw new InvalidReceiveException("Invalid receive (size = " + receiveSize + " larger than " + maxSize + ")");                //分配一个内存空间，这个内存空间的大小                //就是刚刚读出来的那个4字节的int的大小。                this.buffer = ByteBuffer.allocate(receiveSize);            }        }        if (buffer != null) {
               //去读取数据            int bytesRead = channel.read(buffer);            if (bytesRead < 0)                throw new EOFException();            read += bytesRead;        }        return read;    }    public ByteBuffer payload() {
           return this.buffer;    }}

2. 消息体

public NetworkReceive read() throws IOException {
           NetworkReceive result = null;        if (receive == null) {
               receive = new NetworkReceive(maxReceiveSize, id);        }        //一直在读取数据。        receive(receive);        //是否读完一个完整的响应消息        if (receive.complete()) {
               receive.payload().rewind();            result = receive;            receive = null;        }        return result;    }

public boolean complete() {
           //size 没有剩余空间（size读满） &&后面的条件保证 只有消息体填满才可以继续读下个消息        return !size.hasRemaining() && !buffer.hasRemaining();    }

转载地址：https://blog.csdn.net/weixin_37850264/article/details/112338914 如侵犯您的版权，请留言回复原文章的地址，我们会给您删除此文章，给您带来不便请您谅解！