kafka单节点安装和测试

「爱情、让人受尽委屈。」 2022-07-20 15:03 227阅读 0赞

1. 简介

i. 消息的发布（publish）称作producer，消息的订阅（subscribe）称作consumer，中间的存储阵列称作broker。

ii. 多个broker协同合作，producer、consumer和broker三者之间通过zookeeper来协调请求和转发。

iii. producer产生和推送(push)数据到broker，consumer从broker拉取(pull)数据并进行处理。

iv. broker端不维护数据的消费状态，提升了性能。

v. 直接使用磁盘进行存储，线性读写，速度快：避免了数据在JVM内存和系统内存之间的复制，减少耗性能的创建对象和垃圾回收。

vi. Kafka使用scala编写，可以运行在JVM上。

注意这里的ogg只支持java-1.7.0，

2. 安装：

先卸载老版本Java

\[ [root@oracle][root_oracle] bin\]\# yum remove java

首先安装JRE/JDK

\[root@oracle bin\]\# yum -y install java-1.7.0-openjdk\*

[root@bd-qa-vertica-90][root_bd-qa-vertica-90]:/opt/app/kafka/kafka\_2.10-0.8.2.0\#javac -version

javac 1.7.0\_79

[root@bd-qa-vertica-90][root_bd-qa-vertica-90]:/opt/app/kafka/kafka\_2.10-0.8.2.0\#java -version

java version "1.7.0\_79"

OpenJDK Runtime Environment (rhel-2.5.5.4.el6-x86\_64 u79-b14)

OpenJDK 64-Bit Server VM (build 24.79-b02, mixed mode)

Step 1:  下载kafka

\[ [root@oracle][root_oracle] ~\]\# mkdir -p /opt/app/kafka

\[ [root@oracle][root_oracle] ~\]\# cd /opt/app/kafka/

\[ [root@oracle][root_oracle] kafka\]\# tar xzf kafka\_2.11-0.10.0.0.tgz

[root@bd-qa-vertica-90][root_bd-qa-vertica-90]:/opt/app/kafka\#ls

kafka\_2.10-0.8.2.0  kafka\_2.10-0.8.2.0.tgz

[root@bd-qa-vertica-90][root_bd-qa-vertica-90]:/opt/app/kafka/kafka\_2.10-0.8.2.0\#ls

bin  config  libs  LICENSE  NOTICE

\[ [root@oracle][root_oracle] kafka-0.10.0.0-src\]\# cd bin && ls

connect-distributed.sh     kafka-consumer-offset-checker.sh     kafka-replica-verification.sh   kafka-verifiable-producer.sh

connect-standalone.sh      kafka-consumer-perf-test.sh          kafka-run-class.sh              windows

kafka-acls.sh              kafka-mirror-maker.sh                kafka-server-start.sh           zookeeper-security-migration.sh

kafka-configs.sh           kafka-preferred-replica-election.sh  kafka-server-stop.sh            zookeeper-server-start.sh

kafka-console-consumer.sh  kafka-producer-perf-test.sh          kafka-simple-consumer-shell.sh  zookeeper-server-stop.sh

kafka-console-producer.sh  kafka-reassign-partitions.sh         kafka-topics.sh                 zookeeper-shell.sh

kafka-consumer-groups.sh   kafka-replay-log-producer.sh         kafka-verifiable-consumer.sh

\[ [root@oracle][root_oracle] bin\]\# cd ../config/ && ls

connect-console-sink.properties    connect-file-sink.properties    connect-standalone.properties  producer.properties     zookeeper.properties

connect-console-source.properties  connect-file-source.properties  consumer.properties            server.properties

connect-distributed.properties     connect-log4j.properties        log4j.properties               tools-log4j.properties

[root@bd-qa-vertica-90][root_bd-qa-vertica-90]:/opt/app/kafka/kafka\_2.10-0.8.2.0/config\#cd ../libs/ && ls

jopt-simple-3.2.jar              kafka-clients-0.8.2.0.jar  slf4j-log4j12-1.6.1.jar

kafka\_2.10-0.8.2.0.jar           log4j-1.2.16.jar           snappy-java-1.1.1.6.jar

kafka\_2.10-0.8.2.0-javadoc.jar   lz4-1.2.0.jar              zkclient-0.3.jar

kafka\_2.10-0.8.2.0-scaladoc.jar  metrics-core-2.2.0.jar     zookeeper-3.4.6.jar

kafka\_2.10-0.8.2.0-sources.jar   scala-library-2.10.4.jar

kafka\_2.10-0.8.2.0-test.jar      slf4j-api-1.7.6.jar

Step 2: 启动服务

首先必须启动zookeeper，这里使用kafka自带脚本启动

[root@bd-qa-vertica-90][root_bd-qa-vertica-90]:/opt/app/kafka/kafka\_2.10-0.8.2.0\#bin/zookeeper-server-start.sh config/zookeeper.properties >> /opt/app/kafka/log/zookeeper.log &

启动kafka服务

[root@bd-qa-vertica-90][root_bd-qa-vertica-90]:/opt/app/kafka/kafka\_2.10-0.8.2.0\#bin/kafka-server-start.sh config/server.properties >> /opt/app/kafka/log/kafka.log &

停止Kafka server

bin/kafka-server-stop.sh

停止Zookeeper server:

bin/zookeeper-server-stop.sh

Step 3: 创建标题

创建一个名为“测试”的主题，有一个分区，只有一个副本：

[root@bd-qa-vertica-90][root_bd-qa-vertica-90]:/opt/app/kafka/kafka\_2.10-0.8.2.0\#bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic fafa01 >> /opt/app/kafka/log/topic.log &

如果我们运行列表主题命令，我们现在可以看到这个主题：

[root@bd-qa-vertica-90][root_bd-qa-vertica-90]:/opt/app/kafka/kafka\_2.10-0.8.2.0\#bin/kafka-topics.sh --list --zookeeper localhost:2181

fafa

test

Step 4: 发送消息

[root@bd-qa-vertica-90][root_bd-qa-vertica-90]:/opt/app/kafka/kafka\_2.10-0.8.2.0\#bin/kafka-console-producer.sh --broker-list localhost:9092  --topic fafa01

\[2016-07-05 11:04:21,620\] WARN Property topic is not valid (kafka.utils.VerifiableProperties)

fafa

dada

Step 5: 开始消费

[root@bd-qa-vertica-90][root_bd-qa-vertica-90]:/opt/app/kafka/kafka\_2.10-0.8.2.0\#bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic fafa01 --from-beginning

fafa

dada

在server.properties里面有如下配置，记录kafka日志存储路径，例如这里的路径是/tmp/kafka-logs

[root@bd-qa-vertica-90][root_bd-qa-vertica-90]:/opt/app/kafka/kafka\_2.10-0.8.2.0/config\#cd /tmp/kafka-logs

[root@bd-qa-vertica-90][root_bd-qa-vertica-90]:/tmp/kafka-logs\#ls

fafa-0  fafa01-0  recovery-point-offset-checkpoint  replication-offset-checkpoint  test-0

[root@bd-qa-vertica-90][root_bd-qa-vertica-90]:/tmp/kafka-logs\#cd fafa01-0/

这里fafa01-0是topic

[root@bd-qa-vertica-90][root_bd-qa-vertica-90]:/tmp/kafka-logs\#cd fafa01-0/

[root@bd-qa-vertica-90][root_bd-qa-vertica-90]:/tmp/kafka-logs/fafa01-0\#ls

00000000000000000000.index  00000000000000000000.log

这里信息记录在二进制的00000000000000000000.log中