Kafka2.12安装与配置/生产与消费

一、Kafka安装与配置

1.1 Java环境为前提

jdk下载地址链接：jdk1.8 提取码：9plz
zookeeper下载地址链接：zookeeper3.4.14 提取码：zkvq
kafka下载地址链接：kafka2.12 提取码：oroq

1、上传jdk-8u261-linux-x64.rpm到服务器并安装
 #安装命令rpm -ivh jdk-8u261-linux-x64.rpm

2、配置环境变量
 # 编辑配置文件,jdk的bin目录到/etc/profile文件，对所有用户的shell有效vim /etc/profile# 生效source /etc/profile


 export JAVA_HOME=/usr/java/jdk1.8.0_261-amd64export PATH=$PATH:$JAVA_HOME/bin# 验证java -version

至此，jdk安装成功

1.2 Zookeeper的安装配置

1、上传zookeeper-3.4.14.tar.gz到服务器，解压到/opt
 # 解压zk到指定目录tar -zxf zookeeper-3.4.14.tar.gz -C /opt

2、修改Zookeeper保存数据的目录，dataDir
 # 进入conf配置目录cd /opt/zookeeper-3.4.14/conf# 复制zoo_sample.cfg命名为zoo.cfgcp zoo_sample.cfg zoo.cfg# 编辑zoo.cfg文件vim zoo.cfgdataDir=/var/dabing/zookeeper/data

3、编辑/etc/profile，使配置生效
 设置环境变量ZOO_LOG_DIR，指定Zookeeper保存日志的位置；ZOOKEEPER_PREFIX指向Zookeeper的解压目录；将Zookeeper的bin目录添加到PATH中：

 export ZOOKEEPER_PREFIX=/opt/zookeeper-3.4.14export PATH=$PATH:$ZOOKEEPER_PREFIX/binexport ZOO_LOG_DIR=/var/dabing/zookeeper/log//更新配置文件source /etc/profile

4、启动Zookeeper，确认Zookeeper的状态
 zkServer.sh start


至此，zookeeper安装成功

1.3 Kafka的安装与配置

1、上传kafka_2.12-1.0.2.tgz到服务器并解压
 tar -zxf kafka_2.12-1.0.2.tgz -C /opt

2、配置环境变量并生效
 vim /etc/profile #kafkaexport KAFKA=/opt/kafka_2.12-1.0.2export PATH=$PATH:$KAFKA/bin

3、配置/opt/kafka_2.12-1.0.2/config中的server.properties文件
 vi  /opt/kafka_2.12-1.0.2/config/server.properties


 Kafka连接Zookeeper的地址，此处使用本地启动的Zookeeper实例连接地址是localhost:2181后面的 myKafka 是Kafka在Zookeeper中的根节点路径：

配置kafka存储持久化数据目录

创建上述持久化数据目录

4、启动Kafka：

进入Kafka安装的根目录，执行如下命令：
kafka-server-start.sh $KAFKA/config/server.properties


启动成功，可以看到控制台输出的最后一行的started状态：此时kafka安装成功

详细执行日志

