kafka 是日志聚合器和发布/订阅信息系统的交叉。kafka表面上看是一个消息传递系统，但在后台实际是一个日志聚合器。数据在一个主题下以语义的方式分组，它是生产者编写的队列，也是消费者读取的队列。每个主题都划分固定的分区，并存储在代理机器上，其中每个分区时一个独立的只能追加提交日志。
zookeeper 与 kafka之间的关系
zookeeper负责调度
这一段时间也在研究scala，第一次接触这种语言，感觉难度很大
kafka 常用命令
# sh kafka-server-start.sh ../config/server.properties # 启动
# sh kafka-topics.sh --list --zookeeper 192.168.56.101:2181 #查看主题
# bin/kafka-topics.sh --create --topic topicname --replication-factor 2 --partitions 1 --zookeeper localhost:2181
--topic 指定topic名字
--replication-factor 指定副本数，因为我的是集群环境，这里副本数就为2
--partitions 指定分区数，这个参数需要根据broker数和数据量决定，正常情况下，每个broker上两个partition最好
### 查看topic
bin/kafka-topics.sh --list --zookeeper localhost:2181
### 启动producer
sh kafka-console-producer.sh --broker-list master:9092 --topic music
package com.sparktest.bigdata.spark
/*spark 从 kafka中读取消息，spark streaming 和 kafka整合*/
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.kafka.KafkaUtils
object Driver3 {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setMaster("local[5]").setAppName("kafka")
// local必须大于2，只启动一个线程，kafka无法消费消息
val sc=new SparkContext(conf)
val ssc=new StreamingContext(sc,Seconds(5))
//zookeeper的地址
val zkHost="192.168.56.101:2181,192.168.56.102:2181,192.168.56.104:2181"
//定义消费者组名，自定义
val group="gp1"
// key music value 1 (消费者的线程数，应该小于当前主题的分区数)
val topics=Map("music"->1)
//工具类从 kafka 获取数据
// 获取数据
val stream=KafkaUtils.createStream(ssc,zkHost,group,topics,StorageLevel.MEMORY_ONLY_SER).map(x=>x._2)
val result=stream.flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_)
result.print
ssc.start
ssc.awaitTermination()
}
}
kafka对外使用topic的概念，生产者往topic里写消息，消费者从读消息。为了做到水平扩展，一个topic实际是由多个partition组成的，遇到瓶颈时，可以通过增加partition的数量来进行横向扩容。单个parition内是保证消息有序。
每新写一条消息，kafka就是在对应的文件append写，所以性能非常高。
kafka的总体数据流：
kafka.png
Producers往Brokers里面的指定Topic中写消息，Consumers从Brokers里面拉去指定Topic的消息，然后进行业务处理。
图中有两个topic，topic 0有两个partition，topic 1有一个partition，三副本备份。可以看到consumer gourp 1中的consumer 2没有分到partition处理，这是有可能出现的，下面会讲到。
关于broker、topics、partitions的一些元信息用zookeeper来存，监控和路由啥的也都会用到zookeeper。
broker
Kafka 集群包含一个或多个服务器，服务器节点称为broker。
broker存储topic的数据。如果某topic有N个partition，集群有N个broker，那么每个broker存储该topic的一个partition。
如果某topic有N个partition，集群有(N+M)个broker，那么其中有N个broker存储该topic的一个partition，剩下的M个broker不存储该topic的partition数据。
如果某topic有N个partition，集群中broker数目少于N个，那么一个broker存储该topic的一个或多个partition。在实际生产环境中，尽量避免这种情况的发生，这种情况容易导致Kafka集群数据不均衡。
Topic
每条发布到Kafka集群的消息都有一个类别，这个类别被称为Topic。（物理上不同Topic的消息分开存储，逻辑上一个Topic的消息虽然保存于一个或多个broker上但用户只需指定消息的Topic即可生产或消费数据而不必关心数据存于何处）
类似于数据库的表名
Partition
topic中的数据分割为一个或多个partition。每个topic至少有一个partition。每个partition中的数据使用多个segment文件存储。partition中的数据是有序的，不同partition间的数据丢失了数据的顺序。如果topic有多个partition，消费数据时就不能保证数据的顺序。在需要严格保证消息的消费顺序的场景下，需要将partition数目设为1。
Producer
生产者即数据的发布者，该角色将消息发布到Kafka的topic中。broker接收到生产者发送的消息后，broker将该消息追加到当前用于追加数据的segment文件中。生产者发送的消息，存储到一个partition中，生产者也可以指定数据存储的partition。
Consumer
消费者可以从broker中读取数据。消费者可以消费多个topic中的数据。
Consumer Group
每个Consumer属于一个特定的Consumer Group（可为每个Consumer指定group name，若不指定group name则属于默认的group）。
Leader
每个partition有多个副本，其中有且仅有一个作为Leader，Leader是当前负责数据的读写的partition。
Follower
Follower跟随Leader，所有写请求都通过Leader路由，数据变更会广播给所有Follower，Follower与Leader保持数据同步。如果Leader失效，则从Follower中选举出一个新的Leader。当Follower与Leader挂掉、卡住或者同步太慢，leader会把这个follower从“in sync replicas”（ISR）列表中删除，重新创建一个Follower。
网友评论