Jafka源代码分析——Processor


Jafka Acceptor接收到客户端请求并建立连接后，Acceptor会将Socket连接交给Processor进行处理。Processor通过以下的处理步骤进行客户端请求的处理：

1. 读取客户端请求。

2. 根据客户端请求类型的不同，调用相应的处理函数进行处理。

Processor读取客户端请求是一个比较有意思的事情，需要考虑两个方面的事情：第一，请求规则（Processor需要按照一定的规则进行请求的解析）；第二，如何确定一次请求的读取已经结束（因为是非阻塞连接，非常有可能第一次读操作读取了请求的一部分数据，第二次到第N次读取才能把整个客户端请求读取完整）。下面我们详细解析一下客户端请求的格式。客户端请求首先包含一个int，该int指明本次客户端请求的大小(size)。随后，请求包含一个两个byte(short)的请求类型（请求类型包括：CreaterRequest、DeleterRequest、FetchRequest、MultiFetchRequest、MultiProducerRequest、OffsetRequest和ProducerRequest。然后每种请求类型有固定的格式。下图具体说明了ProducerRequest的格式：



知道了上面的格式之后，问题二（如何确定一次请求已经读取完成）就很容易解决了。首先为“请求长度”分配一个4byte的ByteBuffer，直到该Buffer读满，否则说明长度一直没有读取完成。“请求长度”读取完成后，为请求分配一个“请求长度”大小的ByteBuffer，直到该Buffer读满则说明一次请求读取完成。读取完成后，根据“请求类型”调用相应的处理函数（Handler）进行处理。在jafka中，上述的两个Buffer在类BoundedByteBufferReceive中进行声明和管理。Processor接收到Acceptor分配的socket连接后，会为socke连接建立一个BoundedByteBufferReceive并将其与socket连接进行绑定。每当该socket连接“可读”时，将BoundedByteBufferReceive拿出来从上次读取的基础上继续读取，直到一次请求彻底读取完成，具体过程如下面代码（Processor.read)所示：

private void read(SelectionKey key) throws IOException {
SocketChannel socketChannel = channelFor(key);
Receive request = null;
request = new BoundedByteBufferReceive(maxRequestSize);
key.attach(request);
} else {
request = (Receive) key.attachment();
}
int read = request.readFrom(socketChannel);
stats.recordBytesRead(read);
if (read < 0) {
close(key);
} else if (request.complete()) {
Send maybeResponse = handle(key, request);
key.attach(null);
// if there is a response, send it, otherwise do nothing
if (maybeResponse != null) {
key.attach(maybeResponse);
key.interestOps(SelectionKey.OP_WRITE);
}
} else {
// more reading to be done
key.interestOps(SelectionKey.OP_READ);
getSelector().wakeup();
if (logger.isTraceEnabled()) {
logger.trace("reading request not been done. " + request);
}
}
}


BoundedByteBufferReceive.readFrom的实现具体如下：主要是申请两个Buffer并不断的读取数据。

public int readFrom(ReadableByteChannel channel) throws IOException {
        expectIncomplete();
        int read = 0;
        if (sizeBuffer.remaining() > 0) {
            read += Utils.read(channel, sizeBuffer);
        }
        if (contentBuffer == null && !sizeBuffer.hasRemaining()) {
            sizeBuffer.rewind();
            int size = sizeBuffer.getInt();
            if (size <= 0) {
                throw new InvalidRequestException(...);
            }
            if (size > maxRequestSize) {
                final String msg = "Request of length %d is not valid, it is larger than the maximum size of %d bytes.";
                throw new InvalidRequestException(format(msg, size, maxRequestSize));
            }
            contentBuffer = byteBufferAllocate(size);
        }
        //
        if (contentBuffer != null) {
            read = Utils.read(channel, contentBuffer);
            //
            if (!contentBuffer.hasRemaining()) {
                contentBuffer.rewind();
                setCompleted();
            }
        }
        return read;
    }


读取完成后，Processor会解析“请求类型”，根据请求类型的不同调用不同的Handler进行相应请求的处理。

相关内容