一、环境准备
jdk 下载地址链接：jdk 1.8，提取码: dv5h
zookeeper 下载地址链接：zookeeper3.4.14 ，提取码: 3dch
kafka 下载地址链接：kafka2.12，提取码: 61bc
1.1 Java 环境为前提
1.1.1 上传jdk-8u261-linux-x64.rpm到服务器并安装
# 安装命令rpm -ivh jdk-8u261-linux-x64.rpm
复制代码
1.1.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
复制代码
至此，jdk 安装成功。
1.2 Zookeeper 的安装配置
1.2.1 上传zookeeper-3.4.14.tar.gz到服务器，解压到/opt
# 解压zk到指定目录tar -zxf zookeeper-3.4.14.tar.gz -C /opt
复制代码
1.2.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/riemann/zookeeper/data
复制代码
1.2.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/riemann/zookeeper/log
复制代码
配置完以后再生效配置：
1.2.4 启动Zookeeper，确认Zookeeper的状态
至此，zookeeper 安装成功。
1.3 Kafka 的安装与配置
1.3.1 上传kafka_2.12-1.0.2.tgz到服务器并解压
tar -zxf kafka_2.12-1.0.2.tgz -C /opt
复制代码
1.3.2 配置环境变量并生效
export KAFKA=/opt/kafka_2.12-1.0.2export PATH=$PATH:$KAFKA/bin
复制代码
1.3.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 存储持久化数据目录
log.dirs=/var/riemann/kafka/kafka-logs
复制代码
创建上述持久化数据目录
mkdir -p /var/riemann/kafka/kafka-logs
复制代码
1.4 启动 Kafka
进入 Kafka 安装的根目录，执行如下命令：
kafka-server-start.sh ../config/server.properties
复制代码
启动成功，可以看到控制台输出的最后一行的 started 状态：此时 kafka 安装成功。
1.5 重新开一个窗口，查看 Zookeeper 的节点
1.6 此时 Kafka 是前台模式启动，要停止，使用 Ctrl+C
如果要后台启动，使用命令：
kafka-server-start.sh -daemon config/server.properties
复制代码
查看 Kafka 的后台进程：
停止后台运行的 Kafka：
二、生产与消费
查看 zookeeper 状态，zookeeper 启动成功，再启动 kafka。
2.1 kafka-topics.sh 用于管理主题
查看命令的帮助信息
[root@master1 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@master1 bin]#
复制代码
# 列出现有的主题[root@master1 ~]# kafka-topics.sh --list --zookeeper localhost:2181/myKafka# 创建主题，该主题包含一个分区，该分区为Leader分区，它没有Follower分区副本。[root@master1 ~]# kafka-topics.sh --zookeeper localhost:2181/myKafka --create --topic topic_test --partitions 1 --replication-factor 1# 查看分区信息[root@master1 ~]# kafka-topics.sh --zookeeper localhost:2181/myKafka --list# 查看指定主题的详细信息[root@master1 ~]# kafka-topics.sh --zookeeper localhost:2181/myKafka --describe --topic topic_test # 删除指定主题[root@master1 ~]# kafka-topics.sh --zookeeper localhost:2181/myKafka --delete --topic topic_test
复制代码
列出现有主题，创建主题，该主题包含一个分区，该分区为 Leader 分区，它没有 Follower 分区副本。
查看指定主题的详细信息
创建主题，该主题包含多个分区
多个分区：横向扩展多个副本：高可用
2.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
复制代码
2.3 kafka-console-producer.sh 用于生产消息
# 开启生产者[root@node1 ~]# kafka-console-producer.sh --topic topic_test --broker-list localhost:9020
复制代码
2.4 具体操作
开启消费者和生产者，生产并消费消息。
消费者，按照偏移量消费
消费者从头消费，不按照偏移量消费
评论