高效实用Kafka-深入理解Kafka启动配置(使用kafka自身内置Zookeeper)

导语
在上一篇博客中简单的介绍了关于消息系统消息中间件技术以及Kafka的一些基础。这篇文章主要是来介绍关于Kafka的架构以及如何安装Kafka。

首先先来安装Kafka，对于Kafka的安装来说可以使用自己的Zookeeper也可以使用Kafka自带的Zookeeper。下面就来演示如何使用kafka自带的ZK来进行启动测试。

第一步、首先获取Kafka的压缩包

http://kafka.apache.org/ 进入到Kafka的官网。找到下载选项

进行下载，当然也可以使用命令行的方式进行下载

wget http://labfile.oss.aliyuncs.com/courses/859/kafka_2.10-0.10.2.1.tgz

使用tar命令加压缩

首先先来查看一下关于Kafka的目录结构

bin：这个目录主要就是kafka的一些启动脚本
config：存放一些kafka的配置信息
lib：kafka启动的一些类库

第二步、启动kafka

在上面介绍bin目录的时候知道，在bin目录下主要放置的就是关于kafka的启动脚本。在启动kafka的时候应该遵循什么样的启动顺序呢？

根据下面的架构图可以知道，在kafka中其实一直是很依赖ZooKeeper的，所以说在kafka中自带了一个zk。那么第一步就应该启动的kafka中的zk。

会看到在bin目录下存在一个 zookeeper-server-start.sh 的启动脚本。执行启动脚本之后会发现，并没有启动而是给出了一个使用提示。这个是指向了一个配置文件，那么接下来就是要去查看这个配置文件。

进入到Kafka的配置文件路径下面，会看到确实是存在kafka的关于zookeeper的配置文件存在。那么就来看一下这个配置文件都有那些内容。

查看zookeeper.properties 配置文件会发现其中有三个配置参数分别是数据存放的目录、端口号、最大客户端链接数。

可以加载配置文件进行启动

