Kafka源码分析－序列4 －Producer －network层核心原理

在上一篇我们分析了Java NIO的原理和使用方式，本篇将进一步分析Kafka client是如何基于NIO构建自己的network层。

network层的分层架构

下图展示了从最上层的KafkaProducer到最底层的Java NIO的构建层次关系：

图中淡紫色的方框表示接口或者抽象类，白色方框是具体实现。

整个架构图也体现了“面向接口编程”的思想：最底层Java NIO往上层全部以接口形式暴露，上面的3层，也都定义了相应的接口，逐层往上暴露。

接口的实例化（包括KafkaClient, Selectable, ChannelBuilder），也都在最外层的容器类KafkaProducer的构造函数中完成，KafkaProducer也就充当了一个“工厂”的角色，装配所有这些底层组件。

network层组件与NIO组件的映射关系

从上图也可以看出:

KakfaChannel基本是对SocketChannel的封装，只是这个中间多个一个间接层：TransportLayer，为了封装普通和加密的Channel；

Send/NetworkReceive是对ByteBuffer的封装，表示一次请求的数据包；

Kafka的Selector封装了NIO的Selector，内含一个NIO Selector对象。

Kafka Selector实现思路

1.从上图可以看出， Selector内部包含一个Map, 也就是它维护了所有连接的连接池。这些KafkaChannel都由ChannelBuilder接口创建。

    private final Map<String, KafkaChannel> channels;

2.所有的io操作：connect, read, write其实都是在poll这1个函数里面完成的。具体什么意思呢？

NetworkClient的send()函数，调用了selector.send(Send send)， 但这个时候数据并没有真的发送出去，只是暂存在了selector内部相对应的channel里面。下面看代码：

//Selector
    public void send(Send send) {
        KafkaChannel channel = channelOrFail(send.destination());  //找到数据包相对应的connection
        try {
            channel.setSend(send);  //暂存在这个connection(channel)里面
        } catch (CancelledKeyException e) {
            this.failedSends.add(send.destination());
            close(channel);
        }
    }

//KafkaChannel
    public void setSend(Send send) {
        if (this.send != null)  //关键点：当前的没有发出去之前，不能暂存下1个！！！关于这个，后面还要详细分析
            throw new IllegalStateException("Attempt to begin a send operation with prior send operation still in progress.");
        this.send = send;   //暂存这个数据包
        this.transportLayer.addInterestOps(SelectionKey.OP_WRITE);
    }

public class KafkaChannel {
    private final String id;
    private final TransportLayer transportLayer;
    private final Authenticator authenticator;
    private final int maxReceiveSize;
    private NetworkReceive receive;
    private Send send;   //关键点：1个channel一次只能存放1个数据包，在当前的send数据包没有完整发出去之前，不能存放下一个
    ...
}

暂存在channel中之后，poll函数进行处理，我们抽象出一个输入-输出模型如下：

输入：暂存的send数据包

