当Kafka启动时,会启动这个SocketServer来接收客户端的连接,处理客户端请求,发送响应。
这个类的注释说明了这个socket server的结构
/**
* An NIO socket server. The threading model is
* 1 Acceptor thread that handles new connections
* N Processor threads that each have their own selector and read requests from sockets
* M Handler threads that handle requests and produce responses back to the processor threads for writing.
*/
即
其中:
其中使用的类主要包括:
Acceptor : 它是一个SocketServer, 接受新的连接,并且分配连接给Processor
Processor: 读取请求,发送响应
Handler: 处理请求,产生响应。这里的Handler由kafka.server.RequestHandler实现。
RequestChannel: 它包括了一个request queue 和 一个 response queue. 是Handler和Processsor交互时使用的队列。Request由Processor放入RequestChannel, 由Handler取出,然后把Response放回RequestChannel.
Acceptor在接受连接后,就把相当的SocketChannel设成非阻塞模式。因此Processor对这些SocketChannel的读写都是使用Selector,采用非阻塞的处理模式。
问题:
(1) Acceptor是如何把新来的连接分配给对应的Processor,这个算法是什么?是round robin吗?
(2) Processor应该把对应的SocketChannel在自己的Selector上如何注册?
(3) Processor如何读取请求?
/* * Process reads from ready sockets */ def read(key: SelectionKey) { val socketChannel = channelFor(key) //获取可读的SocketChannel var receive = key.attachment.asInstanceOf[Receive] //获取attach到SelectionKey的Receive对象 if(key.attachment == null) { //如果attachment是空,说明这是第一次读,就新建一个Receive对象,attach到这个SocketChannel的SelectionKey上。如果不是空,说明之前已经从中读了一些数据,只是没读完。 receive = new BoundedByteBufferReceive(maxRequestSize) key.attach(receive) } val read = receive.readFrom(socketChannel) //从SocketChannel中读数据 val address = socketChannel.socket.getRemoteSocketAddress(); trace(read + " bytes read from " + address) if(read < 0) { //如果读的数据数小于0,就关闭socket连接。实际上从BoundedByteBufferReceive的实现来看,read的值不会小于0 close(key) } else if(receive.complete) {//如果读完了,就构造request,发送给requestChannel val req = RequestChannel.Request(processor = id, requestKey = key, buffer = receive.buffer, startTimeMs = time.milliseconds, remoteAddress = address) requestChannel.sendRequest(req) key.attach(null) //取消attach的Receive对象 // explicitly reset interest ops to not READ, no need to wake up the selector just yet key.interestOps(key.interestOps & (~SelectionKey.OP_READ))//显示地把这个SocketChannel设为非OP_READ,等到Response发给这个SocketChannel以后,它会被再设为OP_READ,以继续处理来自这个SocketChannel的请求。 } else {//如果没有读完,就把这个SocketChannel注册为OP_READ,然后wakeup对应的selector,继续从SocketChannel中读数据。所以下一次再处理这个SocketChannel时,attach到它的SelectionKey的Receive对象就不是空了。 // more reading to be done trace("Did not finish reading, registering for read again on connection " + socketChannel.socket.getRemoteSocketAddress()) key.interestOps(SelectionKey.OP_READ) wakeup() } }
那么BoundedByteBufferReceive是如何知道一个请求读没读完呢?
原来每个Request的前4个字节标识了这个Request有多长,BoundedByteBufferReceive从SocketChannel中读取前4个字节,转换成整形,以这个整数为大小构造一个ByteBuffer,如果这个ByteBuffer没有写满,就说明请求的内容还没有读完。receive.complete就不被设为true,否则就说明这个Request已经从channel中完全读出。
if(!contentBuffer.hasRemaining) { contentBuffer.rewind() complete = true }
Request Header (all single non-multi requests begin with this)
0
1
2
3
0
1
2
3
4
5
6
7
8
9
0
1
2
3
4
5
6
7
8
9
0
1
2
3
4
5
6
7
8
9
0
1
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| REQUEST_LENGTH |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| REQUEST_TYPE | TOPIC_LENGTH |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
/ /
/ TOPIC (variable length) /
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| PARTITION |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+