【Kafka】基本使用：安装、集群搭建、命令操作（阿里云搭建踩坑记录）

1.Kafka安装部署

1.1 单机模式

1).下载kafka。

https://kafka.apache.org/downloads

2).安装解压。

已经编译好的，直接解压就行（tar -zxvf kafka_2.11-2.0.0.tgz ）

3).启动zookeeper。

因为kafka依赖于zookeeper来做master选举一起其他数据的维护，所以需要先启动zookeeper节点

kafka内置了zookeeper的服务，所以在bin目录下提供了zk的启动和关闭脚本zookeeper-server-start.sh和zookeeper-server-stop.sh。另外，在config目录下，存在一些配置文件zookeeper.properties和server.properties。

所以我们可以通过下面的脚本来启动kafka自带的zk服务：

sh zookeeper-server-start.sh -daemon ../config/zookeeper.properties

当然，也可以自己搭建zk的集群，然后正常启动。

4).修改server.properties。

增加zookeeper的配置，可以是上面自带的也可以是自己搭建的。(这里我使用的是云服务器与docker的zk）

zookeeper.connect=localhost:2181  # 注：若用的docker的zookeeper也是localhost

【重点】若使用的是云服务器搭建的kafka，那server.properties还要配置下面这两项

# kafka消息收发监听配置
listeners=PLAINTEXT://内网IP:端口   
advertised.listeners=PLAINTEXT://外网IP:端口  

上面说的内网ip，可以通过ifconfig命令查看，是eth0网卡（lo是local，本地回环接口）

5).启动kafka

nohup sh kafka-server-start.sh ../config/server.properties &

几点注意：

1. nohup和&表示窗口关闭仍后台运行，若不加，当退出当前命令时kafka也stop了
2. 通过查看进程ps-ef | grep kafka看是否成功启动
3. 若内存不够则修改 vim bin/kafka-server.start.sh的俩1G为256M和128M

export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G"     # Xms是启动内存，大点启动的快，但可能导致机器暂时变慢
export KAFKA_HEAP_OPTS="-Xmx256M -Xms128M" # Xmx是可占用最大内存，超出就会报oom异常

6). 停止kafka

sh kafka-server-stop.sh ../config/server.properties

1.2 集群模式

准备三台虚拟机，分别把kafka的安装包部署在三台机器上，分别修改server.properties配置内容

1). 分别修改三台机器的server.properties配置，同一个集群中的每个机器的id必须唯一

broker.id=0 
broker.id=1 
broker.id=2

2).修改zookeeper的连接配置

zookeeper.connect=192.168.13.106:2181

3).修改listeners配置

如果配置了listeners，那么消息生产者和消费者会使用listeners的配置来进行消息的收发，否则会使用localhost

listeners=PLAINTEXT://192.168.13.102:9092   #PLAINTEXT表示协议，默认是明文，可以选择其他加密协议

4). 分别启动三台服务器

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

2.Kafka基本操作

1.创建topic

sh kafka-topics.sh --create 
--zookeeper 39.105.136.112:2181 
--replication-factor 1  # 表示该topic需要在不同的broker中保存几份，这里设置成1，表示在两个broker中保存两份
-partitions 1           # 分区数 
--topic test

如果zk在本机的话，也可以直接 --zookeeper localhost:2181

2.查看topic

首先是查看kafka中有哪些topic：

sh kafka-topics.sh 
--zookeeper localhost:2181
--list   # 查看topic列表

查看指定topic属性信息：

sh kafka-topics.sh 
--zookeeper localhost:2181 
--describe    # 查看topic描述
--topic test  # 指定topic

3.产生消息

sh kafka-console-producer.sh 
--broker-list localhost:9092  # 指定发送消息的broker
--topic test # 指定发送消息的broker的topic

4.消费消息

生产者一产生消息，消费者就会同时收到

sh kafka-console-consumer.sh 
--bootstrap-server localhost:9092  # 从哪个broker消费消息
--topic test  	  # 从哪个topic消费消息
--from-beginning  # 从何处开始消费消息