从零（抄）写Kafka（一）启动一个Broker

前言

​ Kafka是业界很火的一个消息队列，相信很多Java工程师都用过它，我们公司也是有相应的Kafka集群提供数据库binlog订阅以及日志同步等服务，当然他的应用场景有很多。正好前阶段时间从源码级别学习了一下Kafka的基本实现原理。但是只看视频学习的话还是有点模棱两可，理解的不透彻，于是乎冒出了照着源码敲一敲核心代码的想法。我把它称之为乞丐版Kafka,其中核心代码都是一样的,只是省去了大量的监控和统计代码,以及简化了配置项和一些功能项,例如连接断开,异常处理等.另外由于服务端是Scala的，我又把它翻译成了Java版本。闲言少叙，书归正传，本篇是此系列的第一篇，主要会讲述服务端broker的网络模型和实现。

网络模型

​ Kafka网络底层是采用了java的NIO来开发的，没有使用业界较为流行的Netty框架.它的服务端网络模型如下：

• Acceptor 网络处理入口，主要负责处理且仅处理 OP_ACCEPT事件，当有新的客户端想要连接时，Acceptor把它交给Processor去处理，多个客户端接入时采用轮询算法交给Processor。
• Processor 主要做网络事件的通用处理逻辑，通过不断的轮询去处理不同状态的客户端，例如连接成功，消息接收，消息发送，连接断开等。
• HandlerPool 业务处理线程池，由多个Handler组成,Processor不会进行业务处理，会将相应的消息组装之后放入队列，然后Handler线程拉取队列中的消息进行具体的业务处理。这里的业务就很多了，例如生产消息（ProduceRequest）、元信息拉取消息 （MetadataRequest）等等多达几十种，但是从抄代码的角度来讲，没必要都去实现，只要知道一条网络消息的处理流程即可。

核心代码

Acceptor

可以看到Acceptor代码较为简单，就是开启一个网络服务，比如我们开启本地的9092端口，然后等待客户端连接即可。

public Acceptor(EndPoint endPoint,
                    int sendBufferSize,
                    int receiveBufferSize,
                    int brokerId,
                    List<Processor> processors) {
        /**省略属性赋值代码*/
        try {
            //初始化一个Nio Selector 
            nioSelector = Selector.open();
            //打开本地的ServerSocket
            serverSocketChannel = openServerSocket(endPoint.getHost(), endPoint.getPort());
            synchronized (this) {
                //遍历所有的processor，启动processor线程
                for (Processor processor : processors) {
                    String threadName = String.format("kafka-network-thread-%d-%s-%d",
                            brokerId,
                            endPoint.getProtocol().toString(),
                            processor.getId());
                    logger.info("Start processor thread:{}",threadName);
                    //启动processor线程
                    Utils.newThread(threadName, processor, false).start();
                }
            }
        } catch (IOException e) {
            logger.error("Selector.open error", e);
        }
    }

OpenServerSocket代码很简单,不在赘述。

private ServerSocketChannel openServerSocket(String host,int port) throws IOException {
        InetSocketAddress address = (host == null || host.trim().isEmpty()) ? new InetSocketAddress(port)
                : new InetSocketAddress(host, port);

        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.configureBlocking(false);
        serverSocketChannel.socket().setReceiveBufferSize(receiveBufferSize);

        serverSocketChannel.bind(address);
        logger.info("Awaiting socket connections on {}:{}", address.getHostString(), serverSocketChannel.socket().getLocalPort());
        return serverSocketChannel;
    }

