消息队列之kafka(集群搭建)

Posted 2021-01-31

1.kafka集群搭建

  kafka安装包下载地址：
 官网网址：http://kafka.apache.org/quickstart
 中文官网：http://kafka.apachecn.org/quickstart.html
 在 windows 平台，从官网下载：http://mirrors.hust.edu.cn/apache/kafka/1.1.0/
 在 centos 平台：wgethttp://mirrors.hust.edu.cn/apache/kafka/1.1.0/kafka_2.11-1.1.0.tgz

(1)集群部署的基础环境准备：

 A： 安装 JDK 1.8
 A： 安装 zookeeper 集群（也可以使用自带 ZooKeeper，但是不推荐）

(2)集体搭建：

  版本：kafka_2.11-1.1.0
 集群规划：hadoop01、hadoop02、hadoop03 （三个节点）
① 解压安装包到对应的目录
 tar zxvfkafka_2.11-1.1.0.tgz -C /application/
② 修改配置文件
 [[email protected] ~]$ cd /application/kafka_2.11-1.1.0/config/
 [[email protected] ~]$ vim server.properties
  broker.id=5 ## 当前集群中的每个 broker 节点的一个唯一编号，每个节点都不一样

 
  listeners=PLAINTEXT://:9092
  listeners=PLAINTEXT://hadoop01:9092
  host.name=hadoop01## 每个节点指定为当前主机名，上面也是

 
  log.dirs=/home/hadoop/data/kafka-logs ## kafkabroke工作节点数据存储目录
  num.partitions=1 ## kafka 的 topic 的默认分区数

 
  log.retention.hours=168 ## 日志的最长保存时间

 
  zookeeper.connect=hadoop01:2181,hadoop02:2181,hadoop03:2181 ## zookeeper 地址

③ 批量发送
[[email protected] application]$scp -r /application/kafka_2.11-1.1.0/ hadoop02:$PWD
[[email protected] application]$scp -r /application/kafka_2.11-1.1.0/ hadoop03:$PWD
千万注意：要修改$KAFKA_HOME/config/server.properties 文件中的对应 broker 节点的信息

broker.id=your broker id 
host.name=your broker hostname 
advertised.listeners=PLAINTEXT:// your broker hostname:9092

④ 配置环境变量
[[email protected] application]$ sudo etc/profile
export KAFKA_HOME=/application/kafka_2.11-1.1.0
[[email protected] application]$source/etc/profile

⑤ 启动集群，进行验证(每一个节点都要启动）

nohup kafka-server-start.sh /application/kafka_2.11-1.1.0/config/server.properties 1>~/logs/kafka_std.log 2>~/logs/kafka_err.log &

2.kafka相关API

（1）启动集群每个节点的进程

nohup kafka-server-start.sh  /home/hadoop/apps/kafka_2.11-1.1.0/config/server.properties 1>~/logs/kafka_std.log  2>~/logs/kafka_err.log &

（2） 创建 topic

kafka-topics.sh --create --zookeeper hadoop01:2181,hadoop02:2181,hadoop03:2181 --replication-factor 1 --partitions 1 --topic kafka_test
**参数介绍**
--create 创建 kafka topic
--zookeeper 指定 kafka 的 zookeeper 地址
--partitions 指定分区的个数
--replication-factor 指定每个分区的副本个数

（3） 查看已经创建的所有 kafka topic

kafka-topics.sh --zookeeper hadoop01:2181,hadoop02:2181,hadoop03:2181 --list 

（4） 查看某个指定的 kafka topic 的详细信息

kafka-topics.sh --zookeeper hadoop01:2181,hadoop02:2181,hadoop03:2181 --describe    #查看详信息
--topic kafka_test  #指定需要查看的topic

Topic：topic的名称
Partition：topic的分区编号
Leader：负责处理消息和读写，leader是从所有节点中随机选出
Replicas：列出了所有的副本节点，不管节点是否在服务中。
isr：正在服务中的节点。
（5） 开启生产者模拟生成数据:

kafka-console-producer.sh --broker-list hadoop01:9092   # broker的节点列表
--topic kafka_test

（6） 开启消费者模拟消费数据：

kafka-console-consumer.sh --zookeeper hadoop01:2181,hadoop02:2181,hadoop03:2181 --from-beginning   #从哪里开始消费
--topic kafka_test

（7） 查看某 topic 某个分区的偏移量最大值和最小值

kafka-run-class.sh kafka.tools.GetOffsetShell --topic kafka_test --time -1 --broker-list hadoop01:9092 --partitions 1

（8） 增加 topic 分区数（这个操作是不被允许的）

kafka-topics.sh --alter --zookeeper hadoop01:2181,hadoop02:2181,hadoop03:2181 --topic kafka_test --partitions 5 /
--replication-factor 2

（9） 删除 Topic

kafka-topics.sh --delete --zookeeper hadoop01:2181,hadoop02:2181,hadoop03:2181 --topic kafka_test