kafka broker启动

通过bin下面的kafka-server-start.sh 可以启动一个broker，命令如下：nohup ./kafka-server-start.sh config/server.properties &
这个脚本主要是调用kafka.kafka 这个类
exec $base_dir/kafka-run-class.sh $EXTRA_ARGS kafka.Kafka "$@"
与之对应的是停止一个broker对应的脚本是kafka-server-stop.sh
SIGNAL=${SIGNAL:-TERM}
PIDS=$(ps ax | grep -i 'kafka\.Kafka' | grep java | grep -v grep | awk '{print $1}')if [ -z "$PIDS" ]; thenecho "No kafka server to stop"exit 1
elsekill -s $SIGNAL $PIDS
fi
这个脚本就是通过ps ax 查找到kafka的进程后，然后通过kill 命令将这个进程杀死
我们首先看看kafka这个类
core/src/main/scala/kafka/Kafka.scaladef main(args: Array[String]): Unit = {try {val serverProps = getPropsFromArgs(args)val kafkaServerStartable = KafkaServerStartable.fromProps(serverProps)try {if (!OperatingSystem.IS_WINDOWS && !Java.isIbmJdk)new LoggingSignalHandler().register()} catch {case e: ReflectiveOperationException =>warn("Failed to register optional signal handler that logs a message when the process is terminated " +s"by a signal. Reason for registration failure is: $e", e)}// attach shutdown handler to catch terminating signals as well as normal terminationRuntime.getRuntime().addShutdownHook(new Thread("kafka-shutdown-hook") {override def run(): Unit = kafkaServerStartable.shutdown()})
#这里调用kafkaServerStartable 的startup函数来启动brokerkafkaServerStartable.startup()kafkaServerStartable.awaitShutdown()}KafkaServerStartable 这个类的starup和shutdonw函数如下：
\core\src\main\scala\kafka\server\KafkaServerStartable.scalaclass KafkaServerStartable(val staticServerConfig: KafkaConfig, reporters: Seq[KafkaMetricsReporter]) extends Logging {private val server = new KafkaServer(staticServerConfig, kafkaMetricsReporters = reporters)def this(serverConfig: KafkaConfig) = this(serverConfig, Seq.empty)def startup() {
#这里有调用KafkaServer的startup函数来启动brokertry server.startup()catch {case _: Throwable =>// KafkaServer.startup() calls shutdown() in case of exceptions, so we invoke `exit` to set the status codefatal("Exiting Kafka.")Exit.exit(1)}}def shutdown() {try server.shutdown()catch {case _: Throwable =>fatal("Halting Kafka.")// Calling exit() can lead to deadlock as exit() can be called multiple times. Force exit.Exit.halt(1)}}KafkaServer.scala 的startup函数如下：
core\src\main\scala\kafka\server\KafkaServer.scaladef startup() {try {info("starting")quotaManagers = QuotaFactory.instantiate(config, metrics, time, threadNamePrefix.getOrElse(""))notifyClusterListeners(kafkaMetricsReporters ++ metrics.reporters.asScala)logDirFailureChannel = new LogDirFailureChannel(config.logDirs.size)
这个类会启动很多manager，下面这行就可以看到logManager 是从kafkaserver中启动的/* start log manager */logManager = LogManager(config, initialOfflineDirs, zkClient, brokerState, kafkaScheduler, time, brokerTopicStats, logDirFailureChannel)logManager.startup()
}

kafka broker启动相关推荐

4.2.10 Kafka源码剖析, 阅读环境搭建, broker启动流程, topic创建流程, Producer生产者流程, Consumer消费者流程,
目录 4.1 Kafka源码剖析之源码阅读环境搭建 4.1.1 安装配置Gradle 4.1.2 Scala的安装和配置 4.1.3 Idea配置 4.1.4 源码操作 4.2 Kafka源码剖析之B ...
Kafka broker配置介绍（四）
这部分内容对了解系统和提高软件性能都有很大的帮助,kafka官网上也给出了比较详细的配置详单,但是我们还是直接从代码来看broker到底有哪些配置需要我们去了解的,配置都有英文注释,所以每一部分是干什 ...
【Kafka】Kafka broker id 解释以及自动生成id
1.概述转载:https://honeypps.com/mq/kafka-params-analysis-of-broker-id/ 2.Kafka参数broker.id详解 kafka在启动服务之 ...
kafka的启动失败
在学习搭建单机kafka的时候遇到kafka不能启动的错误. 在zookeeper可以正常running的前提下,启动kafka: bin/kafka-server-start.sh config/s ...
Kafka Broker 工作流程
Zookeeper存储的Kafka的信息查看zookeeper中的kafka节点所存储的信息启动Zookeeper客户端 [atguigu@hadoop104 zookeeper-3.5.7]$ ...
CDH安装kafka组件启动失败
CDH安装kafka组件启动失败提示"Configured broker.id 465 doesn't match stored broker.id 274 in meta.properti ...
RocketMQ源码：Broker启动过程介绍
1. 简介 Broker 启动入口是 BrokerStartup.java => main() ,跟上一篇讲的 NameSrv 类似,首先会从启动命令判断 -c 参数,读取自定义配置文件.然后创 ...
kafka一键启动集群
我的kafka的3个节点是这样的. 启动kafka集群脚本 #!/bin/bash echo "kafka cluster start" ssh Desktop "sou ...
【kafka】kafka broker 限流 topic 限流配额
纪念一波,九师兄博客热门订阅专栏时常名列前茅,我飘了,哈哈哈哈,得意的笑 1. 概述翻译:配额 4.9 配额 Kafka集群有能力对请求执行配额来控制客户端使用的代理资源.Kafka broker可 ...

kafka broker启动

kafka broker启动相关推荐

最新文章

热门文章