root@nihui-PC:~/kafka_2.10-0.10.2.1/bin# ./zookeeper-server-start.sh ../config/zookeeper.properties
Picked up _JAVA_OPTIONS:   -Dawt.useSystemAAFontSettings=gasp
[2019-08-10 11:45:18,810] INFO Reading configuration from: ../config/zookeeper.properties (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2019-08-10 11:45:18,821] INFO autopurge.snapRetainCount set to 3 (org.apache.zookeeper.server.DatadirCleanupManager)
[2019-08-10 11:45:18,821] INFO autopurge.purgeInterval set to 0 (org.apache.zookeeper.server.DatadirCleanupManager)
[2019-08-10 11:45:18,821] INFO Purge task is not scheduled. (org.apache.zookeeper.server.DatadirCleanupManager)
[2019-08-10 11:45:18,821] WARN Either no config or no quorum defined in config, running  in standalone mode (org.apache.zookeeper.server.quorum.QuorumPeerMain)
[2019-08-10 11:45:18,841] INFO Reading configuration from: ../config/zookeeper.properties (org.apache.zookeeper.server.quorum.QuorumPeerConfig)
[2019-08-10 11:45:18,842] INFO Starting server (org.apache.zookeeper.server.ZooKeeperServerMain)
[2019-08-10 11:45:18,853] INFO Server environment:zookeeper.version=3.4.9-1757313, built on 08/23/2016 06:50 GMT (org.apache.zookeeper.server.ZooKeeperServer)
[2019-08-10 11:45:18,853] INFO Server environment:host.name=nihui-PC (org.apache.zookeeper.server.ZooKeeperServer)
[2019-08-10 11:45:18,853] INFO Server environment:java.version=1.8.0_161 (org.apache.zookeeper.server.ZooKeeperServer)
[2019-08-10 11:45:18,853] INFO Server environment:java.vendor=Oracle Corporation (org.apache.zookeeper.server.ZooKeeperServer)
[2019-08-10 11:45:18,853] INFO Server environment:java.home=/root/jdk1.8.0_161/jre (org.apache.zookeeper.server.ZooKeeperServer)
[2019-08-10 11:45:18,853] INFO Server environment:java.class.path=.:/root/jdk1.8.0_161/lib/dt.jar:/root/jdk1.8.0_161/lib/tools.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/aopalliance-repackaged-2.5.0-b05.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/argparse4j-0.7.0.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/connect-api-0.10.2.1.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/connect-file-0.10.2.1.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/connect-json-0.10.2.1.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/connect-runtime-0.10.2.1.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/connect-transforms-0.10.2.1.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/guava-18.0.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/hk2-api-2.5.0-b05.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/hk2-locator-2.5.0-b05.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/hk2-utils-2.5.0-b05.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jackson-annotations-2.8.0.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jackson-annotations-2.8.5.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jackson-core-2.8.5.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jackson-databind-2.8.5.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jackson-jaxrs-base-2.8.5.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jackson-jaxrs-json-provider-2.8.5.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jackson-module-jaxb-annotations-2.8.5.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/javassist-3.20.0-GA.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/javax.annotation-api-1.2.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/javax.inject-1.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/javax.inject-2.5.0-b05.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/javax.servlet-api-3.1.0.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/javax.ws.rs-api-2.0.1.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jersey-client-2.24.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jersey-common-2.24.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jersey-container-servlet-2.24.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jersey-container-servlet-core-2.24.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jersey-guava-2.24.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jersey-media-jaxb-2.24.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jersey-server-2.24.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jetty-continuation-9.2.15.v20160210.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jetty-http-9.2.15.v20160210.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jetty-io-9.2.15.v20160210.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jetty-security-9.2.15.v20160210.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jetty-server-9.2.15.v20160210.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jetty-servlet-9.2.15.v20160210.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jetty-servlets-9.2.15.v20160210.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jetty-util-9.2.15.v20160210.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jopt-simple-5.0.3.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/kafka_2.10-0.10.2.1.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/kafka_2.10-0.10.2.1-sources.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/kafka_2.10-0.10.2.1-test-sources.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/kafka-clients-0.10.2.1.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/kafka-log4j-appender-0.10.2.1.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/kafka-streams-0.10.2.1.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/kafka-streams-examples-0.10.2.1.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/kafka-tools-0.10.2.1.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/log4j-1.2.17.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/lz4-1.3.0.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/metrics-core-2.2.0.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/osgi-resource-locator-1.0.1.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/reflections-0.9.10.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/rocksdbjni-5.0.1.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/scala-library-2.10.6.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/slf4j-api-1.7.21.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/slf4j-log4j12-1.7.21.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/snappy-java-1.1.2.6.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/validation-api-1.1.0.Final.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/zkclient-0.10.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/zookeeper-3.4.9.jar (org.apache.zookeeper.server.ZooKeeperServer)
[2019-08-10 11:45:18,853] INFO Server environment:java.library.path=/usr/java/packages/lib/amd64:/usr/lib64:/lib64:/lib:/usr/lib (org.apache.zookeeper.server.ZooKeeperServer)
[2019-08-10 11:45:18,853] INFO Server environment:java.io.tmpdir=/tmp (org.apache.zookeeper.server.ZooKeeperServer)
[2019-08-10 11:45:18,853] INFO Server environment:java.compiler=<NA> (org.apache.zookeeper.server.ZooKeeperServer)
[2019-08-10 11:45:18,853] INFO Server environment:os.name=Linux (org.apache.zookeeper.server.ZooKeeperServer)
[2019-08-10 11:45:18,853] INFO Server environment:os.arch=amd64 (org.apache.zookeeper.server.ZooKeeperServer)
[2019-08-10 11:45:18,853] INFO Server environment:os.version=4.15.0-29deepin-generic (org.apache.zookeeper.server.ZooKeeperServer)
[2019-08-10 11:45:18,853] INFO Server environment:user.name=root (org.apache.zookeeper.server.ZooKeeperServer)
[2019-08-10 11:45:18,853] INFO Server environment:user.home=/root (org.apache.zookeeper.server.ZooKeeperServer)
[2019-08-10 11:45:18,853] INFO Server environment:user.dir=/root/kafka_2.10-0.10.2.1/bin (org.apache.zookeeper.server.ZooKeeperServer)
[2019-08-10 11:45:18,860] INFO tickTime set to 3000 (org.apache.zookeeper.server.ZooKeeperServer)
[2019-08-10 11:45:18,860] INFO minSessionTimeout set to -1 (org.apache.zookeeper.server.ZooKeeperServer)
[2019-08-10 11:45:18,860] INFO maxSessionTimeout set to -1 (org.apache.zookeeper.server.ZooKeeperServer)
[2019-08-10 11:45:18,911] INFO binding to port 0.0.0.0/0.0.0.0:2181 (org.apache.zookeeper.server.NIOServerCnxnFactory)

