[TOC]
收到的数据为什么要缓存起来?
Kafka中Selector读取从远端回来的数据的时候会先把收到的数据缓存起来
private void attemptRead(SelectionKey key, KafkaChannel channel) throws IOException {
//if channel is ready and has bytes to read from socket or buffer, and has no
//previous receive(s) already staged or otherwise in progress then read from it
if (channel.ready() && (key.isReadable() || channel.hasBytesBuffered()) && !hasStagedReceive(channel)
&& !explicitlyMutedChannels.contains(channel)) {
NetworkReceive networkReceive;
while ((networkReceive = channel.read()) != null) {
madeReadProgressLastPoll = true;
addToStagedReceives(channel, networkReceive);
}
}
}
在NetworkClient中，往下传的是一个完整的ClientRequest，进到Selector，暂存到channel中的，也是一个完整的Send对象(1个数据包)。但这个Send对象，交由底层的channel.write(Bytebuffer b)的时候，并不一定一次可以完全发送，可能要调用多次write，才能把一个Send对象完全发出去。这是因为write是非阻塞的，不是等到完全发出去，才会返回。
Send send = channel.write();
if (send != null) {
this.completedSends.add(send);
this.sensors.recordBytesSent(channel.id(), send.size());
}
这里如果返回send==null就表示没有发送完毕，需要等到下一次Selector.poll再次进行发送。所以当下次发送的时候如果Channel里面的Send只发送了部分，那么此次这个node就不会处于ready状态，就不会从RecordAccumulator取出要往这个node发的数据,等到Send对象发送完毕之后，这个node才会处于ready状态，就又可以取出数据进行处理了。
同样，在接收的时候，channel.read(Bytebuffer b)，一个response也可能要read多次，才能完全接收。所以就有了上面的while循环代码。
如何确定消息接收完成?
从上面知道，底层数据的通信，是在每一个channel上面，2个源源不断的byte流，一个send流，一个receive流。
send的时候，还好说，发送之前知道一个完整的消息的大小。
但是当我们接收消息response的时候，这个信息可能是不完整的(剩余的数据要晚些才能获得)，也可能包含不止一条消息。那么我们是怎么判断消息发送完毕的呢？
对于消息的读取我们必须考虑消息结尾是如何表示的,标识消息结尾通常有以下几种方式：
- 固定的消息大小。
- 将消息的长度作为消息的前缀。
- 用一个特殊的符号来标识消息的结束。
很明显第一种和第三种方式不是很合适，因此Kafka采用了第二种方式来确定要发送消息的大小。在消息头部放入了4个字节来确定消息的大小。
//接收消息，前4个字节表示消息的大小
public class NetworkReceive implements Receive {
private final String source;
//确定消息size
private final ByteBuffer size;
private final int maxSize;
//整个消息response的buffer
private ByteBuffer buffer;
public NetworkReceive(String source) {
this.source = source;
//分配4字节的头部
this.size = ByteBuffer.allocate(4);
this.buffer = null;
this.maxSize = UNLIMITED;
}
}
//消息发送,前4个字节表示消息大小
public class NetworkSend extends ByteBufferSend {
public NetworkSend(String destination, ByteBuffer buffer) {
super(destination, sizeDelimit(buffer));
}
private static ByteBuffer[] sizeDelimit(ByteBuffer buffer) {
return new ByteBuffer[] {sizeBuffer(buffer.remaining()), buffer};
}
private static ByteBuffer sizeBuffer(int size) {
//4个字节表示消息大小
ByteBuffer sizeBuffer = ByteBuffer.allocate(4);
sizeBuffer.putInt(size);
sizeBuffer.rewind();
return sizeBuffer;
}
}
OP_WRITE何时就绪?
OP_WRITE事件的就绪条件并不是发生在调用channel的write方法之后，也不是发生在调用channel.register(selector,SelectionKey.OP_WRITE)后,而是在当底层缓冲区有空闲空间的情况下。因为写缓冲区在绝大部分时候都是有空闲空间的，所以如果你注册了写事件，这会使得写事件一直处于写就绪，选择处理现场就会一直占用着CPU资源。所以，只有当你确实有数据要写时再注册写操作，并在写完以后马上取消注册。
max.in.flight.requests.per.connection
这个参数指定了生产者在收到服务器响应之前可以发送多少个消息，找Kafka Producer中对应有一个类InFlightRequests,表示在天上飞的请求,也就是请求发出去了response还没有回来的请求数,这个参数也是判断节点是否ready的关键因素。只有ready的节点数据才能从Accumulator中取出来进行发送。
网友评论