[root@kafka bin]# kafka-server-start.sh ../config/server.properties
[2020-11-03 03:30:56,120] INFO KafkaConfig values: advertised.host.name = nulladvertised.listeners = nulladvertised.port = nullalter.config.policy.class.name = 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.records.purgatory.purge.interval.requests = 1delete.topic.enable = truefetch.purgatory.purge.interval.requests = 1000group.initial.rebalance.delay.ms = 0group.max.session.timeout.ms = 300000group.min.session.timeout.ms = 6000host.name = inter.broker.listener.name = nullinter.broker.protocol.version = 1.0-IV0leader.imbalance.check.interval.seconds = 300leader.imbalance.per.broker.percentage = 10listener.security.protocol.map = PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSLlisteners = 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 = /var/dabing/kafka/kafka-logs---------------------------------修改 log.flush.interval.messages = 9223372036854775807log.flush.interval.ms = nulllog.flush.offset.checkpoint.interval.ms = 60000log.flush.scheduler.interval.ms = 9223372036854775807log.flush.start.offset.checkpoint.interval.ms = 60000log.index.interval.bytes = 4096log.index.size.max.bytes = 10485760log.message.format.version = 1.0-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 = 1offsets.topic.segment.bytes = 104857600port = 9092principal.builder.class = nullproducer.purgatory.purge.interval.requests = 1000queued.max.request.bytes = -1queued.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 = JKStransaction.abort.timed.out.transaction.cleanup.interval.ms = 60000transaction.max.timeout.ms = 900000transaction.remove.expired.transaction.cleanup.interval.ms = 3600000transaction.state.log.load.buffer.size = 5242880transaction.state.log.min.isr = 1transaction.state.log.num.partitions = 50transaction.state.log.replication.factor = 1transaction.state.log.segment.bytes = 104857600transactional.id.expiration.ms = 604800000unclean.leader.election.enable = falsezookeeper.connect = localhost:2181/myKafka---------------------------------修改zookeeper.connection.timeout.ms = 6000zookeeper.session.timeout.ms = 6000zookeeper.set.acl = falsezookeeper.sync.time.ms = 2000(kafka.server.KafkaConfig)
[2020-11-03 03:30:56,315] INFO starting (kafka.server.KafkaServer)
[2020-11-03 03:30:56,330] INFO Connecting to zookeeper on localhost:2181/myKafka (kafka.server.KafkaServer)
[2020-11-03 03:30:56,369] INFO Starting ZkClient event thread. (org.I0Itec.zkclient.ZkEventThread)
[2020-11-03 03:30:56,509] INFO Client environment:zookeeper.version=3.4.10-39d3a4f269333c922ed3db283be479f9deacaa0f, built on 03/23/2017 10:13 GMT (org.apache.zookeeper.ZooKeeper)
[2020-11-03 03:30:56,510] INFO Client environment:host.name=kafka (org.apache.zookeeper.ZooKeeper)
[2020-11-03 03:30:56,510] INFO Client environment:java.version=1.8.0_261 (org.apache.zookeeper.ZooKeeper)
[2020-11-03 03:30:56,511] INFO Client environment:java.vendor=Oracle Corporation (org.apache.zookeeper.ZooKeeper)
[2020-11-03 03:30:56,511] INFO Client environment:java.home=/usr/java/jdk1.8.0_261-amd64/jre (org.apache.zookeeper.ZooKeeper)
[2020-11-03 03:30:56,512] INFO Client environment:java.class.path=:/opt/kafka_2.12-1.0.2/bin/../libs/aopalliance-repackaged-2.5.0-b32.jar:/opt/kafka_2.12-1.0.2/bin/../libs/argparse4j-0.7.0.jar:/opt/kafka_2.12-1.0.2/bin/../libs/commons-lang3-3.5.jar:/opt/kafka_2.12-1.0.2/bin/../libs/connect-api-1.0.2.jar:/opt/kafka_2.12-1.0.2/bin/../libs/connect-file-1.0.2.jar:/opt/kafka_2.12-1.0.2/bin/../libs/connect-json-1.0.2.jar:/opt/kafka_2.12-1.0.2/bin/../libs/connect-runtime-1.0.2.jar:/opt/kafka_2.12-1.0.2/bin/../libs/connect-transforms-1.0.2.jar:/opt/kafka_2.12-1.0.2/bin/../libs/guava-20.0.jar:/opt/kafka_2.12-1.0.2/bin/../libs/hk2-api-2.5.0-b32.jar:/opt/kafka_2.12-1.0.2/bin/../libs/hk2-locator-2.5.0-b32.jar:/opt/kafka_2.12-1.0.2/bin/../libs/hk2-utils-2.5.0-b32.jar:/opt/kafka_2.12-1.0.2/bin/../libs/jackson-annotations-2.9.6.jar:/opt/kafka_2.12-1.0.2/bin/../libs/jackson-core-2.9.6.jar:/opt/kafka_2.12-1.0.2/bin/../libs/jackson-databind-2.9.6.jar:/opt/kafka_2.12-1.0.2/bin/../libs/jackson-jaxrs-base-2.9.6.jar:/opt/kafka_2.12-1.0.2/bin/../libs/jackson-jaxrs-json-provider-2.9.6.jar:/opt/kafka_2.12-1.0.2/bin/../libs/jackson-module-jaxb-annotations-2.9.6.jar:/opt/kafka_2.12-1.0.2/bin/../libs/javassist-3.20.0-GA.jar:/opt/kafka_2.12-1.0.2/bin/../libs/javassist-3.21.0-GA.jar:/opt/kafka_2.12-1.0.2/bin/../libs/javax.annotation-api-1.2.jar:/opt/kafka_2.12-1.0.2/bin/../libs/javax.inject-1.jar:/opt/kafka_2.12-1.0.2/bin/../libs/javax.inject-2.5.0-b32.jar:/opt/kafka_2.12-1.0.2/bin/../libs/javax.servlet-api-3.1.0.jar:/opt/kafka_2.12-1.0.2/bin/../libs/javax.ws.rs-api-2.0.1.jar:/opt/kafka_2.12-1.0.2/bin/../libs/jersey-client-2.25.1.jar:/opt/kafka_2.12-1.0.2/bin/../libs/jersey-common-2.25.1.jar:/opt/kafka_2.12-1.0.2/bin/../libs/jersey-container-servlet-2.25.1.jar:/opt/kafka_2.12-1.0.2/bin/../libs/jersey-container-servlet-core-2.25.1.jar:/opt/kafka_2.12-1.0.2/bin/../libs/jersey-guava-2.25.1.jar:/opt/kafka_2.12-1.0.2/bin/../libs/jersey-media-jaxb-2.25.1.jar:/opt/kafka_2.12-1.0.2/bin/../libs/jersey-server-2.25.1.jar:/opt/kafka_2.12-1.0.2/bin/../libs/jetty-continuation-9.2.22.v20170606.jar:/opt/kafka_2.12-1.0.2/bin/../libs/jetty-http-9.2.22.v20170606.jar:/opt/kafka_2.12-1.0.2/bin/../libs/jetty-io-9.2.22.v20170606.jar:/opt/kafka_2.12-1.0.2/bin/../libs/jetty-security-9.2.22.v20170606.jar:/opt/kafka_2.12-1.0.2/bin/../libs/jetty-server-9.2.22.v20170606.jar:/opt/kafka_2.12-1.0.2/bin/../libs/jetty-servlet-9.2.22.v20170606.jar:/opt/kafka_2.12-1.0.2/bin/../libs/jetty-servlets-9.2.22.v20170606.jar:/opt/kafka_2.12-1.0.2/bin/../libs/jetty-util-9.2.22.v20170606.jar:/opt/kafka_2.12-1.0.2/bin/../libs/jopt-simple-5.0.4.jar:/opt/kafka_2.12-1.0.2/bin/../libs/kafka_2.12-1.0.2.jar:/opt/kafka_2.12-1.0.2/bin/../libs/kafka_2.12-1.0.2-sources.jar:/opt/kafka_2.12-1.0.2/bin/../libs/kafka_2.12-1.0.2-test-sources.jar:/opt/kafka_2.12-1.0.2/bin/../libs/kafka-clients-1.0.2.jar:/opt/kafka_2.12-1.0.2/bin/../libs/kafka-log4j-appender-1.0.2.jar:/opt/kafka_2.12-1.0.2/bin/../libs/kafka-streams-1.0.2.jar:/opt/kafka_2.12-1.0.2/bin/../libs/kafka-streams-examples-1.0.2.jar:/opt/kafka_2.12-1.0.2/bin/../libs/kafka-tools-1.0.2.jar:/opt/kafka_2.12-1.0.2/bin/../libs/log4j-1.2.17.jar:/opt/kafka_2.12-1.0.2/bin/../libs/lz4-java-1.4.jar:/opt/kafka_2.12-1.0.2/bin/../libs/maven-artifact-3.5.0.jar:/opt/kafka_2.12-1.0.2/bin/../libs/metrics-core-2.2.0.jar:/opt/kafka_2.12-1.0.2/bin/../libs/osgi-resource-locator-1.0.1.jar:/opt/kafka_2.12-1.0.2/bin/../libs/plexus-utils-3.0.24.jar:/opt/kafka_2.12-1.0.2/bin/../libs/reflections-0.9.11.jar:/opt/kafka_2.12-1.0.2/bin/../libs/rocksdbjni-5.7.3.jar:/opt/kafka_2.12-1.0.2/bin/../libs/scala-library-2.12.4.jar:/opt/kafka_2.12-1.0.2/bin/../libs/slf4j-api-1.7.25.jar:/opt/kafka_2.12-1.0.2/bin/../libs/slf4j-log4j12-1.7.25.jar:/opt/kafka_2.12-1.0.2/bin/../libs/snappy-java-1.1.4.jar:/opt/kafka_2.12-1.0.2/bin/../libs/validation-api-1.1.0.Final.jar:/opt/kafka_2.12-1.0.2/bin/../libs/zkclient-0.10.jar:/opt/kafka_2.12-1.0.2/bin/../libs/zookeeper-3.4.10.jar (org.apache.zookeeper.ZooKeeper)
[2020-11-03 03:30:56,518] INFO Client environment:java.library.path=/usr/java/packages/lib/amd64:/usr/lib64:/lib64:/lib:/usr/lib (org.apache.zookeeper.ZooKeeper)
[2020-11-03 03:30:56,518] INFO Client environment:java.io.tmpdir=/tmp (org.apache.zookeeper.ZooKeeper)
[2020-11-03 03:30:56,519] INFO Client environment:java.compiler=<NA> (org.apache.zookeeper.ZooKeeper)
[2020-11-03 03:30:56,525] INFO Client environment:os.name=Linux (org.apache.zookeeper.ZooKeeper)
[2020-11-03 03:30:56,526] INFO Client environment:os.arch=amd64 (org.apache.zookeeper.ZooKeeper)
[2020-11-03 03:30:56,526] INFO Client environment:os.version=3.10.0-957.el7.x86_64 (org.apache.zookeeper.ZooKeeper)
[2020-11-03 03:30:56,526] INFO Client environment:user.name=root (org.apache.zookeeper.ZooKeeper)
[2020-11-03 03:30:56,527] INFO Client environment:user.home=/root (org.apache.zookeeper.ZooKeeper)
[2020-11-03 03:30:56,527] INFO Client environment:user.dir=/opt/kafka_2.12-1.0.2/bin (org.apache.zookeeper.ZooKeeper)
[2020-11-03 03:30:56,531] INFO Initiating client connection, connectString=localhost:2181 sessionTimeout=6000 watcher=org.I0Itec.zkclient.ZkClient@1b1473ab (org.apache.zookeeper.ZooKeeper)
[2020-11-03 03:30:56,565] INFO Waiting for keeper state SyncConnected (org.I0Itec.zkclient.ZkClient)
[2020-11-03 03:30:56,677] INFO Opening socket connection to server localhost.localdomain/127.0.0.1:2181. Will not attempt to authenticate using SASL (unknown error) (org.apache.zookeeper.ClientCnxn)
[2020-11-03 03:30:56,689] INFO Socket connection established to localhost.localdomain/127.0.0.1:2181, initiating session (org.apache.zookeeper.ClientCnxn)
[2020-11-03 03:30:56,757] INFO Session establishment complete on server localhost.localdomain/127.0.0.1:2181, sessionid = 0x100002bf2290000, negotiated timeout = 6000 (org.apache.zookeeper.ClientCnxn)
[2020-11-03 03:30:56,763] INFO zookeeper state changed (SyncConnected) (org.I0Itec.zkclient.ZkClient)
[2020-11-03 03:30:56,834] INFO Created zookeeper path /myKafka (kafka.server.KafkaServer)
[2020-11-03 03:30:56,835] INFO Terminate ZkClient event thread. (org.I0Itec.zkclient.ZkEventThread)
[2020-11-03 03:30:56,842] INFO Session: 0x100002bf2290000 closed (org.apache.zookeeper.ZooKeeper)
[2020-11-03 03:30:56,844] INFO EventThread shut down for session: 0x100002bf2290000 (org.apache.zookeeper.ClientCnxn)
[2020-11-03 03:30:56,844] INFO Initiating client connection, connectString=localhost:2181/myKafka sessionTimeout=6000 watcher=org.I0Itec.zkclient.ZkClient@35ef1869 (org.apache.zookeeper.ZooKeeper)
[2020-11-03 03:30:56,846] INFO Starting ZkClient event thread. (org.I0Itec.zkclient.ZkEventThread)
[2020-11-03 03:30:56,847] INFO Waiting for keeper state SyncConnected (org.I0Itec.zkclient.ZkClient)
[2020-11-03 03:30:57,031] INFO Opening socket connection to server localhost.localdomain/127.0.0.1:2181. Will not attempt to authenticate using SASL (unknown error) (org.apache.zookeeper.ClientCnxn)
[2020-11-03 03:30:57,033] INFO Socket connection established to localhost.localdomain/127.0.0.1:2181, initiating session (org.apache.zookeeper.ClientCnxn)
[2020-11-03 03:30:57,052] INFO Session establishment complete on server localhost.localdomain/127.0.0.1:2181, sessionid = 0x100002bf2290001, negotiated timeout = 6000 (org.apache.zookeeper.ClientCnxn)
[2020-11-03 03:30:57,055] INFO zookeeper state changed (SyncConnected) (org.I0Itec.zkclient.ZkClient)
[2020-11-03 03:30:57,478] INFO Cluster ID = 4U44C6GMQr-oyd1e-7TuEA (kafka.server.KafkaServer)
[2020-11-03 03:30:57,495] WARN No meta.properties file under dir /var/dabing/kafka/kafka-logs/meta.properties (kafka.server.BrokerMetadataCheckpoint)
[2020-11-03 03:30:57,540] INFO [ThrottledRequestReaper-Fetch]: Starting (kafka.server.ClientQuotaManager$ThrottledRequestReaper)
[2020-11-03 03:30:57,544] INFO [ThrottledRequestReaper-Produce]: Starting (kafka.server.ClientQuotaManager$ThrottledRequestReaper)
[2020-11-03 03:30:57,550] INFO [ThrottledRequestReaper-Request]: Starting (kafka.server.ClientQuotaManager$ThrottledRequestReaper)
[2020-11-03 03:30:57,696] INFO Loading logs. (kafka.log.LogManager)
[2020-11-03 03:30:57,727] INFO Logs loading complete in 31 ms. (kafka.log.LogManager)
[2020-11-03 03:30:58,031] INFO Starting log cleanup with a period of 300000 ms. (kafka.log.LogManager)
[2020-11-03 03:30:58,047] INFO Starting log flusher with a default period of 9223372036854775807 ms. (kafka.log.LogManager)
[2020-11-03 03:30:59,285] INFO Awaiting socket connections on 0.0.0.0:9092. (kafka.network.Acceptor)
[2020-11-03 03:30:59,292] INFO [SocketServer brokerId=0] Started 1 acceptor threads (kafka.network.SocketServer)
[2020-11-03 03:30:59,438] INFO [ExpirationReaper-0-Produce]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2020-11-03 03:30:59,438] INFO [ExpirationReaper-0-DeleteRecords]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2020-11-03 03:30:59,438] INFO [ExpirationReaper-0-Fetch]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2020-11-03 03:30:59,457] INFO [LogDirFailureHandler]: Starting (kafka.server.ReplicaManager$LogDirFailureHandler)
[2020-11-03 03:30:59,586] INFO [ExpirationReaper-0-Heartbeat]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2020-11-03 03:30:59,595] INFO [ExpirationReaper-0-topic]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2020-11-03 03:30:59,599] INFO [ExpirationReaper-0-Rebalance]: Starting (kafka.server.DelayedOperationPurgatory$ExpiredOperationReaper)
[2020-11-03 03:30:59,628] INFO Creating /controller (is it secure? false) (kafka.utils.ZKCheckedEphemeral)
[2020-11-03 03:30:59,639] INFO [GroupCoordinator 0]: Starting up. (kafka.coordinator.group.GroupCoordinator)
[2020-11-03 03:30:59,642] INFO [GroupCoordinator 0]: Startup complete. (kafka.coordinator.group.GroupCoordinator)
[2020-11-03 03:30:59,652] INFO Result of znode creation is: OK (kafka.utils.ZKCheckedEphemeral)
[2020-11-03 03:30:59,683] INFO [GroupMetadataManager brokerId=0] Removed 0 expired offsets in 14 milliseconds. (kafka.coordinator.group.GroupMetadataManager)
[2020-11-03 03:30:59,786] INFO [ProducerId Manager 0]: Acquired new producerId block (brokerId:0,blockStartProducerId:0,blockEndProducerId:999) by writing to Zk with path version 1 (kafka.coordinator.transaction.ProducerIdManager)
[2020-11-03 03:30:59,859] INFO [TransactionCoordinator id=0] Starting up. (kafka.coordinator.transaction.TransactionCoordinator)
[2020-11-03 03:30:59,861] INFO [TransactionCoordinator id=0] Startup complete. (kafka.coordinator.transaction.TransactionCoordinator)
[2020-11-03 03:30:59,874] INFO [Transaction Marker Channel Manager 0]: Starting (kafka.coordinator.transaction.TransactionMarkerChannelManager)
[2020-11-03 03:31:00,138] INFO Creating /brokers/ids/0 (is it secure? false) (kafka.utils.ZKCheckedEphemeral)
[2020-11-03 03:31:00,179] INFO Result of znode creation is: OK (kafka.utils.ZKCheckedEphemeral)
[2020-11-03 03:31:00,249] INFO Registered broker 0 at path /brokers/ids/0 with addresses: EndPoint(kafka,9092,ListenerName(PLAINTEXT),PLAINTEXT) (kafka.utils.ZkUtils)
[2020-11-03 03:31:00,267] WARN No meta.properties file under dir /var/dabing/kafka/kafka-logs/meta.properties (kafka.server.BrokerMetadataCheckpoint)
[2020-11-03 03:31:00,421] INFO [SocketServer brokerId=0] Started processors for 1 acceptors (kafka.network.SocketServer)
[2020-11-03 03:31:00,439] INFO Kafka version : 1.0.2 (org.apache.kafka.common.utils.AppInfoParser)
[2020-11-03 03:31:00,439] INFO Kafka commitId : 2a121f7b1d402825 (org.apache.kafka.common.utils.AppInfoParser)
[2020-11-03 03:31:00,440] INFO [KafkaServer id=0] started (kafka.server.KafkaServer)
5、重新开一个窗口，查看Zookeeper的节点