Acceptor是继承自Runnable的，run 方法是它的核心，主要是用来监听SelectionKey.OP_ACCEPT事件，然后将带有这个事件的连接交给Processor处理。

  @Override
    public void run() {
        try {
            //注册 OP_ACCEPT 事件
            serverSocketChannel.register(nioSelector, SelectionKey.OP_ACCEPT);
        } catch (ClosedChannelException e) {
            logger.error("Channel closed");
        }
        startupComplete();
        //轮询processor进行处理
        try {
            int currentProcessorIndex = 0;
            while (isRunning()) {
                    //调用底层select
                    int ready = nioSelector.select(500);
                    if (ready > 0) {
                        //到这里说明有客户端尝试连接服务器了
                        Set<SelectionKey> keys = nioSelector.selectedKeys();
                        Iterator<SelectionKey> iterator = keys.iterator();
                        while (iterator.hasNext() && isRunning()) {
                            try {
                                //拿到SelectionKey
                                SelectionKey key = iterator.next();
                                iterator.remove();
                                
                                //这里的Key只能是Acceptable
                                if (key.isAcceptable()) {
                                    accept(key, processors.get(currentProcessorIndex));
                                } else {
                                    throw new IllegalStateException("Unrecognized key state for acceptor thread.");
                                }
                                //轮询下一个processor处理
                                currentProcessorIndex = (currentProcessorIndex + 1) % processors.size();
                            } catch (Exception e) {
                                logger.error("Error while accepting connection", e);
                            }
                        }
                    }
            }
    }

然后就将SelectionKey交给Processor处理

private void accept(SelectionKey key,Processor processor) throws IOException {
        ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();
        //调用accept方法获取到SocketChannel
        SocketChannel socketChannel = serverSocketChannel.accept();
        /**忽略中间的一些配置*/
        processor.accept(socketChannel);
    }

Processor

Processor的职责是做一些网络公共业务处理，维护新的连接，执行poll，接收网络消息，处理之后发送到接收队列。同理业务Handler处理消息之后也发送到响应队列。Processor的run方法就是不断的监听消各个消息和不同状态类型的数据变化来进行相应的处理。

处理新的连接

当一个客户端连接服务器成功之后，往往就要开始发送数据了，所以Processor要做的第一步就是将新加入的连接注册OP_READ事件，保证下次循环在进行 select.poll 的时候能够及时读取数据。

poll

每次循环的时候poll过程是少不了的，不过它是调用了Selector的poll方法,这里的Selector是Kafka封装的一个类,后边会详细介绍.

处理接收到的请求包

将接受完成的请求包简单包装然后发送到请求队列中，后续交给HandlerPool中的Handler执行具体的业务处理逻辑

处理响应包

Handler处理完业务逻辑之后，会将响应发送到响应队列，此时Processor从队列里获取到响应之后会组装消息返回给客户端

处理已经发送成功的响应

对于已经响应客户端的连接，就可以继续注册OP_READ事件进行数据读取了

处理断开的连接

连接断开之后，维护一些连接状态

核心代码如下：

 while (isRunning()) {
            try {
                // 初始化所有的连接，注册OP_READ事件，准备读数据
                configureNewConnections();
                // 处理响应，把响应放入每个Processor响应队列列里，然后发送给客户端
                processNewResponses();
                //轮询处理请求，Selector监听各个SocketChannel，是否有请求发送过来
                poll();
                //处理已经接收完的消息
                processCompletedReceives();
                //处理发送完的消息
                processCompletedSends();
                //处理断开的连接
                processDisconnected();
            } catch (Exception e) {
                logger.error("");
            }
        }

启动日志

[main] INFO c.f.a.network.SocketServer - SocketServer starting...
[main] INFO c.f.a.network.SocketServer - Initializing 0 processor
[main] INFO c.f.a.network.SocketServer - Initializing 1 processor
[main] INFO c.f.a.network.SocketServer - Initializing 2 processor
[main] INFO c.f.a.network.Acceptor - Awaiting socket connections on 127.0.0.1:9092
[main] INFO c.f.a.network.Acceptor - Start processor thread:kafka-network-thread-0-PLAINTEXT-0
[main] INFO c.f.a.network.Acceptor - Start processor thread:kafka-network-thread-0-PLAINTEXT-1
[main] INFO c.f.a.network.Acceptor - Start processor thread:kafka-network-thread-0-PLAINTEXT-2
[main] INFO c.f.a.network.SocketServer - Start acceptor thread:kafka-socket-acceptor-PLAINTEXT-9092
[main] INFO c.f.a.server.KafkaRequestHandler - Start handler thread:kafka-request-handler-0
[main] INFO c.f.a.server.KafkaRequestHandler - Start handler thread:kafka-request-handler-1
[main] INFO c.f.a.server.KafkaRequestHandler - Start handler thread:kafka-request-handler-2
[main] INFO c.f.a.server.KafkaServer - Kafka Server started

这样一个简单的Kafka服务端就启动了，从日志上来看，主要是网络相关的程序启动，此时还没有加日志管理，副本分片管理以及Zookeeper等。

总结

​ 本篇主要是根据Kafka的服务端网络模型搭建出了一个简单的broker服务。源码版本是基于0.10.0.1，服务端是Scala语言，博主照猫画虎翻译成了Java语言

核心知识点：

• java.nio 编程
• 多线程处理
• reactor 网络模型

本文相关代码地址

下篇预告：从零（抄）写Kafka（二）深入理解Kafka的核心组件KafkaChannel、Selector