- 這篇將是網(wǎng)絡(luò)層源碼分析的最后一篇
- 北京的天,無力吐槽啊~
- 快年底了, 每個(gè)團(tuán)隊(duì)都在旁邊錄制新年寄語,各種口號(hào)~
對(duì)nio的封裝:Selector類
- 所在文件: clients/src/main/java/org/apache/kafka/commmon/network/Selector.java
- 源碼中的注釋:
A nioSelector interface for doing non-blocking multi-connection network I/O. This class works with NetworkSend} and NetworkReceive to transmit size-delimited network requests and responses.
- 重要函數(shù)解析:
(1) register(String id, SocketChannel socketChannel): 注冊(cè)這個(gè)socketChannel到一個(gè)nio selector, 將其讀事件添加到selector的監(jiān)聽隊(duì)列; 這個(gè)socketChannel通常是服務(wù)器接收到的客戶端的連接:
SelectionKey key = socketChannel.register(nioSelector, SelectionKey.OP_READ);
同時(shí)創(chuàng)建KafkaChannel, 負(fù)責(zé)實(shí)際的數(shù)據(jù)接收和發(fā)送:
KafkaChannel channel = channelBuilder.buildChannel(id, key, maxReceiveSize);
key.attach(channel);
this.channels.put(id, channel);
上面的id即為我們?cè)谏掀榻B的非常重要的ConnectionId;
(2) connect: 使用nio的SocketChannel連接到給定的地址,并且注冊(cè)到nio selector,同時(shí)也創(chuàng)建了KafkaChannel,負(fù)責(zé)實(shí)際的數(shù)據(jù)接收和發(fā)送;
SocketChannel socketChannel = SocketChannel.open();
socketChannel.configureBlocking(false);
Socket socket = socketChannel.socket();
socket.setKeepAlive(true);
socketChannel.connect(address);
SelectionKey key = socketChannel.register(nioSelector, SelectionKey.OP_CONNECT);
KafkaChannel channel = channelBuilder.buildChannel(id, key, maxReceiveSize);
key.attach(channel);
this.channels.put(id, channel);
(3) poll: 核心函數(shù):
Do whatever I/O can be done on each connection without blocking. This includes completing connections, completing disconnections, initiating new sends, or making progress on in-progress sends or receives.
處理作為客戶端的主動(dòng)連接事件:
if (key.isConnectable()) {
channel.finishConnect();
this.connected.add(channel.id());
this.sensors.connectionCreated.record();
}
處理連接建立或接收后的ssl握手或sasl簽權(quán)操作:
if (channel.isConnected() && !channel.ready())
channel.prepare();
處理觸發(fā)的讀事件:
if (channel.ready() && key.isReadable() && !hasStagedReceive(channel)) {
NetworkReceive networkReceive;
while ((networkReceive = channel.read()) != null)
addToStagedReceives(channel, networkReceive);
}
使用一個(gè)while循環(huán)力求每次讀事件觸發(fā)時(shí)都讀盡可能多的數(shù)據(jù);
channel.read()里會(huì)作拆包處理(后面會(huì)講到),返回非null表示當(dāng)前返回的NetworkReceive里包含了完整的應(yīng)用層協(xié)議數(shù)據(jù);
處理觸發(fā)的寫事件:
if (channel.ready() && key.isWritable()) {
Send send = channel.write();
if (send != null) {
this.completedSends.add(send);
this.sensors.recordBytesSent(channel.id(), send.size());
}
}
需要發(fā)送數(shù)據(jù)通過調(diào)用Selector::send方法,設(shè)置封裝了寫數(shù)據(jù)的NetworkSend,再將這個(gè)NetworkSend通過KafkaChannel::setSend接口設(shè)置到KafkaChannel,同時(shí)將寫事件添加到selector的監(jiān)聽隊(duì)列中,等待寫事件被觸發(fā)后,通過KafkaChannel::write將數(shù)據(jù)發(fā)送出去;
addToCompletedReceives()
將當(dāng)前接收到的完整的的request到添加到completedReceives中,上一篇中介紹的SocketServer會(huì)作completedReceives中取出這些request作處理;
封裝對(duì)單個(gè)連接的讀寫操作:KafkaChannel類
- 所在文件: clients/src/main/java/org/apache/kafka/common/network/KafkaChannel.java
- 包括transportLayer和authenticator, 完成ssh握手,sasl簽權(quán),數(shù)據(jù)的接收和發(fā)送;
傳輸層:TransportLayer類
- 所在文件 clients/src/main/java/org/apache/kafka/common/network/TransportLayer.java
- 兩個(gè)子類: PlanintextTransportLayer和SslTransportLayer
- PlanintextTransportLayer的實(shí)現(xiàn)主要是通過NetworkReceive和NetworkSend;
- SslTransportLayer的實(shí)現(xiàn)主要是通過SocketChannel,ByteBuffers和SSLEngine實(shí)際了加密數(shù)據(jù)的接收和發(fā)送(看到ssl就頭大啊,這部分先忽略~~~);
Kafka協(xié)議的包結(jié)構(gòu):
- 前4個(gè)字節(jié)固定, 值是后面的實(shí)際數(shù)據(jù)的長(zhǎng)度;
- NetworkReceive: 接收時(shí)先接收4個(gè)字節(jié), 獲取到長(zhǎng)度,然后再接收實(shí)際的數(shù)據(jù);
- NetworkSend: 發(fā)送時(shí)實(shí)際數(shù)據(jù)前先加上4個(gè)字節(jié)的數(shù)據(jù)長(zhǎng)度再發(fā)送;
上圖:
