Kafka Producer 消息收发设计
前几篇文章分析了Kafka的发送流程以及NIO的使用方式,但是还是留下了不少坑,这里就对剩下的问题做一个总结。
收到的数据为什么要缓存起来?
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何时就绪?
上一篇文章虽然讲了epoll的原理,但是我相信还是有人觉得很迷惘,这里换个简单的说法再说下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中取出来进行发送。