1.环境准备

集群规划

jar包下载

http://kafka.apache.org/downloads.html

安装jdk

安装Zookeeper

0）集群规划

在Master、Ubuntu 和CentOS 三个节点上部署Zookeeper。

1）解压安装

解压zookeeper 安装包到/usr/local/目录下

tar -zxvf zookeeper-3.4.10.tar.gz -C /usr/local/

重命名文件夹

sudo mv zookeeper-3.4.10 zookeeper

修改文件权限

sudo chown -Rhadoop ./ zookeeper

在zookeeper目录下创建data与datalog目录

mkdir -p data

mkdir -p datalog

重命名/usr/local/zookeeper/conf 这个目录下的zoo_sample.cfg 为zoo.cfg

mv zoo_sample.cfg zoo.cfg

2）配置zoo.cfg 文件

具体配置

dataDir=/usr/local/zookeeper/data

dataLogDir=/usr/local/zookeeper/datalog增加如下配置

#######################cluster##########################

server.1=0.0.0.0:2888:3888

server.2={Ubuntu公网ip}:2888:3888

server.3={CentOS公网ip}:2888:3888

配置参数解读

Server.A=B:C:D。

A 是一个数字，表示这个是第几号服务器；

B是这个服务器的ip地址；

C是这个服务器与集群中的Leader服务器交换信息的端口；

D是万一集群中的Leader服务器挂了，需要一个端口来重新进行选举，选出一个新的Leader，而这个端口就是用来执行选举时服务器相互通信的端口。

集群模式下配置一个文件myid，这个文件在dataDir目录下，这个文件里面有一个数据就是A的值，Zookeeper启动时读取此文件，拿到里面的数据与zoo.cfg里面的配置信息比较从而判断到底是哪个server。

注意：这里我用的三台公网ip互联，所以在ip配置时，自己server的ip配置为0.0.0.0，其他server为公网IP，如果用内网互联，则全部配置为内网ip即可。

3）集群操作

在/usr/local/zookeeper/data目录下创建一个myid的文件

touch myid

添加myid文件，注意一定要在linux里面创建，在windows里面编辑很可能乱码

编辑myid文件        

vi myid

在文件中添加与server对应的编号：如1

拷贝配置好的zookeeper到其他机器上

scp -r zookeeper/ hadoop@Ubuntu:/usr/local

scp -r zookeeper/ hadoop@CentOS:/usr/local

并分别修改myid文件中内容为2、3 同时修改zoo.cfg中cluster信息

分别启动zookeeper

/usr/local/zookeeper/bin/zkServer.sh start

查看状态

/usr/local/zookeeper/bin//zkServer.shstatus

由于这里是三个server，则应该是一个leader 两个follower

2 Kafka集群部署

1）解压安装包

tar -zxvf kafka_2.11-2.1.0.tgz -C /usr/local/

2）修改解压后的文件名称

sudo mv kafka_2.11-2.1.0/ kafka

3）修改文件权限

sudo chown -Rhadoop ./ kafka

4）在/usr/local/kafka目录下创建logs文件夹

mkdir logs

5）修改配置文件

cd config/

vi server.properties

输入以下内容：

#broker的全局唯一编号，不能重复

broker.id=0

#删除topic功能使能

delete.topic.enable=true

#处理网络请求的线程数量

num.network.threads=3

#用来处理磁盘IO的现成数量

num.io.threads=8

#发送套接字的缓冲区大小

socket.send.buffer.bytes=102400

#接收套接字的缓冲区大小

socket.receive.buffer.bytes=102400

#请求套接字的缓冲区大小

socket.request.max.bytes=104857600

#kafka运行日志存放的路径

log.dirs=/usr/local/kafka/logs

#topic在当前broker上的分区个数

num.partitions=1

#用来恢复和清理data下数据的线程数量

num.recovery.threads.per.data.dir=1

#segment文件保留的最长时间，超时将被删除

log.retention.hours=168

#配置连接Zookeeper集群地址

zookeeper.connect=Master:2181,Ubuntu:2181,CentOS:2181

6）配置环境

vi ~/.bashrc

# Kafka Environment Variable

export KAFKA_HOME=/usr/local/kafka

7）分别在hadoop103和hadoop104上修改配置文件/usr/local/kafka/config/server.properties中的broker.id=1、broker.id=2

注意：broker.id不得重复

8）启动集群

依次在Master、Ubuntu、CentOS节点上启动kafka

bin/kafka-server-start.shconfig/server.properties &

3 Kafka命令行操作

1）查看当前服务器中的所有topic

kafka-topics.sh --zookeepermaster:2181,ubuntu:2181,centos:2181--list

2）创建topic

kafka-topics.sh--create--zookeepermaster:2181,ubuntu:2181,centos:2181--replication-factor 3 --partitions 3--topic test

3）删除topic

kafka-topics.sh --zookeepermaster:2181,ubuntu:2181,centos:2181--delete--topic test

4）发送消息

kafka-console-producer.sh--broker-listmaster:9092,ubuntu:9092,centos:9092 --topic test

>hello world

>hello kafka

5）消费消息

kafka-console-consumer.sh--bootstrap-server master:9092,ubuntu:9092,centos:9092 --topic test

--from-beginning：会把first主题中以往所有的数据都读取出来。根据业务场景选择是否增加该配置。