Kafka 主函数在kafka.Kafka.scala 如下：

      val serverConfig = new KafkaConfig(props)      KafkaMetricsReporter.startReporters(serverConfig.props)      val kafkaServerStartble = new KafkaServerStartable(serverConfig)            kafkaServerStartble.startup

读取命令行中第一个参数作为配置文件

bin/zookeeper-server-start.sh config/zookeeper.properties

kafka.utils.Utils从中读取配置项 生成props 作为KafkaConfig参数

KafkaConfig构造函数还是已VerifiableProperties作为参数

KafkaMetricsReporter调用startReporters(props) 

class KafkaServerStartable(val serverConfig: KafkaConfig) extends Logging {  private var server: KafkaServer = null  def startup() {      server.startup()  }}

KafkaServerStartable看上去就是一个KafkaServer ……  直接看kafka.server.KafkaServer吧

    /* start scheduler */    kafkaScheduler.startup()    /* setup zookeeper */    zkClient = initZk()    /* start log manager */    logManager = createLogManager(zkClient)    logManager.startup()    socketServer = new SocketServer(config.brokerId,      config.hostName,      config.port,      config.numNetworkThreads,      config.queuedMaxRequests,      config.socketSendBufferBytes,      config.socketReceiveBufferBytes,      config.socketRequestMaxBytes)    socketServer.startup()    replicaManager = new ReplicaManager(config,       time,       zkClient,       kafkaScheduler,       logManager,       isShuttingDown)    kafkaController = new KafkaController(config, zkClient)    /* start processing requests */    apis = new KafkaApis(socketServer.requestChannel,       replicaManager,       zkClient,       config.brokerId,       config, kafkaController)          requestHandlerPool = new KafkaRequestHandlerPool(config.brokerId,       socketServer.requestChannel,       apis,       config.numIoThreads)    Mx4jLoader.maybeLoad()    replicaManager.startup()    kafkaController.startup()    topicConfigManager = new TopicConfigManager(zkClient, logManager)    topicConfigManager.startup()    /* tell everyone we are alive */    kafkaHealthcheck = new KafkaHealthcheck(config.brokerId,       config.advertisedHostName,       config.advertisedPort,       config.zkSessionTimeoutMs,       zkClient)          kafkaHealthcheck.startup()    registerStats()    startupComplete.set(true);

读kafka源码到了这里，突然间有一种豁然开朗的感觉，

在这里先后启动了

KafkaScheduler、LogManager、SocketServer、ReplicaManager、KakfaController、KafkaApi ....

好戏~ 就此上演 待我~ 细说从头