输出：完成的sends, 完成的receive(针对上1次的send), 建立的连接， 断掉的连接。

    @Override
    public void poll(long timeout) throws IOException {
        if (timeout < 0)
            throw new IllegalArgumentException("timeout should be >= 0");
        clear();  //关键点：每次poll之前，会清空“输出”
        if (hasStagedReceives())
            timeout = 0;
        /* check ready keys */
        long startSelect = time.nanoseconds();
        int readyKeys = select(timeout);
        long endSelect = time.nanoseconds();
        currentTimeNanos = endSelect;
        this.sensors.selectTime.record(endSelect - startSelect, time.milliseconds());

        if (readyKeys > 0) {
            Set<SelectionKey> keys = this.nioSelector.selectedKeys();
            Iterator<SelectionKey> iter = keys.iterator();
            while (iter.hasNext()) {
                SelectionKey key = iter.next();
                iter.remove();
                KafkaChannel channel = channel(key);

                // register all per-connection metrics at once
                sensors.maybeRegisterConnectionMetrics(channel.id());
                lruConnections.put(channel.id(), currentTimeNanos);

                try {
                    /* complete any connections that have finished their handshake */
                    if (key.isConnectable()) {
                        channel.finishConnect();    //把建立的连接，加入输出结果集合
                        this.connected.add(channel.id());
                        this.sensors.connectionCreated.record();
                    }

                    ...

                    if (channel.ready() && key.isReadable() && !hasStagedReceive(channel)) {
                        NetworkReceive networkReceive;
                        while ((networkReceive = channel.read()) != null)
                            addToStagedReceives(channel, networkReceive);
                    }

                    if (channel.ready() && key.isWritable()) {
                        Send send = channel.write();
                        if (send != null) {
                            this.completedSends.add(send);  //把完成的发送，加入输出结果集合
                            this.sensors.recordBytesSent(channel.id(), send.size());
                        }
                    }

                    if (!key.isValid()) {
                        close(channel);
                        this.disconnected.add(channel.id());  //把断掉的连接，加入输出结果集合
                    }
                } catch (Exception e) {
                    String desc = channel.socketDescription();
                    if (e instanceof IOException)
                        log.debug("Connection with {} disconnected", desc, e);
                    else
                        log.warn("Unexpected error from {}; closing connection", desc, e);
                    close(channel);
                    this.disconnected.add(channel.id()); //把断掉的连接，加入输出结果集合
                }
            }
        }

        addToCompletedReceives(); //把完成的接收，加入输出结果集合

        long endIo = time.nanoseconds();
        this.sensors.ioTime.record(endIo - endSelect, time.milliseconds());
        maybeCloseOldestConnection();
    }

核心原理之1 – 消息的分包

在上面的代码中，为什么会有addToStagedReceives？ 什么叫做staged receives呢？ 这叫要从数据的分包说起：

在NetworkClient中，往下传的是一个完整的ClientRequest，进到Selector，暂存到channel中的，也是一个完整的Send对象（1个数据包）。但这个Send对象，交由底层的channel.write(Bytebuffer b)的时候，并不一定一次可以完全发送，可能要调用多次write，才能把一个Send对象完全发出去。这是因为write是非阻塞的，不是等到完全发出去，才会返回。所以才有上面的代码：

                    if (channel.ready() && key.isWritable()) {
                        Send send = channel.write(); //send不为空，表示完全发送出去，返回发出去的这个Send对象。如果没完全发出去，返回null
                        if (send != null) {
                            this.completedSends.add(send);
                            this.sensors.recordBytesSent(channel.id(), send.size());
                        }
                    }

同样，在接收的时候，channel.read(Bytebuffer b)，一个response也可能要read多次，才能完全接收。所以就有了上面的while循环代码：

                    if (channel.ready() && key.isReadable() && !hasStagedReceive(channel)) {
                        NetworkReceive networkReceive;
                        while ((networkReceive = channel.read()) != null)  //循环接收，直到1个response完全接收到，才会从while循环退出
                            addToStagedReceives(channel, networkReceive);
                    }

核心原理之2 – 消息的分界

从上面知道，底层数据的通信，是在每一个channel上面，2个源源不断的byte流，一个send流，一个receive流。

send的时候，还好说，发送之前知道一个完整的消息的大小；

那接收的时候，我怎么知道一个msg response什么时候结束，然后开始接收下一个response呢？

这就需要一个小技巧：在所有request，response头部，首先是一个定长的，4字节的头，receive的时候，至少调用2次read，先读取这4个字节，获取整个response的长度，接下来再读取消息体。

public class NetworkReceive implements Receive {
    private final String source;
    private final ByteBuffer size;  //头部4字节的buffer
    private final int maxSize;
    private ByteBuffer buffer;  //后面整个消息response的buffer

    public NetworkReceive(String source) {
        this.source = source;
        this.size = ByteBuffer.allocate(4);   //先分配4字节的头部
        this.buffer = null;
        this.maxSize = UNLIMITED;
   }
}

核心原理之3 － 消息时序保证

在InFlightRequests中，存放了所有发出去，但是response还没有回来的request。request发出去的时候，入对；response回来，就把相对应的request出对。

final class InFlightRequests {