6、此时Kafka是前台模式启动，要停止，使用Ctrl+C

如果要后台启动，使用命令：

 kafka-server-start.sh -daemon $KAFKA/config/server.properties

查看Kafka的后台进程：

 ps aux | grep kafka

停止后台运行的Kafka：

二、生产与消费

查看zookeeper状态，zookeeper启动成功，再启动kafka

1、kafka-topics.sh 用于管理主题
查看命令的帮助信息
 [root@kafka bin]# kafka-topics.shCreate, 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.

 # 列出现有的主题[root@node1 ~]# kafka-topics.sh --list --zookeeper localhost:2181/myKafka# 创建主题，该主题包含一个分区，该分区为Leader分区，它没有Follower分区副本。[root@node1 ~]# kafka-topics.sh --zookeeper localhost:2181/myKafka --create --topic topic_test --partitions 1   --replication-factor 1# 查看分区信息[root@node1 ~]# kafka-topics.sh --zookeeper localhost:2181/myKafka --list# 查看指定主题的详细信息[root@node1 ~]# kafka-topics.sh --zookeeper localhost:2181/myKafka --describe --topic topic_test # 删除指定主题[root@node1 ~]# kafka-topics.sh --zookeeper localhost:2181/myKafka --delete --topic topic_test
列出现有主题，创建主题，该主题包含一个分区，该分区为Leader分区，它没有Follower分区副本

