Kafka研究

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原理，先说一下传统的利用socket发送文件要进行拷贝：Sendfile系统调用：（2）Exactlyoncemessagetransfer怎样记录每个consumer处理的信息的状态？在Kafka中仅保存了每个consumer已经处理数据的offset。这样有两个好处：1）保存的数据量少2）当consumer出错时，重新启动consumer处理数据时，只需从最近的offset开始处理数据即可。（3）Push/pullProducer向Kafka（push）推数据，consumer从kafka拉（pull）数据。（4）负载均衡和容错Producer和broker之间没有负载均衡机制。broker和consumer之间利用zookeeper进行负载均衡。所有broker和consumer都会在zookeeper中进行注册，且zookeeper会保存他们的一些元数据信息。如果某个broker和consumer发生了变化，所有其他的broker和consumer都会得到通知。kafka术语TopicTopic,是KAFKA对消息分类的依据;一条消息,必须有一个与之对应的Topic;比如现在又两个Topic,分别是TopicA和TopicB,Producer向TopicA发送一个消息messageA,然后向TopicB发送一个消息messaeB;那么,订阅TopicA的Consumer就会收到消息messageA,订阅TopicB的Consumer就会收到消息messaeB;(每个Consumer可以同时订阅多个Topic,也即是说,同时订阅TopicA和TopicB的Consumer可以收到messageA和messaeB)。同一个Groupid的consumers在同一个Topic的同一条消息只能被一个consumer消费，实现了点对点模式，不同Groupid的Consumers在同一个Topic上的同一条消息可以同时消费到，则实现了发布订阅模式。通过Consumer的Groupid实现了JMS的消息模式MessageMessage就是消息,是KAfKA操作的对象,消息是按照Topic存储的;KAFKA中按照一定的期限保存着所有发布过的Message,不管这些Message是否被消费过;例如这些Message的保存期限被这只为两天,那么一条Message从发布开始的两天时间内是可用的,超过保存期限的消息会被清空以释放存储空间。消息都是以字节数组进行网络传递。Partition每一个Topic可以有多个Partition,这样做是为了提高KAFKA系统的并发能力，每个Partition中按照消息发送的顺序保存着Producer发来的消息,每个消息用ID标识,代表这个消息在改Partition中的偏移量,这样,知道了ID,就可以方便的定位一个消息了;每个新提交过来的消息,被追加到Partition的尾部;如果一个Partition被写满了,就不再追加;(注意,KAFKA不保证不同Partition之间的消息有序保存)LeaderPartition中负责消息读写的节点;Leader是从Partition的节点中随机选取的。每个Partition都会在集中的其中一台服务器存在Leader。一个Topic如果有多个Partition，则会有多个Leader。ReplicationFactor一个Partition中复制数据的所有节点,包括已经挂了的;数量不会超过集群中broker的数量isrReplicationFactor的子集,存活的且和Leader保持同步的节点;ConsumerGroup传统的消息系统提供两种使用方式:队列和发布-订阅;队列：是一个池中有若干个Consumer,一条消息发出来以后,被其中的一个Consumer消费;发布-订阅：是一个消息被广播出去,之后被所有订阅该主题的Consumer消费;KAFKA提供的使用方式可以达到以上两种方式的效果:ConsumerGroup;每一个Consumer用ConsumerGroupName标识自己,当一条消息产生后,改消息被订阅了其Topic的ConsumerGroup收到,之后被这个ConsumerGroup中的一个Consumer消费;如果所有的Consumer都在同一个ConsumerGroup中,那么这就和传统的队列形式的消息系统一样了;如果每一个Consumer都在一个不同的ConsumerGroup中,那么就和传统的发布-订阅的形式一样了;Offset消费者自己维护当前读取数据的offser，或者同步到zookeeper。auto.commit.interval.ms是consumer同步offset到zookeeper的时间间隔。这个值设置问题会影响到多线程consumer，重复读取的问题。安装启动配置环境安装下载kafka_2.11-0.8.2.1，并在linux上解压tar-xzfkafka_2.11-0.8.2.1.tgzcdkafka_2.11-0.8.2.1/bin可用的命令如下:启动命令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--zookeeperlocalhost:2181--replication-factor1--partitions1--topicp2p使用命令查看topic./kafka-topics.sh--list--zookeeperlocalhost:2181p2p除了使用命令创建Topic外，可以让kafka自动创建，在客户端使用的时候，指定一个不存在的topic，kafka会自动给创建topic，自动创建将不能自定义partition和relica。集群多broker将上述的单节点kafka扩展为3个节点的集群。从原始配置文件拷贝配置文件。cp../config/server.properties../config/server-1.propertiescp../config/server.properties../config/server-2.properties修改配置文件。config/server-1.properties:broker.id=1port=9093log.dir=/tmp/kafka-logs-1config/server-2.properties:broker.id=2port=9094log.dir=/tmp/kafka-logs-2注意在集群中broker.id是唯一的。现在在前面单一节点和zookeeper的基础上，再启动两个kafka节点。./kafka-server-start.sh../config/server-1.properties&./kafka-server-start.sh../config/server-2.properties&创建一个新的topic，带三个ReplicationFactor./kafka-topics.sh--create--zookeeperlocalhost:2181--replication-factor3--partitions1--topicp2p-replicated-topic查看刚刚创建的topic。./kafka-topics.sh--describe--zookeeperlocalhost:2181--topicp2p-replicated-topicpartiton：partionid，由于此处只有一个partition，因此partitionid为0leader：当前负责读写的leadbrokeridrelicas：当前partition的所有replicationbrokerlistisr：relicas的子集，只包含出于活动状态的brokerTopic-Partition-Leader-ReplicationFactor之间的关系样图以上创建了三个节点的kafka集群，在集群上又用命令创建三个topic，分别是：replicated3-partitions3-topic：三份复制三个partition的topicreplicated2-partitions3-topic：二份复制三个partition的topictest:1份复制，一个partition的topic以我做测试创建的三个topic说明他们之间的关系。./kafka-topics.sh--describe--zookeeperlocalhost:2181--topicreplicated3-partitions3-topic./kafka-topics.sh--describe--zookeeperlocalhost:2181--topicreplicated2-partitions3-topic./kafka-topics.sh--describe--zookeeperlocalhost:2181--topictest以kafka当前的描述画出以下关系图：Partition0LeaderBrokerId0Topic：testBrokerId1BrokerId2Partition0LeaderTopic：replicated3-partitions3-topicPartition0ReplicasTopic：replicated3-partitions3-topicPartition0ReplicasTopic：replicated3-partitions3-topicPartition1ReplicasTopic：replicated3-partitions3-topicPartition1ReplicasTopic：replicated3-partitions3-topicPartition1LeaderTopic：replicated3-partitions3-topicPartition2LeaderTopic：replicated3-partitions3-topicPartition2ReplicasTopic：replicated3-partitions3-topicPartition2ReplicasTopic：replicated3-partitions3-topicPartition0LeaderTopic：replicated2-partitions2-topicPartition0ReplicasTopic：replicated2-partitions2-topicPartition1LeaderTopic：replicated2-partitions2-topicPartition1ReplicasTopic：replicated2-partitions2-topic从图上可以看到test没有备份，当brokeId0宕机后，虽