一 简介

Kafka是一种高吞吐量的分布式发布订阅消息系统，它可以处理消费者规模的网站中的所有动作流数据，其主要特性如下：

• 通过O(1)的磁盘数据结构提供消息的持久化，这种结构对于即使数以TB的消息存储也能够保持长时间的稳定性。
• 高吞吐量：即使是非常普通的硬件，Kafka也可以支持每秒数百万的消息。
• 支持通过Kafka服务器和消费机集群来分区消息。
• 支持Hadoop并行数据加载。

Kafka主要术语

• broker：Kafka集群中包含的服务器。
• topic：每条发布到Kafka集群的消息都有一个类别，这个类别被称为Topic。（物理上不同Topic的消息分开存储，逻辑上一个Topic的消息虽然保存于一个或多个broker上但用户只需指定消息的Topic即可生产或消费数据而不必关心数据存于何处）。
• partition：partition是物理上的概念，每个 topic 包含一个或多个 partition。
• replica：partition 的副本，保障 partition 的高可用。
• leader：replica 中的一个角色， producer 和 consumer 只跟 leader 交互。
• follower：replica 中的一个角色，从 leader 中复制数据。
• producer：消息生产者，发布消息到 Kafka集群的终端或服务。
• consumer：消息消费者，从Kafka集群中消费消息的终端或服务。
• consumer group：high-level consumer API 中，每个 consumer 都属于一个 consumer group，每条消息只能被consumer group 中的一个 consumer 消费，但可以被多个 consumer group 消费。

二 ZooKeeper单服务器配置伪集群

三 Kafka伪集群安装

Kafka下载地址：http:///kafka/2.1.1/(版本你随意)

（1）解压：

[root@localhost bin]# cd /usr/local/src [root@localhost src]# tar -zxvf kafka_2.11-1.0.0.tgz [root@localhost src]# mv kafka_2.11-1.0.0 /usr/local/

 

（2）修改配置文件：

[root@localhost src]# cd /usr/local/kafka_2.11-1.0.0

i）创建伪集群的三个Kafka节点：

[root@localhost kafka_2.11-1.0.0]# mkdir server1 server2 server3 [root@localhost kafka_2.11-1.0.0]# mkdir server1/logs server2/logs server3/logs

ii）分别修改server.properties配置文件：

[root@localhost kafka_2.11-1.0.0]# cp config/server.properties server1/

分别修改以下内容：

broker.id=0 listeners=PLAINTEXT://192.168.1.159:9092 log.dirs=/usr/local/kafka_2.11-1.0.0/server1/logs zookeeper.connect=192.168.1.159:2100,192.168.1.159:2101,192.168.1.159:2102

注：这里的“192.168.1.159”为当前服务器IP，后面的9092端口表示此Kafka节点的访问端口。另外，后面的2100、2101、2102端口表示ZooKeeper集群的访问端口

[root@localhost kafka_2.11-1.0.0]# cp config/server.properties server2/

分别修改以下内容：

broker.id=1 listeners=PLAINTEXT://192.168.1.159:9093 log.dirs=/usr/local/kafka_2.11-1.0.0/server2/logs zookeeper.connect=192.168.1.159:2100,192.168.1.159:2101,192.168.1.159:2102[root@localhost kafka_2.11-1.0.0]# cp config/server.properties server3/

分别修改以下内容：

broker.id=2 listeners=PLAINTEXT://192.168.1.159:9094 log.dirs=/usr/local/kafka_2.11-1.0.0/server3/logs zookeeper.connect=192.168.1.159:2100,192.168.1.159:2101,192.168.1.159:2102

（3）启动：

[root@localhost kafka_2.11-1.0.0]# cd /usr/local/kafka_2.11-1.0.0 [root@localhost kafka_2.11-1.0.0]# ./bin/ -daemon ./server1/server.properties [root@localhost kafka_2.11-1.0.0]# ./bin/ -daemon ./server2/server.properties [root@localhost kafka_2.11-1.0.0]# ./bin/ -daemon ./server3/server.properties

注：如何关闭启动起来的Kafka节点？

直接杀掉其进程即可，比如：

[root@localhost bin]# ps -ef | grep kafka [root@localhost bin]# kill -9 kafka进程号

（4）测试：

i）创建一个名为”topic-test”的topic，该topic切分为3份，每一份备份数为3：

[root@localhost kafka_2.11-1.0.0]# ./bin/kafka-topics.sh --create --zookeeper 192.168.1.159:2100,192.168.1.159:2101,192.168.1.159:2102 --replication-factor 3 --partitions 3 --topic topic-test

注:为topic创建分区时，–partitions(分区数)最好是broker数量的整数倍，这样才能使一个topic的分区均匀的分布在整个Kafka集群中。

ii）列出所有已创建的topic：

[root@localhost kafka_2.11-1.0.0]# ./bin/kafka-topics.sh --list --zookeeper 192.168.1.159:2100,192.168.1.159:2101,192.168.1.159:2102

iii）查看指定topic详情信息：

[root@localhost kafka_2.11-1.0.0]# ./bin/kafka-topics.sh --describe --zookeeper 192.168.1.159:2100,192.168.1.159:2101,192.168.1.159:2102 --topic topic-test

iv）修改topic参数，增加partition：

[root@localhost kafka_2.11-1.0.0]# ./bin/kafka-topics.sh --alter --zookeeper 192.168.1.159:2100,192.168.1.159:2101,192.168.1.159:2102 --topic topic-test --partitions 6

注：对于已经存在的topic，只能增加partition数量，而不能减少其数量

v）删除topic：

[root@localhost kafka_2.11-1.0.0]# ./bin/kafka-topics.sh --delete --zookeeper 192.168.1.159:2100,192.168.1.159:2101,192.168.1.159:2102 --topic topic-test

vi）模拟生产者发送消息：

[root@localhost kafka_2.11-1.0.0]# ./bin/ --broker-list 192.168.1.159:9092,192.168.1.159:9093,192.168.1.159:9094 --topic topic-test

vii）模拟消费者接收消息：

[root@localhost kafka_2.11-1.0.0]# ./bin/kafka-console-consumer.sh --bootstrap-server 192.168.1.159:9092,192.168.1.159:9093,192.168.1.159:9094 --topic topic-test --from-beginning

可以看到已经成功接收到前面的消息了，至此Kafka伪集群就全部安装结束了