    private final int maxInFlightRequestsPerConnection;
    private final Map<String, Deque<ClientRequest>> requests = new HashMap<String, Deque<ClientRequest>>();
｝

这个有个关键点：我们注意到request与response的配对，在这里是用队列表达的，而不是Map。用队列的入队，出队，完成2者的匹配。要实现这个，服务器就必须要保证消息的时序：即在一个socket上面，假如发出去的reqeust是0, 1, 2，那返回的response的顺序也必须是0, 1, 2。

但是服务器是1 + N + M模型，所有的请求进入一个requestQueue，然后是多线程并行处理的。那它如何保证消息的时序呢？

答案是mute/unmute机制：每当一个channel上面接收到一个request，这个channel就会被mute，然后等response返回之后，才会再unmute。这样就保证了同1个连接上面，同时只会有1个请求被处理。

下面是服务端的代码：

    selector.completedReceives.asScala.foreach { receive =>
          try {
            val channel = selector.channel(receive.source)
            val session = RequestChannel.Session(new KafkaPrincipal(KafkaPrincipal.USER_TYPE, channel.principal.getName),
              channel.socketAddress)
            val req = RequestChannel.Request(processor = id, connectionId = receive.source, session = session, buffer = receive.payload, startTimeMs = time.milliseconds, securityProtocol = protocol)
            requestChannel.sendRequest(req)
          } catch {
            case e @ (_: InvalidRequestException | _: SchemaException) =>
              // note that even though we got an exception, we can assume that receive.source is valid. Issues with constructing a valid receive object were handled earlier
              error("Closing socket for " + receive.source + " because of error", e)
              close(selector, receive.source)
          }
          selector.mute(receive.source)    //收到请求，把这个请求对应的channel, mute
        }

        selector.completedSends.asScala.foreach { send =>
          val resp = inflightResponses.remove(send.destination).getOrElse {
            throw new IllegalStateException(s"Send for ${send.destination} completed, but not in `inflightResponses`")
          }
          resp.request.updateRequestMetrics()
          selector.unmute(send.destination)  //发送response之后，把这个responese对应的channel, unmute
        }

NetworkClient实现思路

上面已经讲到：

（1）Selector维护了所有连接的连接池，所有连接上，消息的发送、接收都是通过poll函数进行的

（2）一个channel一次只能暂存1个Send对象。

但如果这个Send对象，一次poll之后，没有完全发送出去怎么办呢？看上层NetworkClient怎么处理的：

关键的client.ready函数

先从Sender的run()函数看起：

    public void run(long now) {
        Cluster cluster = metadata.fetch();
        // get the list of partitions with data ready to send
        RecordAccumulator.ReadyCheckResult result = this.accumulator.ready(cluster, now);

        if (result.unknownLeadersExist)
            this.metadata.requestUpdate();

        // remove any nodes we aren‘t ready to send to
        Iterator<Node> iter = result.readyNodes.iterator();
        long notReadyTimeout = Long.MAX_VALUE;
        while (iter.hasNext()) {
            Node node = iter.next();
            if (!this.client.ready(node, now)) {   //关键函数！！！
                iter.remove();
                notReadyTimeout = Math.min(notReadyTimeout, this.client.connectionDelay(node, now));
            }
        }

        // create produce requests
        Map<Integer, List<RecordBatch>> batches = this.accumulator.drain(cluster,
                                                                         result.readyNodes,
                                                                         this.maxRequestSize,
                                                                         now);

        List<RecordBatch> expiredBatches = this.accumulator.abortExpiredBatches(this.requestTimeout, cluster, now);
        // update sensors
        for (RecordBatch expiredBatch : expiredBatches)
            this.sensors.recordErrors(expiredBatch.topicPartition.topic(), expiredBatch.recordCount);

        sensors.updateProduceRequestMetrics(batches);
        List<ClientRequest> requests = createProduceRequests(batches, now);

        long pollTimeout = Math.min(result.nextReadyCheckDelayMs, notReadyTimeout);
        if (result.readyNodes.size() > 0) {
            log.trace("Nodes with data ready to send: {}", result.readyNodes);
            log.trace("Created {} produce requests: {}", requests.size(), requests);
            pollTimeout = 0;
        }

        for (ClientRequest request : requests)  //每个request分属于不同的Node
            client.send(request, now);   //client的send就是直接调用了selector.send，消息暂存在channel里面，没有发送

        this.client.poll(pollTimeout, now); //调用selector.poll，处理连接、发送、接收
    }

在上面的代码中，有一个关键函数：client.ready(Node n, ..)， 这个函数内部会判断这个node有没有ready，如果没有ready，就会从readNodes里面移除，接下来就不会往这个Node发送消息。

那什么叫ready呢？ 我们看一下代码：