启动之后会发现2181端口被监听，可以同查看端口号来确定2181端口却是被监听

接下来就是启动kafka的服务器也就是在bin目录下的 kafka-server-start.sh 启动脚本，同样不输入任何的参数。

会看到提示要加载对应的配置文件。在config路径下面找到对应的配置文件

# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.# see kafka.server.KafkaConfig for additional details and defaults############################# Server Basics ############################## The id of the broker. This must be set to a unique integer for each broker.
broker.id=0# Switch to enable topic deletion or not, default value is false
#delete.topic.enable=true############################# Socket Server Settings ############################## The address the socket server listens on. It will get the value returned from
# java.net.InetAddress.getCanonicalHostName() if not configured.
#   FORMAT:
#     listeners = listener_name://host_name:port
#   EXAMPLE:
#     listeners = PLAINTEXT://your.host.name:9092
#listeners=PLAINTEXT://:9092# Hostname and port the broker will advertise to producers and consumers. If not set,
# it uses the value for "listeners" if configured.  Otherwise, it will use the value
# returned from java.net.InetAddress.getCanonicalHostName().
#advertised.listeners=PLAINTEXT://your.host.name:9092# Maps listener names to security protocols, the default is for them to be the same. See the config documentation for more details
#listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL# The number of threads handling network requests
num.network.threads=3# The number of threads doing disk I/O
num.io.threads=8# The send buffer (SO_SNDBUF) used by the socket server
socket.send.buffer.bytes=102400# The receive buffer (SO_RCVBUF) used by the socket server
socket.receive.buffer.bytes=102400# The maximum size of a request that the socket server will accept (protection against OOM)
socket.request.max.bytes=104857600############################# Log Basics ############################## A comma seperated list of directories under which to store log files
log.dirs=/tmp/kafka-logs# The default number of log partitions per topic. More partitions allow greater
# parallelism for consumption, but this will also result in more files across
# the brokers.
num.partitions=1# The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.
# This value is recommended to be increased for installations with data dirs located in RAID array.
num.recovery.threads.per.data.dir=1############################# Log Flush Policy ############################## Messages are immediately written to the filesystem but by default we only fsync() to sync
# the OS cache lazily. The following configurations control the flush of data to disk.
# There are a few important trade-offs here:
#    1. Durability: Unflushed data may be lost if you are not using replication.
#    2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush.
#    3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to exceessive seeks.
# The settings below allow one to configure the flush policy to flush data after a period of time or
# every N messages (or both). This can be done globally and overridden on a per-topic basis.# The number of messages to accept before forcing a flush of data to disk
#log.flush.interval.messages=10000# The maximum amount of time a message can sit in a log before we force a flush
#log.flush.interval.ms=1000############################# Log Retention Policy ############################## The following configurations control the disposal of log segments. The policy can
# be set to delete segments after a period of time, or after a given size has accumulated.
# A segment will be deleted whenever *either* of these criteria are met. Deletion always happens
# from the end of the log.# The minimum age of a log file to be eligible for deletion due to age
log.retention.hours=168# A size-based retention policy for logs. Segments are pruned from the log as long as the remaining
# segments don't drop below log.retention.bytes. Functions independently of log.retention.hours.
#log.retention.bytes=1073741824# The maximum size of a log segment file. When this size is reached a new log segment will be created.
log.segment.bytes=1073741824# The interval at which log segments are checked to see if they can be deleted according
# to the retention policies
log.retention.check.interval.ms=300000############################# Zookeeper ############################## Zookeeper connection string (see zookeeper docs for details).
# This is a comma separated host:port pairs, each corresponding to a zk
# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
# You can also append an optional chroot string to the urls to specify the
# root directory for all kafka znodes.
zookeeper.connect=localhost:2181# Timeout in ms for connecting to zookeeper
zookeeper.connection.timeout.ms=6000

会看到其中也有一些参数需要进行配置，使用如下方式进行启动，在配置文件中我们会看到存在一个端口号的配置kafka默认配置的9092端口。

