Jafka source analysis processor

Hello everyone , I meet you again , I'm the king of the whole stack .

Jafka Acceptor Accept client And the connection request after establishment ,Acceptor Will Socket Connect to Processor To deal with .Processor Proceed through the following processing steps client Processing of requests ：

1. Read client request .

2. basis client Different request types , Call the corresponding processing function to process .

Processor Read client Request is an interesting thing , There are two things to consider ： First of all , Request rules （Processor The request must be parsed according to certain rules ）. second , How to make sure that the reading of a request has ended （ Because it is a non blocking connection , It is likely that the first read operation read part of the requested data , Second to second N Only after reading once can the whole client Request to read complete ）. Let's analyze it in detail below client Format of request .

client The request first includes a int, The int Indicate this client Requested size (size). And then , The request includes one or two byte(short) Request type for （ Request types include ：CreaterRequest、DeleterRequest、FetchRequest、MultiFetchRequest、MultiProducerRequest、OffsetRequest and ProducerRequest. Then each request type has a fixed format . The figure below illustrates ProducerRequest The format of ：

After knowing the above format , Question two （ How to ensure that a request has been read ） It's very easy Got it .

First of all to “ Request length ” Allocate one 4byte Of ByteBuffer, Until it's time to Buffer Read full , Otherwise, the length has not been read .“ Request length ” After reading , Assign a to the request “ Request length ” The size of ByteBuffer, Until it's time to Buffer If it is full, it means that one request is read . After reading , basis “ Request type ” Call the corresponding processing function （Handler） To deal with . stay jafka in , Two of the above Buffer In the class BoundedByteBufferReceive Declaration and management .Processor Received Acceptor The distribution of socket After connection . Would be socke Connect to create a BoundedByteBufferReceive And with socket Connect to bind . Whenever it's time to socket Connect “ Can be read ” when . take BoundedByteBufferReceive Take it out and continue reading based on the last reading . Until a request is completely read , The detailed process is shown in the following code （Processor.read) Seen ：

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);
}
}
}

Copy

BoundedByteBufferReceive.readFrom For example, the following ： Mainly apply for two Buffer And constantly read data .

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;
    }

Copy

After reading ,Processor Can parse “ Request type ”, Call different... According to different request types Handler Processing corresponds to the request .

Copyright notice ： This article is the original article of the blogger , Blog , Do not reprint without permission .

Publisher ： Full stack programmer stack length , Reprint please indicate the source ：https://javaforall.cn/116982.html Link to the original text ：https://javaforall.cn