kafka

一、kafka简介

Kafka 被称为下一代分布式消息系统，是非营利性组织 ASF(Apache Software Foundation，简称为 ASF)基金会中的一个开源项目，比如 HTTP Server、Hadoop、ActiveMQ、Tomcat 等开源软件都属于 Apache 基金会的开源软件，类似的消息系统还有 RbbitMQ、ActiveMQ、ZeroMQ。

Kafka®用于构建实时数据管道和流应用程序。它具有水平可伸缩性，容错性，快速快速性，可在数千家公司中投入生产。


kafka 最主要的优势是其具备分布式功能、并可以结合 zookeeper 可以实现动态扩容，Kafka 是一种高吞吐量的分布式发布订阅消息系统

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

kafka 角色： 
Broker：Kafka 集群包含一个或多个服务器，这种服务器被称为 broker。
Topic ：每条发布到 Kafka 集群的消息都有一个类别，这个类别被称为 topic，（物理上不同 topic 的消息分开存储在不同的文件夹，逻辑上一个 topic 的消息虽然保存于一个或多个 broker 上但用户只需指定消息的 topic 即可生产或消费数据而不必关心数据存于何处）。
Partition ：是物理上的概念，每个 topic 包含一个或多个 partition，创建 topic 时可指定 parition 数量，每个 partition 对应于一个文件夹，该文件夹下存储该partition 的数据和索引文件。
Producer：负责发布消息到 Kafka broker。
Consumer：消费消息，每个 consumer 属于一个特定的 consuer group（可为每个consumer 指定 group name，若不指定 group name 则属于默认的 group），使用consumer high level API 时，同一 topic 的一条消息只能被同一个 consumer group内的一个 consumer 消费，但多个 consumer group 可同时消费这一消息。

版本选择：
http://kafka.apache.org/downloads
版本格式 如：kafka_2.12-2.3.1： #kafka_scala 版本_kafka 版本，scala 版本 2.12   kafka 版本2.3.1
http://distfiles.macports.org/ # scala
https://baike.baidu.com/item/Scala/2462287?fr=aladdin #scale 与 java

二、zookeeper实现高可用集群

1.安装zookeeper.参考地址：https://www.cnblogs.com/Yuanbangchen/p/17056168.html

2.解压kafaka
 [root@localhost7h local]#ln -sv /usr/local/src/kafka_2.12-2.3.1 /usr/local/kafka
'/usr/local/kafka' -> '/usr/local/src/kafka_2.12-2.3.1'

[root@localhost7h local]## cd /usr/local/kafka/config/

3.设置配置文件
[root@localhost7h local]# cat  kafka/config/server.properties
21 broker.id=1         #每个 broker 在集群中的唯一标识，正整数。
31 listeners=PLAINTEXT://192.168.80.170:9092     #监听地址，地址不能是0.0.0.0
60 log.dirs=/usr/local/kafka/kafka-logs         #kakfa 用于保存数据的目录，所有的消息都会存储在该目录当中，自动创建。
65 num.partitions=1                             #设置创新新的 topic 默认分区数量
103 log.retention.hours=168                        #设置 kafka 中消息保存时间，默认为 168 小时即 7天
123 zookeeper.connect=192.168.80.170:2181,192.168.80.180:2181,192.168.80.190:2181  #zookeeper.connect 指定连接的 zk 的地址,zk 中存储了 broker 的元数据信息
126 zookeeper.connection.timeout.ms=6000 #设置连接 zookeeper 的超时时间，默认 6 秒钟

4.启动 kafka：
#此方式 zookeeper 会在 shell 断开后关闭
[root@localhost7h local]# /usr/local/kafka/bin/kafka-server-start.sh /usr/local/kafka/config/server.properties &

#以守护进程启动
[root@localhost7h local]# /usr/local/kafka/bin/kafka-server-start.sh -daemon  /usr/local/kafka/config/server.properties  

5.kafka 元数据说明
1、Broker 依赖于 Zookeeper，每个 Broker 的 id 和 Topic、Partition 这些元数据信息都会写入 Zookeeper 的 ZNode 节点中；
2、Consumer 依赖于 Zookeeper，Consumer 在消费消息时，每消费完一条消息，会将产生的 offset 保存到 Zookeeper 中，下次消费在当前 offset 往后继续消费；
    ps：kafka0.9 之前 Consumer 的 offset 存储在 Zookeeper 中，kafka0,9 以后 offset存储在本地。
3、Partition 依赖于 Zookeeper，Partition 完成 Replication 备份后，选举出一个Leader，这个是依托于 Zookeeper 的选举机制实现的；



6.测试

在任意 kafka 服务器操作，创建 topic：
连接zookeeper的方式：创建名为 magedu ，partitions(分区)为 1，replication(复制)为 1 的 topic(主题)：
[root@localhost7h local]# /usr/local/kafka/bin/kafka-topics.sh --create --zookeeper 192.168.80.180:2181 --partitions 1 --replication-factor 1 --topic magedu
连接kafka的方式：创建名为 AAAA ，partitions(分区)为 3，replication(复制)为 3 的 topic(主题)：
[root@localhost7h local]# /usr/local/kafka/bin/kafka-topics.sh --create --bootstrap-server 192.168.80.170:9092 --replication-factor 3 --partitions 3 --topic AAAA

获取所有 topic
[root@localhost7h local]# /usr/local/kafka/bin/kafka-topics.sh --list --zookeeper 192.168.80.180:2181
AAAA
magedu

[root@localhost7h local]# /usr/local/kafka/bin/kafka-topics.sh --describe --zookeeper 192.168.80.180:2181 --topic magedu
Topic:magedu PartitionCount:1 ReplicationFactor:1 Configs:
Topic: magedu Partition: 0 Leader: 1 Replicas: 1 Isr: 1

状态说明：AAAA 有三个分区分别为0、1、2，分区0的leader是3（broker.id），分区 0 有三个副本，并且状态都为 lsr（ln-sync，表示可以参加选举成为 leader）。
[root@localhost7h local]# /usr/local/kafka/bin/kafka-topics.sh --describe --zookeeper 192.168.80.180:2181 --topic AAAA
Topic:AAAA PartitionCount:3 ReplicationFactor:3 Configs:
Topic: AAAA Partition: 0 Leader: 3 Replicas: 3,1,2 Isr: 3,1,2
Topic: AAAA Partition: 1 Leader: 1 Replicas: 1,2,3 Isr: 1,2,3
Topic: AAAA Partition: 2 Leader: 2 Replicas: 2,3,1 Isr: 2,3,1

测试发送消息：
[root@localhost7h local]# ./kafka/bin/kafka-console-producer.sh --broker-list 192.168.80.170:9092,192.168.80.180:9092,192.168.80.190:9092 --topic magedu
>mesg1
>mesg3^H2
>^[[A
>mesg3
>aaa
>bbb

测试获取消息：可以到任意一台 kafka 服务器测试消息获取，只要有相应的消息获取客户端即可
./bin/kafka-console-consumer.sh --topic magedu --bootstrap-server 192.168.80.170:9092,192.168.80.180:9092,192.168.80.190:9092 --from-beginning
>mesg1
>mesg3^H2
>^[[A
>mesg3
>aaa
>bbb

删除 topic：
/usr/local/kafka/bin/kafka-topics.sh --delete --zookeeper 192.168.80.170:2181,192.168.80.180:2181,192.168.80.190:2181 --topic AAAA

[root@localhost7h local]# /usr/local/kafka/bin/kafka-topics.sh --list --zookeeper 192.168.80.180:2181
magedu

kafkatool工具查看