08、Kafka 实战 - 生产者:选择器处理网络请求

涉及网络通信时,一般使用选择器模型,选择器使用 Java NIO 异步非阻塞方式管理连接和读写请求,好处就是单个线程就能管理多个网络连接通道,生产者只需要使用一个选择器就能同时和 Kafka 集群的多个服务端进行网络通信

一、Java NIO 的一些概念

  • SocketChannel 客户端网络连接通道,底层的字节数据读写都发生在通道上,通道会和字节缓冲区一起使用(channel.read(buffer) / channel.write(buffer))
  • Selector 选择器,发送在通道上的事件有读和写,选择器通过选择键的方式监听读写事件
  • SelectionKey 选择键,将通道注册到选择器上,channel.register(selector) 返回选择键,读写事件发生时,通过选择键就可以得到对应的通道进行读写操作

二、客户端连接服务端并建立 Kafka 通道

  • 选择器的 connect() 方法创建客户端到指定远程服务器的网络连接,使用的是 Java NIO 的 SocketChannel 对象完成

  • 这里创建了更抽象的 KafkaChannel,并使用 SelectionKeys.attach(KafkaChannel) 将选择键和 KafkaChannel 关联起来

  • 轮询的时候可以使用 SelectionKeys.attachment() 获取对应的 KafkaChannel

  • 选择器还会维护一个节点编号和 KafkaChannel 的 Map,方便客户端根据节点编号获取 KafkaChannel

  • 关系梳理

  • SocketChannel 注册到 Selector 上返回 SelectionKeys,将 Selector 用于构建传输层,再把传输层用于构造 KafkaChannel

  • 这样 KafkaChannel 和 SocketChannel 通过键进行了关联,本质上 KafkaChannel 就是对 SocketChannel 的一层封装

/**
 * Begin connecting to the given address and add the connection to this nioSelector associated with the given id
 * number.
 * 开始连接到给定的地址,并将连接添加到与给定id号关联的nioSelector中。
 * <p>
 * Note that this call only initiates the connection, which will be completed on a future {@linkpoll(long)}
 * call. Check {@linkconnected()} to see which (if any) connections have completed after a given poll call.
 * 注意,此调用只启动连接,连接将在未来的{@linkpoll(long)}调用中完成。检查{@linkconnected()},查看在给定的轮询调用之后完成了哪些连接(如果有)。
 */
@Override
public void connect(String id, InetSocketAddress address, int sendBufferSize, int receiveBufferSize) throws IOException {
   
     
    if (this.channels.containsKey(id))
        throw new IllegalStateException("There is already a connection for id " + id);
    // Java NIO 代码
    // 获取 SocketChannel
    SocketChannel socketChannel = SocketChannel.open();
    // 设置非阻塞
    socketChannel.configureBlocking(false);
    Socket socket = socketChannel.socket();
    socket.setKeepAlive(true);
    // 设置参数,Producer 里面有些默认值
    if (sendBufferSize != Selectable.USE_DEFAULT_BUFFER_SIZE)
        socket.setSendBufferSize(sendBufferSize);
    if (receiveBufferSize != Selectable.USE_DEFAULT_BUFFER_SIZE)
        socket.setReceiveBufferSize(receiveBufferSize);
    // 默认值是 false,代表要开启 Nagle 的算法,会把网络中小的数据包收集起来组成一个大的包再发送
    socket.setTcpNoDelay(true);
    boolean connected;
    try {
   
     
        // 尝试连接,立马成功返回 true,很久才成功会返回 false
        connected = socketChannel.connect(address);
    } catch (UnresolvedAddressException e) {
   
     
        socketChannel.close();
        throw new IOException("Can't resolve address: " + address, e);
    } catch (IOException e) {
   
     
        socketChannel.close();
        throw e;
    }
    // 注册 Selector
    SelectionKey key = socketChannel.register(nioSelector, SelectionKey.OP_CONNECT);
    // 封装一个 KafkaChannel
    KafkaChannel channel = channelBuilder.buildChannel(id, key, maxReceiveSize);
    // 把 key 和 KafkaChannel 关联起来
    key.attach(channel);
    // 缓存起来
    this.channels.put(id, channel);
    // 正常情况下,网络不能完成连接
    // 如果连接上了,取消前面的注册 OP_CONNECT 事件
    if (connected) {
   
     
        // OP_CONNECT won't trigger for immediately connected channels 对于立即连接的通道,不会触发OP_CONNECT
        log.debug("Immediately connected to node {}", channel.id());
        immediatelyConnectedKeys.add(key);
        key.interestOps(0);
    }
}