    public boolean ready(Node node, long now) {
        if (isReady(node, now))
            return true;

        if (connectionStates.canConnect(node.idString(), now))
            initiateConnect(node, now);
        return false;
    }

    public boolean isReady(Node node, long now) {
        return !metadataUpdater.isUpdateDue(now) && canSendRequest(node.idString());
    }

    private boolean canSendRequest(String node) {
        return connectionStates.isConnected(node) && selector.isChannelReady(node) && inFlightRequests.canSendMore(node);
    }

    public boolean canSendMore(String node) {
        Deque<ClientRequest> queue = requests.get(node);
        return queue == null || queue.isEmpty() ||
               (queue.peekFirst().request().completed() && queue.size() < this.maxInFlightRequestsPerConnection);
    }

    public boolean completed() {
        return remaining <= 0 && !pending;
    }

上面的代码封了好几层，但总结下来，一个Node ready，可以向其发送请求，需要符合以下几个条件：

1. metadata正常，不需要update： !metadataUpdater.isUpdateDue(now)

2. 连接正常 connectionStates.isConnected(node)

3. channel是ready状态：这个对于PlaintextChannel， 一直返回true

4. 当前该channel中，没有in flight request，所有请求都处理完了

5. 当前该channel中，队列尾部的request已经完全发送出去, request.completed()，并且inflight request数目，没有超过设定的最大值（缺省为5，即允许在“天上飞”的request最多有5个，所谓在“天上飞”，就是发出去了，response还没有回来)

而上面的第5个条件，正是解决了上面的问题：一个channel里面的Send对象要是只发送了部分，下1次就不会处于ready状态了。

client.poll函数

下面看一下client.poll，是如何封装selector.poll的：

    public List<ClientResponse> poll(long timeout, long now) {
        long metadataTimeout = metadataUpdater.maybeUpdate(now);
        try {
            this.selector.poll(Utils.min(timeout, metadataTimeout, requestTimeoutMs));
        } catch (IOException e) {
            log.error("Unexpected error during I/O", e);
        }

        //上面说到，selector.poll函数，会把处理结果，放到一堆的状态变量里面（输出结果集），现在就是处理这堆输出结果的时候了。

        long updatedNow = this.time.milliseconds();
        List<ClientResponse> responses = new ArrayList<>();
        handleCompletedSends(responses, updatedNow);
        handleCompletedReceives(responses, updatedNow);
        handleDisconnections(responses, updatedNow);
        handleConnections();
        handleTimedOutRequests(responses, updatedNow);

        // invoke callbacks
        for (ClientResponse response : responses) {
            if (response.request().hasCallback()) {
                try {
                    response.request().callback().onComplete(response);
                } catch (Exception e) {
                    log.error("Uncaught error in request completion:", e);
                }
            }
        }

        return responses;
   }

//Selector中的那堆状态变量，在每次poll之前,被clear情况掉，每次poll之后，填充。
//然后在client.poll里面，这堆输出结果被处理
public class Selector implements Selectable {
    。。。
    private final List<Send> completedSends;
    private final List<NetworkReceive> completedReceives;
    private final Map<KafkaChannel, Deque<NetworkReceive>> stagedReceives;
    private final List<String> disconnected;
    private final List<String> connected;
。。。
}

连接检测 & 自动重连机制

在所有tcp长链接的编程中，都有一个基本问题要解决：如何判断1个连接是否断开？客户端需要维护所有连接的状态（connecting, connected, disconnected)，然后根据连接状态做不同逻辑。

但在NIO中，并没有一个函数，可以直接告诉你一个连接是否断开了；在NetworkClient里面，也并没有开一个线程，不断发送心跳消息，来检测连接。那它是如何处理的呢？

检测连接断开的手段

在networkClient的实现中，用了3种手段，来判断一个连接是否断开：

手段1：所有的IO函数，connect, finishConnect, read, write都会抛IOException，因此任何时候，调用这些函数的时候，只要抛异常，就认为连接已经断开。

手段2：selectionKey.isValid()

