《Kafka深度分析.doc》由会员分享，可在线阅读，更多相关《Kafka深度分析.doc（48页珍藏版）》请在金锄头文库上搜索。
1、Kafka深度分析架构kafka是显式分布式架构，producer、broker（Kafka）和consumer都可以有多个。Kafka的运行依赖于ZooKeeper，Producer推送消息给kafka，Consumer从kafka拉消息。kafka关键技术点（1） zero-copy在Kafka上，有两个原因可能导致低效：1）太多的网络请求 2）过多的字节拷贝。为了提高效率，Kafka把message分成一组一组的，每次请求会把一组message发给相应的consumer。 此外， 为了减少字节拷贝，采用了sendfile系统调用。为了理解sendfile原理，先说一下传统的利用socke
2、t发送文件要进行拷贝：Sendfile系统调用：（2）Exactly once message transfer怎样记录每个consumer处理的信息的状态？在Kafka中仅保存了每个consumer已经处理数据的offset。这样有两个好处：1）保存的数据量少 2）当consumer出错时，重新启动consumer处理数据时，只需从最近的offset开始处理数据即可。（3）Push/pullProducer 向Kafka（push）推数据，consumer 从kafka 拉（pull）数据。（4）负载均衡和容错Producer和broker之间没有负载均衡机制。broker和consumer
3、之间利用zookeeper进行负载均衡。所有broker和consumer都会在zookeeper中进行注册，且zookeeper会保存他们的一些元数据信息。如果某个broker和consumer发生了变化，所有其他的broker和consumer都会得到通知。kafka术语TopicTopic,是KAFKA对消息分类的依据;一条消息,必须有一个与之对应的Topic;比如现在又两个Topic,分别是TopicA和TopicB,Producer向TopicA发送一个消息messageA,然后向TopicB发送一个消息messaeB;那么,订阅TopicA的Consumer就会收到消息messag
4、eA,订阅TopicB的Consumer就会收到消息messaeB;(每个Consumer可以同时订阅多个Topic,也即是说,同时订阅TopicA和TopicB的Consumer可以收到messageA和messaeB)。同一个Group id的consumers在同一个Topic的同一条消息只能被一个consumer消费，实现了点对点模式，不同Group id的Consumers在同一个Topic上的同一条消息可以同时消费到，则实现了发布订阅模式。通过Consumer的Group id实现了JMS的消息模式MessageMessage就是消息,是KAfKA操作的对象,消息是按照Topic存
5、储的;KAFKA中按照一定的期限保存着所有发布过的Message,不管这些Message是否被消费过;例如这些Message的保存期限被这只为两天,那么一条Message从发布开始的两天时间内是可用的,超过保存期限的消息会被清空以释放存储空间。消息都是以字节数组进行网络传递。Partition每一个Topic可以有多个Partition,这样做是为了提高KAFKA系统的并发能力，每个Partition中按照消息发送的顺序保存着Producer发来的消息,每个消息用ID标识,代表这个消息在改Partition中的偏移量,这样,知道了ID,就可以方便的定位一个消息了;每个新提交过来的消息,被追加到
6、Partition的尾部;如果一个Partition被写满了,就不再追加;(注意,KAFKA不保证不同Partition之间的消息有序保存)LeaderPartition中负责消息读写的节点;Leader是从Partition的节点中随机选取的。每个Partition都会在集中的其中一台服务器存在Leader。一个Topic如果有多个Partition，则会有多个Leader。ReplicationFactor一个Partition中复制数据的所有节点,包括已经挂了的;数量不会超过集群中broker的数量isrReplicationFactor的子集,存活的且和Leader保持同步的节点;Co
7、nsumerGroup传统的消息系统提供两种使用方式:队列和发布-订阅;队列：是一个池中有若干个Consumer,一条消息发出来以后,被其中的一个Consumer消费;发布-订阅：是一个消息被广播出去,之后被所有订阅该主题的Consumer消费;KAFKA提供的使用方式可以达到以上两种方式的效果:ConsumerGroup;每一个Consumer用ConsumerGroupName标识自己,当一条消息产生后,改消息被订阅了其Topic的ConsumerGroup收到,之后被这个ConsumerGroup中的一个Consumer消费;如果所有的Consumer都在同一个ConsumerGroup
8、中,那么这就和传统的队列形式的消息系统一样了;如果每一个Consumer都在一个不同的ConsumerGroup中,那么就和传统的发布-订阅的形式一样了;Offset消费者自己维护当前读取数据的offser，或者同步到zookeeper。mit.interval.ms 是consumer同步offset到zookeeper的时间间隔。这个值设置问题会影响到多线程consumer，重复读取的问题。安装启动配置环境安装下载kafka_2.11-0.8.2.1，并在linux上解压 tar -xzf kafka_2.11-0.8.2.1.tgz cd kafka_2.11-0.8.2.1/bin 可
9、用的命令如下:启动命令Kafka需要用到zookeeper,所有首先需要启动zookeeper。 ./zookeeper-server-start.sh ./config/zookeeper.properties &然后启动kafka服务 ./kafka-server-start.sh ./config/server.properties &创建Topic创建一个名字是”p2p”的topic，使用一个单独的partition和和一个replica ./kafka-topics.sh -create -zookeeper localhost:2181 -replication-factor 1
10、-partitions 1 -topic p2p使用命令查看topic ./kafka-topics.sh -list -zookeeper localhost:2181p2p除了使用命令创建Topic外，可以让kafka自动创建，在客户端使用的时候，指定一个不存在的topic，kafka会自动给创建topic，自动创建将不能自定义partition和relica。集群多broker将上述的单节点kafka扩展为3个节点的集群。从原始配置文件拷贝配置文件。 cp ./config/server.properties ./config/server-1.properties cp ./confi
11、g/server.properties ./config/server-2.properties修改配置文件。config/server-1.properties: broker.id=1 port=9093 log.dir=/tmp/kafka-logs-1 config/server-2.properties: broker.id=2 port=9094 log.dir=/tmp/kafka-logs-2注意在集群中broker.id是唯一的。现在在前面单一节点和zookeeper的基础上，再启动两个kafka节点。 ./kafka-server-start.sh ./config/ser
12、ver-1.properties & ./kafka-server-start.sh ./config/server-2.properties &创建一个新的topic，带三个ReplicationFactor ./kafka-topics.sh -create -zookeeper localhost:2181 -replication-factor 3 -partitions 1 -topic p2p-replicated-topic查看刚刚创建的topic。 ./kafka-topics.sh -describe -zookeeper localhost:2181 -topic p2p-
13、replicated-topicpartiton： partion id，由于此处只有一个partition，因此partition id 为0leader：当前负责读写的lead broker idrelicas：当前partition的所有replication broker listisr：relicas的子集，只包含出于活动状态的brokerTopic-Partition-Leader-ReplicationFactor 之间的关系样图以上创建了三个节点的kafka集群，在集群上又用命令创建三个topic，分别是：l replicated3-partitions3-topic：三份复制
14、三个partition的topicl replicated2-partitions3-topic：二份复制三个partition的topicl test:1份复制，一个partition的topic以我做测试创建的三个topic说明他们之间的关系。./kafka-topics.sh -describe -zookeeper localhost:2181 -topic replicated3-partitions3-topic./kafka-topics.sh -describe -zookeeper localhost:2181 -topic replicated2-partitions3-t
15、opic./kafka-topics.sh -describe -zookeeper localhost:2181 -topic test以kafka当前的描述画出以下关系图：从图上可以看到test没有备份，当broke Id 0 宕机后，虽然集群还有两个节点可以使用，但test这个topic却不能正常转发消息了。所以为了系统的可靠性，创建的replicas尽量的多，但却不能超过broker的数量。客户端使用APIProducer API从0.8.2版本开始，apache提供了新的java版本的Producer的API。这个java版本在测试中表现比之前的scala客户端性能要好。Pom获取java客户端： org.apache.kafka kafka-clients 0.8.2.1/