三、Kafka 通道和网络传输层

  • 构建 Kafka 通道的传输层有多种实现,比如纯文本模式、sasl、ssl 加密模式,PlaintextTransportLayer 就是纯文本的传输层实现

  • 传输层面向底层的字节缓冲区,操作的是字节流,KafkaChannel 的读写操作会利用传输层操作底层字节缓冲区,从而构造出 NetworkReceive 和 Send 对象

  • Send 字节缓冲区表示要发送出去的数据,NetworkReceive 的 size 缓冲区表示数据的长度,buffer 缓冲区表示数据的内容

  • 传输层对 SocketChannel 做了轻量级的封装,和 SocketChannel 一样都实现了 ScatteringByteChannel, GatheringByteChannel

  • TransportLayer 作为 kafkaChannel 的成员变量,Selector 在调用 KafkaChannel read() 和 write() 时,最终会通过 NetworkReceive.readFrom() 和 Send.writeTo() 来调用 TransportLayer 底层的 SocketChannel read() write() 方法

四、Kafka 通道上的读写操作

1. write 调用流程

  • NetworkClient.send() -> Selector.send() -> KafkaChannel.setSend() 这时候已经绑定了一个 OP_WRITE 事件,同时将 Send 设置为成员变量
// NetworkClient.send()
public void send(ClientRequest request, long now) {
   
     
	......
	selector.send(request.request());
}
// Selector.send()
public void send(Send send) {
   
     
    // 获取到 KafkaChannel,往里面存放一个发送请求
    KafkaChannel channel = channelOrFail(send.destination());
    try {
   
     
        // setSend 绑定一个 OP_WRITE 事件
        channel.setSend(send);
    } catch (CancelledKeyException e) {
   
     
    	......
    }
}
// KafkaChannel.setSend()
public void setSend(Send send) {
   
     
    // 往KafkaChannel 里面绑定一个发送出去到请求
    this.send = send;
    // 往 Selector 上绑定一个 OP_WRITE 事件,绑定之后就可以发送请求了
    this.transportLayer.addInterestOps(SelectionKey.OP_WRITE);
}

  • Selector.poll() 会轮询监听,监听到 write 事件会调用 KafkaChannel.write() -> KafkaChannel.send() 方法调用 Send.writeTo() 方法,最终使用 TransportLayer.write() 方法调用 SocketChannel.write 方法写数据

  • 如果一次没有发送完,Selector.poll() 会循环继续发送,发送完了会取消写操作

// Selector.poll() 里面调用到了 Selector.pollSelectionKeys(),这里是监听读写操作的方法
private void pollSelectionKeys(...) {
   
     
......
	/* if channel is ready write to any sockets that have space in their buffer and for which we have data */
	/**
	 * TODO 处理发送请求事件
	 * 如果通道已准备好,则写入缓冲区中有空间且有数据的任何套接字
	 * selector 已经注册了一个 OP_WRITE 事件
	 */
	if (channel.ready() && key.isWritable()) {
   
     
	    // 获取到要发送的那个网络请求,是这个句代码就往服务端发送数据了
	    // 消息被发送出去之后,移除 OP_WRITE
	    Send send = channel.write();
	    // 已经完成消息发送
	    if (send != null) {
   
     
	        this.completedSends.add(send);
	        this.sensors.recordBytesSent(channel.id(), send.size());
	    }
	}
......
}
// KafkaChannel.write()
public Send write() throws IOException {
   
     
    Send result = null;
    // TODO send 就是发送方法
    if (send != null && send(send)) {
   
     
        result = send;
        send = null;
    }
    return result;
}
// KafkaChannel.send()
private boolean send(Send send) throws IOException {
   
     
    // TODO 最终执行发送请求的代码
    send.writeTo(transportLayer);
    // 如果发送完成,移除 OP_WRITE
    if (send.completed())
        transportLayer.removeInterestOps(SelectionKey.OP_WRITE);
    return send.completed();
}
// Send.writeTo()
@Override
public long writeTo(GatheringByteChannel channel) throws IOException {
   
     
    long written = channel.write(buffers);
    remaining -= written;
    ......
    return written;
}
// TransportLayer.write() 
@Override
public long write(ByteBuffer[] srcs) throws IOException {
   
     
    return socketChannel.write(srcs);
}

  • 一个完整的发送请求和对应的事件监听步骤

  • 设置 Send 请求到 KafkaChannel -> 注册写操作 -> 发送请求 -> Send 请求发送完成 -> 取消写操作