手段3：inflightRequests，所有发出去的request，都设置有一个response返回的时间。在这个时间内，response没有回来，就认为连接断了。

前2种手段，都集中在Select.poll函数里面:

    public void poll(long timeout) throws IOException {
        if (timeout < 0)
            throw new IllegalArgumentException("timeout should be >= 0");
        clear();
        if (hasStagedReceives())
            timeout = 0;
        /* check ready keys */
        long startSelect = time.nanoseconds();
        int readyKeys = select(timeout);
        long endSelect = time.nanoseconds();
        currentTimeNanos = endSelect;
        this.sensors.selectTime.record(endSelect - startSelect, time.milliseconds());

        if (readyKeys > 0) {
            Set<SelectionKey> keys = this.nioSelector.selectedKeys();
            Iterator<SelectionKey> iter = keys.iterator();
            while (iter.hasNext()) {
                SelectionKey key = iter.next();
                iter.remove();
                KafkaChannel channel = channel(key);

                // register all per-connection metrics at once
                sensors.maybeRegisterConnectionMetrics(channel.id());
                lruConnections.put(channel.id(), currentTimeNanos);

                try {
                    /* complete any connections that have finished their handshake */
                    if (key.isConnectable()) {
                        channel.finishConnect();
                        this.connected.add(channel.id());
                        this.sensors.connectionCreated.record();
                    }

                    /* 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;
                        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 */
                    if (channel.ready() && key.isWritable()) {
                        Send send = channel.write();
                        if (send != null) {
                            this.completedSends.add(send);
                            this.sensors.recordBytesSent(channel.id(), send.size());
                        }
                    }

                    if (!key.isValid()) {   //手段2
                        close(channel);
                        this.disconnected.add(channel.id());
                    }
                } catch (Exception e) {  //手段1：任何一个io函数，只要抛错，就认为连接断了
                    String desc = channel.socketDescription();
                    if (e instanceof IOException)
                        log.debug("Connection with {} disconnected", desc, e);
                    else
                        log.warn("Unexpected error from {}; closing connection", desc, e);
                    close(channel);
                    this.disconnected.add(channel.id());
                }
            }
        }

        addToCompletedReceives();

        long endIo = time.nanoseconds();
        this.sensors.ioTime.record(endIo - endSelect, time.milliseconds());
        maybeCloseOldestConnection();
    }

第3种手段，在NetworkClient里面：

    public List<ClientResponse> poll(long timeout, long now) {
        long metadataTimeout = metadataUpdater.maybeUpdate(now);
        try {
            this.selector.poll(Utils.min(timeout, metadataTimeout, requestTimeoutMs));
        } catch (IOException e) {
            log.error("Unexpected error during I/O", e);
        }

        long updatedNow = this.time.milliseconds();
        List<ClientResponse> responses = new ArrayList<>();
        handleCompletedSends(responses, updatedNow);
        handleCompletedReceives(responses, updatedNow);
        handleDisconnections(responses, updatedNow);
        handleConnections();
        handleTimedOutRequests(responses, updatedNow); //手段3：处理所有TimeOutRequests

        for (ClientResponse response : responses) {
            if (response.request().hasCallback()) {
                try {
                    response.request().callback().onComplete(response);
                } catch (Exception e) {
                    log.error("Uncaught error in request completion:", e);
                }
            }
        }

        return responses;
    }

    private void processDisconnection(List<ClientResponse> responses, String nodeId, long now) {
        connectionStates.disconnected(nodeId, now);
        for (ClientRequest request : this.inFlightRequests.clearAll(nodeId)) {
            log.trace("Cancelled request {} due to node {} being disconnected", request, nodeId);
            if (!metadataUpdater.maybeHandleDisconnection(request)) //把MetaDataRequest排除在外，其它所有请求，只要超时，就认为连接断开
                responses.add(new ClientResponse(request, now, true, null));
        }
    }

除了上述的2个地方，还要一个地方，就是初始化的时候

    private void initiateConnect(Node node, long now) {
        String nodeConnectionId = node.idString();
        try {
            log.debug("Initiating connection to node {} at {}:{}.", node.id(), node.host(), node.port());
            this.connectionStates.connecting(nodeConnectionId, now);
            selector.connect(nodeConnectionId,
                             new InetSocketAddress(node.host(), node.port()),
                             this.socketSendBuffer,
                             this.socketReceiveBuffer);
        } catch (IOException e) { //检测到连接断开
            connectionStates.disconnected(nodeConnectionId, now);
            metadataUpdater.requestUpdate();
            log.debug("Error connecting to node {} at {}:{}:", node.id(), node.host(), node.port(), e);
        }
    }