参数介绍
broker.id=0:标识集群中的broker。
num.network.threads=3：网络线程数
num.io.threads=8：磁盘操作IO线程数
socket.send.buffer.bytes=102400 发送数据缓存大小
socket.receive.buffer.bytes=102400 接收数据缓存大小
zookeeper.connect=localhost:2181 zk 地址
zookeeper.connection.timeout.ms=6000 zk链接超时时间

root@nihui-PC:~/kafka_2.10-0.10.2.1/bin# ./kafka-server-start.sh ../config/server.properties
Picked up _JAVA_OPTIONS:   -Dawt.useSystemAAFontSettings=gasp
[2019-08-10 11:55:38,544] INFO KafkaConfig values: advertised.host.name = nulladvertised.listeners = nulladvertised.port = nullauthorizer.class.name = auto.create.topics.enable = trueauto.leader.rebalance.enable = truebackground.threads = 10broker.id = 0broker.id.generation.enable = truebroker.rack = nullcompression.type = producerconnections.max.idle.ms = 600000controlled.shutdown.enable = truecontrolled.shutdown.max.retries = 3controlled.shutdown.retry.backoff.ms = 5000controller.socket.timeout.ms = 30000create.topic.policy.class.name = nulldefault.replication.factor = 1delete.topic.enable = falsefetch.purgatory.purge.interval.requests = 1000group.max.session.timeout.ms = 300000group.min.session.timeout.ms = 6000host.name = inter.broker.listener.name = nullinter.broker.protocol.version = 0.10.2-IV0leader.imbalance.check.interval.seconds = 300leader.imbalance.per.broker.percentage = 10listener.security.protocol.map = SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,TRACE:TRACE,SASL_SSL:SASL_SSL,PLAINTEXT:PLAINTEXTlisteners = nulllog.cleaner.backoff.ms = 15000log.cleaner.dedupe.buffer.size = 134217728log.cleaner.delete.retention.ms = 86400000log.cleaner.enable = truelog.cleaner.io.buffer.load.factor = 0.9log.cleaner.io.buffer.size = 524288log.cleaner.io.max.bytes.per.second = 1.7976931348623157E308log.cleaner.min.cleanable.ratio = 0.5log.cleaner.min.compaction.lag.ms = 0log.cleaner.threads = 1log.cleanup.policy = [delete]log.dir = /tmp/kafka-logslog.dirs = /tmp/kafka-logslog.flush.interval.messages = 9223372036854775807log.flush.interval.ms = nulllog.flush.offset.checkpoint.interval.ms = 60000log.flush.scheduler.interval.ms = 9223372036854775807log.index.interval.bytes = 4096log.index.size.max.bytes = 10485760log.message.format.version = 0.10.2-IV0log.message.timestamp.difference.max.ms = 9223372036854775807log.message.timestamp.type = CreateTimelog.preallocate = falselog.retention.bytes = -1log.retention.check.interval.ms = 300000log.retention.hours = 168log.retention.minutes = nulllog.retention.ms = nulllog.roll.hours = 168log.roll.jitter.hours = 0log.roll.jitter.ms = nulllog.roll.ms = nulllog.segment.bytes = 1073741824log.segment.delete.delay.ms = 60000max.connections.per.ip = 2147483647max.connections.per.ip.overrides = message.max.bytes = 1000012metric.reporters = []metrics.num.samples = 2metrics.recording.level = INFOmetrics.sample.window.ms = 30000min.insync.replicas = 1num.io.threads = 8num.network.threads = 3num.partitions = 1num.recovery.threads.per.data.dir = 1num.replica.fetchers = 1offset.metadata.max.bytes = 4096offsets.commit.required.acks = -1offsets.commit.timeout.ms = 5000offsets.load.buffer.size = 5242880offsets.retention.check.interval.ms = 600000offsets.retention.minutes = 1440offsets.topic.compression.codec = 0offsets.topic.num.partitions = 50offsets.topic.replication.factor = 3offsets.topic.segment.bytes = 104857600port = 9092principal.builder.class = class org.apache.kafka.common.security.auth.DefaultPrincipalBuilderproducer.purgatory.purge.interval.requests = 1000queued.max.requests = 500quota.consumer.default = 9223372036854775807quota.producer.default = 9223372036854775807quota.window.num = 11quota.window.size.seconds = 1replica.fetch.backoff.ms = 1000replica.fetch.max.bytes = 1048576replica.fetch.min.bytes = 1replica.fetch.response.max.bytes = 10485760replica.fetch.wait.max.ms = 500replica.high.watermark.checkpoint.interval.ms = 5000replica.lag.time.max.ms = 10000replica.socket.receive.buffer.bytes = 65536replica.socket.timeout.ms = 30000replication.quota.window.num = 11replication.quota.window.size.seconds = 1request.timeout.ms = 30000reserved.broker.max.id = 1000sasl.enabled.mechanisms = [GSSAPI]sasl.kerberos.kinit.cmd = /usr/bin/kinitsasl.kerberos.min.time.before.relogin = 60000sasl.kerberos.principal.to.local.rules = [DEFAULT]sasl.kerberos.service.name = nullsasl.kerberos.ticket.renew.jitter = 0.05sasl.kerberos.ticket.renew.window.factor = 0.8sasl.mechanism.inter.broker.protocol = GSSAPIsecurity.inter.broker.protocol = PLAINTEXTsocket.receive.buffer.bytes = 102400socket.request.max.bytes = 104857600socket.send.buffer.bytes = 102400ssl.cipher.suites = nullssl.client.auth = nonessl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]ssl.endpoint.identification.algorithm = nullssl.key.password = nullssl.keymanager.algorithm = SunX509ssl.keystore.location = nullssl.keystore.password = nullssl.keystore.type = JKSssl.protocol = TLSssl.provider = nullssl.secure.random.implementation = nullssl.trustmanager.algorithm = PKIXssl.truststore.location = nullssl.truststore.password = nullssl.truststore.type = JKSunclean.leader.election.enable = truezookeeper.connect = localhost:2181zookeeper.connection.timeout.ms = 6000zookeeper.session.timeout.ms = 6000zookeeper.set.acl = falsezookeeper.sync.time.ms = 2000(kafka.server.KafkaConfig)
[2019-08-10 11:55:38,576] INFO starting (kafka.server.KafkaServer)
[2019-08-10 11:55:38,578] INFO Connecting to zookeeper on localhost:2181 (kafka.server.KafkaServer)
[2019-08-10 11:55:38,589] INFO Starting ZkClient event thread. (org.I0Itec.zkclient.ZkEventThread)
[2019-08-10 11:55:38,593] INFO Client environment:zookeeper.version=3.4.9-1757313, built on 08/23/2016 06:50 GMT (org.apache.zookeeper.ZooKeeper)
[2019-08-10 11:55:38,593] INFO Client environment:host.name=nihui-PC (org.apache.zookeeper.ZooKeeper)
[2019-08-10 11:55:38,593] INFO Client environment:java.version=1.8.0_161 (org.apache.zookeeper.ZooKeeper)
[2019-08-10 11:55:38,593] INFO Client environment:java.vendor=Oracle Corporation (org.apache.zookeeper.ZooKeeper)
[2019-08-10 11:55:38,593] INFO Client environment:java.home=/root/jdk1.8.0_161/jre (org.apache.zookeeper.ZooKeeper)
[2019-08-10 11:55:38,593] INFO Client environment:java.class.path=.:/root/jdk1.8.0_161/lib/dt.jar:/root/jdk1.8.0_161/lib/tools.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/aopalliance-repackaged-2.5.0-b05.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/argparse4j-0.7.0.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/connect-api-0.10.2.1.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/connect-file-0.10.2.1.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/connect-json-0.10.2.1.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/connect-runtime-0.10.2.1.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/connect-transforms-0.10.2.1.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/guava-18.0.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/hk2-api-2.5.0-b05.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/hk2-locator-2.5.0-b05.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/hk2-utils-2.5.0-b05.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jackson-annotations-2.8.0.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jackson-annotations-2.8.5.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jackson-core-2.8.5.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jackson-databind-2.8.5.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jackson-jaxrs-base-2.8.5.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jackson-jaxrs-json-provider-2.8.5.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jackson-module-jaxb-annotations-2.8.5.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/javassist-3.20.0-GA.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/javax.annotation-api-1.2.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/javax.inject-1.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/javax.inject-2.5.0-b05.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/javax.servlet-api-3.1.0.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/javax.ws.rs-api-2.0.1.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jersey-client-2.24.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jersey-common-2.24.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jersey-container-servlet-2.24.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jersey-container-servlet-core-2.24.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jersey-guava-2.24.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jersey-media-jaxb-2.24.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jersey-server-2.24.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jetty-continuation-9.2.15.v20160210.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jetty-http-9.2.15.v20160210.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jetty-io-9.2.15.v20160210.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jetty-security-9.2.15.v20160210.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jetty-server-9.2.15.v20160210.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jetty-servlet-9.2.15.v20160210.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jetty-servlets-9.2.15.v20160210.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jetty-util-9.2.15.v20160210.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/jopt-simple-5.0.3.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/kafka_2.10-0.10.2.1.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/kafka_2.10-0.10.2.1-sources.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/kafka_2.10-0.10.2.1-test-sources.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/kafka-clients-0.10.2.1.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/kafka-log4j-appender-0.10.2.1.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/kafka-streams-0.10.2.1.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/kafka-streams-examples-0.10.2.1.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/kafka-tools-0.10.2.1.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/log4j-1.2.17.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/lz4-1.3.0.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/metrics-core-2.2.0.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/osgi-resource-locator-1.0.1.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/reflections-0.9.10.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/rocksdbjni-5.0.1.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/scala-library-2.10.6.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/slf4j-api-1.7.21.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/slf4j-log4j12-1.7.21.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/snappy-java-1.1.2.6.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/validation-api-1.1.0.Final.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/zkclient-0.10.jar:/root/kafka_2.10-0.10.2.1/bin/../libs/zookeeper-3.4.9.jar (org.apache.zookeeper.ZooKeeper)
[2019-08-10 11:55:38,593] INFO Client environment:java.library.path=/usr/java/packages/lib/amd64:/usr/lib64:/lib64:/lib:/usr/lib (org.apache.zookeeper.ZooKeeper)
[2019-08-10 11:55:38,593] INFO Client environment:java.io.tmpdir=/tmp (org.apache.zookeeper.ZooKeeper)
[2019-08-10 11:55:38,593] INFO Client environment:java.compiler=<NA> (org.apache.zookeeper.ZooKeeper)
[2019-08-10 11:55:38,593] INFO Client environment:os.name=Linux (org.apache.zookeeper.ZooKeeper)
[2019-08-10 11:55:38,593] INFO Client environment:os.arch=amd64 (org.apache.zookeeper.ZooKeeper)
[2019-08-10 11:55:38,593] INFO Client environment:os.version=4.15.0-29deepin-generic (org.apache.zookeeper.ZooKeeper)
[2019-08-10 11:55:38,593] INFO Client environment:user.name=root (org.apache.zookeeper.ZooKeeper)
[2019-08-10 11:55:38,593] INFO Client environment:user.home=/root (org.apache.zookeeper.ZooKeeper)
[2019-08-10 11:55:38,593] INFO Client environment:user.dir=/root/kafka_2.10-0.10.2.1/bin (org.apache.zookeeper.ZooKeeper)
[2019-08-10 11:55:38,594] INFO Initiating client connection, connectString=localhost:2181 sessionTimeout=6000 watcher=org.I0Itec.zkclient.ZkClient@4567f35d (org.apache.zookeeper.ZooKeeper)
[2019-08-10 11:55:38,605] INFO Waiting for keeper state SyncConnected (org.I0Itec.zkclient.ZkClient)
[2019-08-10 11:55:38,609] INFO Opening socket connection to server localhost/127.0.0.1:2181. Will not attempt to authenticate using SASL (unknown error) (org.apache.zookeeper.ClientCnxn)
[2019-08-10 11:55:38,619] INFO Socket connection established to localhost/127.0.0.1:2181, initiating session (org.apache.zookeeper.ClientCnxn)
[2019-08-10 11:55:38,686] INFO Session establishment complete on server localhost/127.0.0.1:2181, sessionid = 0x16c79a1704c0000, negotiated timeout = 6000 (org.apache.zookeeper.ClientCnxn)
[2019-08-10 11:55:38,687] INFO zookeeper state changed (SyncConnected) (org.I0Itec.zkclient.ZkClient)
[2019-08-10 11:55:38,911] INFO Cluster ID = VogbcQUaRR2E3y6tLPRDTg (kafka.server.KafkaServer)
[2019-08-10 11:55:38,917] WARN No meta.properties file under dir /tmp/kafka-logs/meta.properties (kafka.server.BrokerMetadataCheckpoint)
[2019-08-10 11:55:38,939] INFO [ThrottledRequestReaper-Fetch], Starting  (kafka.server.ClientQuotaManager$ThrottledRequestReaper)
[2019-08-10 11:55:38,940] INFO [ThrottledRequestReaper-Produce], Starting  (kafka.server.ClientQuotaManager$ThrottledRequestReaper)
[2019-08-10 11:55:38,965] INFO Log directory '/tmp/kafka-logs' not found, creating it. (kafka.log.LogManager)
[2019-08-10 11:55:38,976] INFO Loading logs. (kafka.log.LogManager)
[2019-08-10 11:55:38,980] INFO Logs loading complete in 4 ms. (kafka.log.LogManager)
[2019-08-10 11:55:39,029] INFO Starting log cleanup with a period of 300000 ms. (kafka.log.LogManager)
[2019-08-10 11:55:39,030] INFO Starting log flusher with a default period of 9223372036854775807 ms. (kafka.log.LogManager)
[2019-08-10 11:55:39,067] INFO Awaiting socket connections on 0.0.0.0:9092. (kafka.network.Acceptor)
[2019-08-10 11:55:39,069] INFO [Socket Server on Broker 0], Started 1 acceptor threads (kafka.network.SocketServer)
[2019-08-10 11:55:39,081] INFO [ExpirationReaper-0], Starting  (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2019-08-10 11:55:39,082] INFO [ExpirationReaper-0], Starting  (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2019-08-10 11:55:39,104] INFO Creating /controller (is it secure? false) (kafka.utils.ZKCheckedEphemeral)
[2019-08-10 11:55:39,121] INFO Result of znode creation is: OK (kafka.utils.ZKCheckedEphemeral)
[2019-08-10 11:55:39,122] INFO 0 successfully elected as leader (kafka.server.ZookeeperLeaderElector)
[2019-08-10 11:55:39,198] INFO [ExpirationReaper-0], Starting  (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2019-08-10 11:55:39,200] INFO [ExpirationReaper-0], Starting  (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2019-08-10 11:55:39,201] INFO [ExpirationReaper-0], Starting  (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2019-08-10 11:55:39,208] INFO [GroupCoordinator 0]: Starting up. (kafka.coordinator.GroupCoordinator)
[2019-08-10 11:55:39,209] INFO [GroupCoordinator 0]: Startup complete. (kafka.coordinator.GroupCoordinator)
[2019-08-10 11:55:39,210] INFO [Group Metadata Manager on Broker 0]: Removed 0 expired offsets in 0 milliseconds. (kafka.coordinator.GroupMetadataManager)
[2019-08-10 11:55:39,229] INFO Will not load MX4J, mx4j-tools.jar is not in the classpath (kafka.utils.Mx4jLoader$)
[2019-08-10 11:55:39,244] INFO New leader is 0 (kafka.server.ZookeeperLeaderElector$LeaderChangeListener)
[2019-08-10 11:55:39,247] INFO Creating /brokers/ids/0 (is it secure? false) (kafka.utils.ZKCheckedEphemeral)
[2019-08-10 11:55:39,253] INFO Result of znode creation is: OK (kafka.utils.ZKCheckedEphemeral)
[2019-08-10 11:55:39,254] INFO Registered broker 0 at path /brokers/ids/0 with addresses: EndPoint(nihui-PC,9092,ListenerName(PLAINTEXT),PLAINTEXT) (kafka.utils.ZkUtils)
[2019-08-10 11:55:39,255] WARN No meta.properties file under dir /tmp/kafka-logs/meta.properties (kafka.server.BrokerMetadataCheckpoint)
[2019-08-10 11:55:39,369] INFO Kafka version : 0.10.2.1 (org.apache.kafka.common.utils.AppInfoParser)
[2019-08-10 11:55:39,369] INFO Kafka commitId : e89bffd6b2eff799 (org.apache.kafka.common.utils.AppInfoParser)
[2019-08-10 11:55:39,370] INFO [Kafka Server 0], started (kafka.server.KafkaServer)