2. KafkaChannel write 具体步骤

  • 通过 KafkaChannel.setSend() 方法设置要发送的请求对象,并注册写事件
  • 客户端轮询到写事件时,会取出 KafkaChannel 中的发送请求,并发送给网络通道
  • 如果本次写操作没有全部完成,那么写事件依然存在,客户端还会再次轮询到写事件
  • 客户端新的轮询会继续发送请求,发送完成后就取消写事件,并设置返回结果
  • 请求发送完成后,加入到 completedSends 集合中,这个数据会被调用者使用
  • 请求全部发送完成,send 对象会被重置为空,下一次新的请求才可以继续进行

3. read 调用流程

  • 和 write 操作类似,一次没有 read 完,也会调用多次,只有缓冲区数据被填满了,才表示接收到一个完整的 NetworkReceive

  • Selector.poll() 监听到 read 事件,调用 KafkaChannel.read() -> NetworkReceive.readFrom() 调用 TransportLayer.read() 方法

  • 完成连接后会注册 OP_READ 事件,可以接收服务端发送回来的响应了

// Selector.poll() 里面调用到了 Selector.pollSelectionKeys(),这里是监听读写操作的方法
private void pollSelectionKeys(...) {
   
     
......
	/* if channel is ready read from any connections that have readable data */
	// 如果通道就绪,则从具有可读数据的任何连接读取
	if (channel.ready() && key.isReadable() && !hasStagedReceive(channel)) {
   
     
	    NetworkReceive networkReceive;
	    // 接收服务端发送回来的响应(请求)
	    // networkReceive 服务端发送回来的响应
	    // 不断的读取数据,里面还设计粘包拆包问题
	    while ((networkReceive = channel.read()) != null)
	        // 这里迭代保证读取完整的响应,如果没有读取完整,就一直读
	        addToStagedReceives(channel, networkReceive);
	}
......
}
// KafkaChannel.read()
public NetworkReceive read() throws IOException {
   
     
    NetworkReceive result = null;
    if (receive == null) {
   
     
        receive = new NetworkReceive(maxReceiveSize, id);
    }
    // 一直在读取数据
    receive(receive);
    // 是否读完一个完整的响应消息
    if (receive.complete()) {
   
     
        receive.payload().rewind();
        result = receive;
        receive = null;
    }
    return result;
}
private long receive(NetworkReceive receive) throws IOException {
   
     
    return receive.readFrom(transportLayer);
}

// NetworkReceive.readFromReadableChannel()
public long readFrom(ScatteringByteChannel channel) throws IOException {
   
     
    return readFromReadableChannel(channel);
}
public long readFromReadableChannel(ReadableByteChannel channel) throws IOException {
   
     
    int read = 0;
    if (size.hasRemaining()) {
   
     
        // 先读取 4字节 数据,代表后面消息体大小,如果还有剩余空间会一直循环读取
        int bytesRead = channel.read(size);
        if (bytesRead < 0)
            throw new EOFException();
        read += bytesRead;
        // 一直读取到没有剩余空间,说明已经读取到一个 4字节 的 int 类型的数
        if (!size.hasRemaining()) {
   
     
            size.rewind();
            int receiveSize = size.getInt();
            if (receiveSize < 0)
                throw new InvalidReceiveException("Invalid receive (size = " + receiveSize + ")");
            if (maxSize != UNLIMITED && receiveSize > maxSize)
                throw new InvalidReceiveException("Invalid receive (size = " + receiveSize + " larger than " + maxSize + ")");
            // 分配一个内存大小空间,就是刚刚读出来的 4字节 int 大小
            this.buffer = ByteBuffer.allocate(receiveSize);
        }
    }
    if (buffer != null) {
   
     
        // 读取数据
        int bytesRead = channel.read(buffer);
        if (bytesRead < 0)
            throw new EOFException();
        read += bytesRead;
    }
    return read;
}

  • 这里 NetworkReceive.readFrom() 会调用 NetworkReceive.readFromReadableChannel(),这个方法是一个处理粘包拆包问题的方法