检测时机

从上面代码我们可以看出，连接的检测时机，有2个：

一个是初始建立连接的时候，一个就是每次poll循环，每poll一次，就收集到一个断开的连接集合。

下面分别是Selector和NetworkClient中，关于连接状态的数据结构：

//Selector中的连接状态
public class Selector implements Selectable {
    private final List<String> disconnected;
    private final List<String> connected;
    ..
}
//NetworkClient中的连接状态维护
public class NetworkClient implements KafkaClient {
    private final ClusterConnectionStates connectionStates;
    ...
}

final class ClusterConnectionStates {
    private final long reconnectBackoffMs; //重连的时间间隔
    private final Map<String, NodeConnectionState> nodeState;
}

    private static class NodeConnectionState {
        ConnectionState state;
        long lastConnectAttemptMs;  //上1次发起重连的时间
        ...
    }

public enum ConnectionState {
    DISCONNECTED, CONNECTING, CONNECTED
}

总结：

1. Selector中的连接状态，在每次poll之前，会调用clear清空；在poll之后，收集。

2. Selector中的连接状态，会传给上层NetworkClient，用于它更新自己的连接状态

3. 出了来自Selctor，NetworkClient自己内部的inflightRequests，也就是上面的手段3， 也用于检测连接状态。

通过上面的机制，就保证了NetworkClient可以实时准确维护所有connection的状态。

自动重连 － ready函数

状态知道了，那剩下的就是自动重连了。这个发生在更上层的Send的run函数里面：