正常启动之后进行端口的查看，这时候会发现端口被正常启动监听。

第三步、验证Kafka的正常使用

在之前的分析中提到过kafka其实是基于发布订阅模式的那么在验证的时候就需要创建一个Topic来进行验证那么怎么去创建一个Topic呢？在kafka的bin目录下面有一个对应的创建Topic的脚本。可以使用同样的方式来查看它的使用方式。

root@nihui-PC:~/kafka_2.10-0.10.2.1/bin# ./kafka-topics.sh
Picked up _JAVA_OPTIONS:   -Dawt.useSystemAAFontSettings=gasp
Create, delete, describe, or change a topic.
Option                                   Description
------                                   -----------
--alter                                  Alter the number of partitions,        replica assignment, and/or           configuration for the topic.
--config <String: name=value>            A topic configuration override for the topic being created or altered.The   following is a list of valid         configurations:                      cleanup.policy                        compression.type                      delete.retention.ms                   file.delete.delay.ms                  flush.messages                        flush.ms                              follower.replication.throttled.       replicas                             index.interval.bytes                  leader.replication.throttled.replicas max.message.bytes                     message.format.version                message.timestamp.difference.max.ms   message.timestamp.type                min.cleanable.dirty.ratio             min.compaction.lag.ms                 min.insync.replicas                   preallocate                           retention.bytes                       retention.ms                          segment.bytes                         segment.index.bytes                   segment.jitter.ms                     segment.ms                            unclean.leader.election.enable        See the Kafka documentation for full   details on the topic configs.
--create                                 Create a new topic.
--delete                                 Delete a topic
--delete-config <String: name>           A topic configuration override to be   removed for an existing topic (see   the list of configurations under the --config option).
--describe                               List details for the given topics.
--disable-rack-aware                     Disable rack aware replica assignment
--force                                  Suppress console prompts
--help                                   Print usage information.
--if-exists                              if set when altering or deleting       topics, the action will only execute if the topic exists
--if-not-exists                          if set when creating topics, the       action will only execute if the      topic does not already exist
--list                                   List all available topics.
--partitions <Integer: # of partitions>  The number of partitions for the topic being created or altered (WARNING:   If partitions are increased for a    topic that has a key, the partition  logic or ordering of the messages    will be affected
--replica-assignment <String:            A list of manual partition-to-broker   broker_id_for_part1_replica1 :           assignments for the topic being      broker_id_for_part1_replica2 ,           created or altered.                  broker_id_for_part2_replica1 :                                                broker_id_for_part2_replica2 , ...>
--replication-factor <Integer:           The replication factor for each        replication factor>                      partition in the topic being created.
--topic <String: topic>                  The topic to be create, alter or       describe. Can also accept a regular  expression except for --create option
--topics-with-overrides                  if set when describing topics, only    show topics that have overridden     configs
--unavailable-partitions                 if set when describing topics, only    show partitions whose leader is not  available
--under-replicated-partitions            if set when describing topics, only    show under replicated partitions
--zookeeper <String: urls>               REQUIRED: The connection string for    the zookeeper connection in the form host:port. Multiple URLS can be      given to allow fail-over.

调用完成之后会发现其实它的参数还是蛮多的，但是其中最为重要就是制定zk的列表，下面就来创建一个名字为test的topic

./kafka-topics.sh --zookeeper localhost:2181 --create --topic test --partitions 3 --replication-factor 1

对于其中的细节性的参数可以先不用了解。

其实到这里，会看到在broker的控制台和在zookeeper的控制台都有相应的信息输出

broker

zookeeper

这个时候通过命令查看，会发现这个topic已经被注册到Kafka

./kafka-topics.sh --zookeeper localhost:2181 --describe --topic test

启动之后接下来的事情就是启动一个consumer来进行订阅

 ./kafka-console-consumer.sh --zookeeper localhost:2181 --topic test

接下来就是在新的命令行窗口启动一个producer来进行消息的发布。

./kafka-console-producer.sh --broker-list localhost:9092 --topic test

双端启动之后，就开始测试关于消息的发送。
producer

consumer

到这里的测试算是做完了，整个kafka搭建完成。

通过以上的操作从简单的入门层面上了解了kafka的工作原理，以及启动原理，结合下面的架构图了解了整个架构中各个组件的使用情况。

Kafka架构图