摘要:Kafka网络模块之Server端,介绍Server端启动、接收请求和处理请求的过程。
SocketServer 是 Kafka server 端用于处理请求的模块,在 Kafka 启动过程创建、初始化、启动。
SocketServer启动过程:
- 按照 endpoint 顺序初始化 Acceptor,每个 endpoint 对应一个 Acceptor,为每个 Acceptor 创建 Processor(数量由 num.network.threads 配置项决定),并启动 Acceptor,Acceptor 启动后会通过 selector 监听连接,并将新建立的连接交给 Processor 处理(轮询选择 Processor)
-
启动所有 Processor
Acceptor启动、监听连接过程:
- Acceptor启动后,会创建一个 serverSocketChannel,监听在该 acceptor 对应的 endpoint 上,并在 selector 上注册 OP_ACCEPT,然后进入死循环,每次循环,通过 selector 获取就绪的 key(即前面注册的 serverSocketChannel),表明有连接来到,然后通过 accept() 创建一个和该连接对应的 socketChannel,然后从该 acceptor 负责的 processors 中轮询选择一个,将该 socketChannel 交给选择的 processor 处理,即将连接交给 processor。
- Acceptor 将连接交给 processor 处理,是将 socketChannel 加入 processor 的连接队列 newConnection 中,processor 在 run 方法中会不断地从中获取并处理。
- Processor 从 newConnection 获取到 socketChannel 后,在 selector 上注册 OP_READ,并创建对应的 KafkaChannel。
Server端接收请求、处理的过程:
- Processor 收到 OP_READ 的事件就绪后,检查并尝试完成 SSL 握手和 SASL 校验(此时不一定握手完成,所以在 Processor 收到 OP_READ 的事件就绪后,要先检查并确保握手已经完成,SSL/SASL相关参考 9.4 节)
- SSL 握手和 SASL 校验完成后,从 channel 中读取数据,构造 NetworkReceive 对象,并入队 stagedReceives
- 取出 stagedReceives 队首元素(移除),加入 completedReceives
- 将 completedReceives 中的元素取出(不移除),构造 Request 对象,加入 requestQueue,移除对 OP_READ 的事件注册,并将对应的 KafkaChannel 置为 MUTED,再置为 MUTED_AND_RESPONSE_PENDING
- KafkaRequestHandler 从 requestQueue 取出元素(移除),并交给 KafkaApi 模块处理请求
- KafkaApi 处理完请求后,将响应放入对应的 processor 的 responseQueue 和 inflightResponses 中,并唤醒其 selector
- Processor 从 responseQueue 中取出响应(移除),若响应是需要发回给客户端的,则将响应的 send 赋值给 KafkaChannel,并注册 OP_WRITE 事件
- 当 channel 写就绪后,将 send 写入 channel 的写 buffer,当 send 写完后,移除对 OP_WRITE 事件的注册,并将 send 加入 completedSends
- 从 inflightResponses 中移除对应的响应,执行响应回调,将 KafkaChannel 置为 MUTED,再从 MUTED 置为 NOT_MUTED,并重新添加 OP_READ 事件注册