查看指定主题的详细信息

创建主题，该主题包含多个分区
 多个分区：横向扩展多个副本：高可用

2、kafka-console-consumer.sh用于消费消息：
# 开启消费者
[root@node1 ~]# kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topic_test
# 开启消费者方式二，从头消费，不按照偏移量消费
[root@node1 ~]# kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topic_test --from-beginning
3、kafka-console-producer.sh用于生产消息：
# 开启生产者
[root@node1 ~]# kafka-console-producer.sh --topic topic_test --broker-list  localhost:9020

具体操作

开启消费者和生产者，生产并消费消息

消费者，按照偏移量消费

消费者从头消费，不按照偏移量消费

通过程序访问链接：略

Kafka2.12安装与配置/生产与消费相关推荐

  1. mysql8.0.12插件_MySQL8.0.12 安装及配置

    MySQL8.0.12 安装及配置 发布时间:2018-08-07 10:39, 浏览次数:274 , 标签: MySQL 一.安装 1.从网上下载MySQL8.0.12版本,下载地址:https:/ ...

  2. sql配置管理器服务是空的_PostgreSQL 12 安装和配置

    点击上方晟数学院可以订阅哦! PostgreSQL 12 安装和配置  1 PostgreSQL 介绍 PostgreSQL是一个由加利福尼亚大学计算机科学部门基于Postgres 4.2版本开发的面 ...

  3. MySQL8.0.12 安装及配置

    一.安装 1.从网上下载MySQL8.0.12版本,下载地址:https://dev.mysql.com/downloads/mysql/ 2. 下载完成后解压 我解压的路径是:D:Javamysql ...

  4. mysql8.0.12怎么配置_MySQL8.0.12安装及配置

    一.下载 下载页面http://dev.mysql.com/downloads/mysql/ 选择系统平台后,点击download(根据系统选择64或32位) 二.配置 1.下载成功后,解压安装包到要 ...

  5. 生产环境安装、配置、管理PostgreSQL14.5数据库集群。pgpool 4.3.3参数中文说明

    1 生产环境安装.配置.管理PostgreSQL14.5 欢迎观看我发布的视频 "生产环境安装.配置.管理PostgreSQL14.5数据库集群" 数据库软件不是简单的安装上就好, ...

  6. mysql 5.6.35anzhaung_MySQL安装与配置(5.6.12，参考乐沙弥博客，未完成)

    gcc/g++ : MySQL 5.6开始,需要使用g++进行编译.2 cmake : MySQL 5.5开始,使用cmake进行工程管理,cmake需要2.8以上版本.3 bison : MySQL ...

  7. 第12章 CentOS7 中 Docker 的安装与配置

    第08章 CentOS7 中 Docker 的安装与配置 Docker 是什么? Docker 是世界领先的软件容器平台.[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img- ...

  8. ESXi6.5环境搭建（一：VMware Workstations 12 Pro 环境的安装及配置）

    实验目的及要求 完成VMware workstations安装,会应用相关操作: 完成虚拟机中ESXI6.5平台的安装及网络环境配置: 完成VMware vSphere Client 6.0软件在PC ...

  9. Fedora 12 安装配置subversion

    Fedora 12 安装配置subversion svn服务器有2种运行方式:独立服务器和借助apache.2种方式各有利弊. svn存储版本数据也有2种方式:BDB和FSFS.因为BDB方式在服务器 ...

最新文章

  1. 两队选手每队5人进行一对一的比赛（算法）
  2. 检查Bash数组是否包含值
  3. 同质化软件“吃掉”创新 企业因何深陷罗网？
  4. 4000+系统，10w+服务的立体式监控是如何炼成的?
  5. boost::uuids::entropy_error相关的测试程序
  6. 日期/时间格式/解析，Java 8样式
  7. 神经网络与推荐系统初步简介
  8. Oracle基本安全之用户、角色和权限操作
  9. 阿里巴巴矢量图标库（Iconfont）-利于UI和前端的搭配
  10. leetcode笔记：Majority Element
  11. android动画的实现过程
  12. Python~FTP文件下载
  13. java教学视频全集迅雷，BAT大厂面试总结
  14. Tableau数据可视化案例
  15. Android自定义PreferenceScreen的Layout布局，并获取控件
  16. “玄姐”孙玄，特立独行的架构师
  17. 电子计算机与媒体阅读答案,电子计算机与多媒体课课练.docx
  18. 基于udp的服务器消息转发（群发）
  19. .NET Framework v2.0 Obsolete APIs - 与大家分享
  20. windows 无法对计算机进行启动到,windows无法启动无法启动怎么办，windows无法启动的9种解决方法...

热门文章

  1. HDU 5752 Sqrt Bo【枚举，大水题】
  2. IPsec   在企业网中的应用
  3. ListView和SlidingDrawer
  4. 部署一个Windows Server 2008 的只读域控制器
  5. 全面IT资源监控 构建新一代数字化校园——天津大学
  6. 通过meta进行重定向
  7. Bone collector//动态规划
  8. linux 无线命令
  9. 50个Android开发人员必备UI效果源码
  10. 职场七种最致命的想法