4. KafkaChannel read 具体步骤

  • 客户端轮询到读事件时,调用 KafkaChannel.read() 方法,如果网络接收对象不存在,就新建一个
  • 客户端读取网络通道的数据,并将数据填充到网络连接对象
  • 如果本次读操作没有全部完成,客户端还会再次轮询到读事件
  • 客户端新的轮询会继续读取,如果读取完成,则返回结果
  • 读取完成后,放到暂时完成的列表中,这个数据会被调用者使用
  • 读取全部完成,重置网络接收对象为空,下一次新的读取请求才可以继续进行

五、选择器的轮询

  • 选择器的轮询根据选择键读写,分别调用 KafkaChannel 的 read() 和 write(),通过不断注册事件、执行事件处理、取消事件,客户端才会发送请求给客户端,并从服务端读取响应结果

  • 选择器在轮询到各种事件,要么被提前注册(CONNECT),要么在处理事件的时候被注册(finishConnect() 注册 READ,setSend() 注册 WRITE)

  • 都是交给 KafkaChannel 处理,通过底层网络连接通往的就是远程服务端节点,这就完成了服务端和客户端的通信

  • 不同的注册事件在选择器的轮询下,会触发不同的事件处理

  • 客户端建立连接时注册连接事件

  • 发送请求时注册写事件

只有成功连接后,写事件才会被接着选择到

  • 连接事件的会确认成功连接,并注册读事件
  • 写事件发生时会将请求发送到服务端,接着客户端就开始等待服务端返回响应结果
@Override
public void poll(long timeout) throws IOException {
   
     
    if (hasStagedReceives() || !immediatelyConnectedKeys.isEmpty())
        timeout = 0;
    // 从 Selector 上找到有多少个 key 注册了
    int readyKeys = select(timeout);
    if (readyKeys > 0 || !immediatelyConnectedKeys.isEmpty()) {
   
     
        pollSelectionKeys(this.nioSelector.selectedKeys(), false, endSelect);
        pollSelectionKeys(immediatelyConnectedKeys, true, endSelect);
    }
    // TODO 对 stagedReceives 里面的状态进行处理
    // 没有新的选择键,说明要读取的已经都读取完
    addToCompletedReceives();
    ......
}
private void pollSelectionKeys(Iterable<SelectionKey> selectionKeys,
                               boolean isImmediatelyConnected,
                               long currentTimeNanos) {
   
     
    // 遍历所有的 key
    Iterator<SelectionKey> iterator = selectionKeys.iterator();
    while (iterator.hasNext()) {
   
     
        SelectionKey key = iterator.next();
        iterator.remove();
        // 根据 key 找到对应的 KafkaChannel
        KafkaChannel channel = channel(key);
        try {
   
     
            /* complete any connections that have finished their handshake (either normally or immediately) */
            // 完成任何已经完成握手的连接(正常或立即)
            // OP_CONNECT
            if (isImmediatelyConnected || key.isConnectable()) {
   
     
                // TODO 最后完成网络连接
                // 完成连接会绑定 OP_READ 事件
                if (channel.finishConnect()) {
   
     ......}
            /* if channel is not ready finish prepare */
            // 如果通道没有准备好
            if (channel.isConnected() && !channel.ready())
                channel.prepare();
            /* if channel is ready read from any connections that have readable data */
            // 如果通道就绪,则从具有可读数据的任何连接读取
            if (channel.ready() && key.isReadable() && !hasStagedReceive(channel)) {
   
     
                NetworkReceive networkReceive;
                // 接收服务端发送回来的响应(请求)
                // networkReceive 服务端发送回来的响应
                // 不断的读取数据,里面还设计粘包拆包问题
                while ((networkReceive = channel.read()) != null)
                    // 这里迭代保证读取完整的响应,如果没有读取完整,就一直读
                    addToStagedReceives(channel, networkReceive);
            }
            /* if channel is ready write to any sockets that have space in their buffer and for which we have data */
            /**
             * TODO 处理发送请求事件
             * 如果通道已准备好,则写入缓冲区中有空间且有数据的任何套接字
             * selector 已经注册了一个 OP_WRITE 事件
             */
            if (channel.ready() && key.isWritable()) {
   
     
                // 获取到要发送的那个网络请求,是这个句代码就往服务端发送数据了
                // 消息被发送出去之后,移除 OP_WRITE
                Send send = channel.write();
                // 已经完成消息发送
                if (send != null) {
   
     
                    this.completedSends.add(send);
                    this.sensors.recordBytesSent(channel.id(), send.size());
                }
            }
        } catch (Exception e) {
   
     
        	......
        }
    }
}