//Sender
    public void run(long now) {
        Cluster cluster = metadata.fetch();
        RecordAccumulator.ReadyCheckResult result = this.accumulator.ready(cluster, now);

        if (result.unknownLeadersExist)
            this.metadata.requestUpdate();

        Iterator<Node> iter = result.readyNodes.iterator();
        long notReadyTimeout = Long.MAX_VALUE;
        while (iter.hasNext()) {
            Node node = iter.next();
            if (!this.client.ready(node, now)) {  //关键的ready函数
                iter.remove();
                notReadyTimeout = Math.min(notReadyTimeout, this.client.connectionDelay(node, now));
            }
        }

    public boolean ready(Node node, long now) {
        if (isReady(node, now))
            return true;

        if (connectionStates.canConnect(node.idString(), now))
            initiateConnect(node, now);   //发起重连

        return false;
    }

    public boolean canConnect(String id, long now) {
        NodeConnectionState state = nodeState.get(id);
        if (state == null)
            return true;
        else
            return state.state == ConnectionState.DISCONNECTED && now - state.lastConnectAttemptMs >= this.reconnectBackoffMs;
    }

从上面函数可以看出，每次Send发数据之前，会先调用client.ready(node)判断该node的连接是否可用。

在ready内部，如果连接不是connected状态，会再判断是否可以发起自动重连，检测条件有2个：

条件1： 它不能是connecting状态，必须是disconnected

条件2： 重连不能太频繁。当前时间距离上1次重连时间，要有一定的间隔。如果broker挂了，你太频繁的重连也不起作用。

这里有个关键点：因为都是非阻塞调用，本次虽然检测到连接断了，但只是发起连接，不会等到连接建立好了，再执行下面的代码。

会在poll之后，判断连接是否建立；在下1次或者下几次poll之前，可能连接才会建立好，ready才会返回true.

时间： 2024-08-12 13:04:02

Kafka源码分析－序列4 －Producer －network层核心原理的相关文章

apache kafka源码分析走读-Producer分析

apache kafka中国社区QQ群:162272557 producer的发送方式剖析 Kafka提供了Producer类作为java producer的api,该类有sync和async两种发送方式. sync架构图 async架构图 调用流程如下: 代码流程如下: Producer:当new Producer(new ProducerConfig()),其底层实现,实际会产生两个核心类的实例:Producer.DefaultEventHandler.在创建的同时,会默认new一个Prod

ZRender源码分析3：Painter(View层)-上

回顾 上一篇说到:ZRender源码分析2:Storage(Model层),这次咱看来看看Painter-View层 总体理解 Painter这个类主要负责MVC中的V(View)层,负责将Storage中的shape对象绘制到canvas中,包括了:更新.渲染.变化大小.导出.修改等操作. Painter这个类还是很明显的构造函数,然后把方法赋值到Painter.prototype上,无新奇之处,下面为示例代码.只有在Painter.js末尾有一个内部的createDom函数, 很明显,传入i

ZRender源码分析4：Painter(View层)-中

回顾 上一篇说到:ZRender源码分析3:Painter(View层)-上,接上篇,开始Shape对象 总体理解 先回到上次的Painter的render方法 /** * 首次绘图,创建各种dom和context * 核心方法,zr.render() --> painter.render * * render和refersh的区别:render是clear所有,refresh是清除已经改变的layer * * @param {Function=} callback 绘画结束后的回调函数 */

SpringMVC源码分析6：SpringMVC的视图解析原理

title: SpringMVC源码分析6:SpringMVC的视图解析原理 date: 2018-06-07 11:03:19 tags: - SpringMVC categories: - 后端 - SpringMVC --- 转自 SpringMVC视图机制详解[附带源码分析] 本系列文章首发于我的个人博客:https://h2pl.github.io/ 欢迎阅览我的CSDN专栏:Spring源码解析 https://blog.csdn.net/column/details/21851.h

apache kafka源码分析-Producer分析---转载

原文地址:http://www.aboutyun.com/thread-9938-1-1.html 问题导读1.Kafka提供了Producer类作为java producer的api,此类有几种发送方式?2.总结调用producer.send方法包含哪些流程?3.Producer难以理解的在什么地方? producer的发送方式剖析Kafka提供了Producer类作为java producer的api,该类有sync和async两种发送方式.sync架构图 async架构图 调用流程如下:

Kafka源码分析及图解原理之Producer端

一.前言 任何消息队列都是万变不离其宗都是3部分,消息生产者(Producer).消息消费者(Consumer)和服务载体(在Kafka中用Broker指代).那么本篇主要讲解Producer端,会有适当的图解帮助理解底层原理. 一.开发应用 首先介绍一下开发应用,如何构建一个KafkaProducer及使用,还有一些重要参数的简介. 1.1 一个栗子 1 /** 2 * Kafka Producer Demo实例类. 3 * 4 * @author GrimMjx 5 */ 6 public

kafka源码分析之一server启动分析

1. 分析kafka源码的目的 深入掌握kafka的内部原理 深入掌握scala运用 2. server的启动 如下所示(本来准备用时序图的,但感觉时序图没有思维图更能反映,故采用了思维图): 2.1 启动入口Kafka.scala 从上面的思维导图,可以看到Kafka的启动入口是Kafka.scala的main()函数: def main(args: Array[String]): Unit = { try { val serverProps = getPropsFromArgs(args)

ZRender源码分析2：Storage(Model层)

回顾 上一篇请移步:zrender源码分析1:总体结构 本篇进行ZRender的MVC结构中的M进行分析 总体理解 上篇说到,Storage负责MVC层中的Model,也就是模型,对于zrender来说,这个model就是shape对象,在1.x表现的还不强烈,到了2.x, 在zr.addShape()的时候,传入的参数就必须是new出来的对象了详情请看这里 2.x相比1.x的变化,关于这个变化多说点吧,那就是从1.x升级到2.x的时候,因为方式变了,总不能改掉所有的代码,总不能像ext一样,

Kafka源码分析－序列2 －Producer －Metadata的数据结构与读取、更新策略

在上一篇,我们从使用方式和策略上,对消息队列做了一个宏观描述.从本篇开始,我们将深入到源码内部,仔细分析Kafka到底是如何实现一个分布式消息队列.我们的分析将从Producer端开始. 从Kafka 0.8.2开始,发布了一套新的Java版的client api, KafkaProducer/KafkaConsumer,替代之前的scala版的api.本系列的分析将只针对这套Java版的api. 多线程异步发送模型 下图是经过源码分析之后,整理出来的Producer端的架构图: 在上一篇我们讲