kafka高性能：
1.高效使用磁盘
2.零拷贝
3.批处理和压缩
4.partition
5.ISR

1.1 顺序写磁盘，性能高于随机写
零拷贝(没有用户态参与的拷贝，只是在内核态进行的拷贝)：
1.传统模式下数据从文件传输到网络需要四次拷贝(2次系统，2次cpu)，4次上下文切换，2次系统调用
File.read(fileDesc,buf,len)
Socket.send(socket,but,len)
2.kafka中实现
通过NIO的transferTo/transferFrom调用系统级别的sendFile函数实现零拷贝，共2次数据拷贝和2次上下文切换，1次系统函数调用，消除了cpu数据拷贝
数据拷贝：
1.文件->系统态的read buff
2.文件描述符的拷贝descriptor
上下文切换：
用户态->内核态->用户态(next cycle)

批处理和压缩
producer和consumer均支持批量处理数据，从而减少网络开销
producer可将数据压缩后发送给broker，从而减少网络传输代价，目前支持snappy,gzip和LZ4压缩

partition:
1.通过partition实现了并行处理和水平扩展
2.partition是kafka并行处理的最小单元
3.不同kafka处于不同的broker中，充分利用多机资源
4.同一broker上的不同partition可处于不同的directory(磁盘目录)，如果节点上有多个disk driver(磁盘驱动)，可将不同的driver对应不同的
directory,从而使kafka利用多disk driver的磁盘优势


partition中的数据只有在被broker commit了才能被consumer消费，只有所有isr中的replica都ack了leader，broker才能commit这条消息

isr实现了可用性和一致性的动态平衡：
replica.lag.time.max.ms=10000,超过10S，follower没有想leader fetch数据(或fetch数据后，还没有和leader保持同步)，则会从isr中移除
Isr可容忍更多的节点失败：
majority quorum如果要容忍f个节点失败，则需要2f+1个节点
isr要容忍f个节点失败，则需要f+1个节点

启动：
./kafka-server-start.sh ../config/server2.properties &
./kafka-topics.sh --create --zookeeper localhost:2181,localhost:2182,localhost:2183 --replication-factor 3 --partitions 3 --topic topic1
创建消息：
./kafka-topics.sh --create --zookeeper localhost:2181,localhost:2182,localhost:2183 --replication-factor 3 --partitions 3 --topic topic1
./kafka-console-consumer.sh --bootstrap-service localhost:9091,localhost:9092,localhost:9093 --from-beginning --topic topic1 --new-consumer
./kafka-topics.sh --describe --zookeeper localhost:2181,localhost:2182,localhost:2183 --topic topic1 
删除topic
./kafka-topics.sh --zookeeper localhost:2181,localhost:2182,localhost:2183  --delete --topic topic1
#生产消息
./kafka-console-producer.sh --broker-list 127.0.0.1:9091,127.0.0.1:9092,127.0.0.1:9093 --topic toefl.preclass.correct.rate.topic.local
./kafka-console-consumer.sh --zookeeper localhost:2181/kafka --from-beginning --topic topic1 --group group1
#展示所有的topic
./kafka-topics.sh --list --zookeeper localhost:2181
./kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic test2
#指定test2从test1中消费消息
./kafka-replay-log-producer.sh --broker-list localhost:9092 --zookeeper localhost:2181 --inputtopic test1 --outputtopic test2

同步和异步比较
Sync producer:同步
低延迟
低吞吐率
不会丢失数据

Async producer:异步
高延迟
高吞吐率
可能丢失数据

consumer只能读取到被commmit的数据
isr的常见配置如下
Server配置：
replica.lag.time.max.ms=10000
replica.lag.max.message=4000
topic配置：
min.insync.replicas=1
Producer配置：
request.required.acks=0
0:不需要partition返回ack
1:需要partition返回ack
-1:isr中所有的follower返回ack，leader才会commit

如果处理Replica全部数据宕机
a.等待ISR中任一Replica恢复，并选举为Leader
1.等待时间叫较长，降低可用性
2.或isr中的所有数据都无法恢复或者数据丢失，则该partition将用不可用
b.选择第一个恢复的replica为新的leader,无论他是否在isr中
1.并为包含之前commit过的数据，因此会造成数据丢失
2.可用性较高

kafka中使用zookeeper
1.zk是什么？
.naming service
.配置管理
.leader election
.服务发现
.同步
.group service
.barrier
.分布式队列
.两阶段提交

.zookeeper集群包含一个leader和多个follower
.所有follower提供read服务
.所有写操作都会被forward到leader
.client和service通过nio通信
.全局串型化所有write操作
.保证同一客户端的指令被fifo执行
.保证消息通知的fifo

zab 广播协议：
.leader将所有更新(称为proposal)，顺序发送给follow
.当leader收到半数以上的的follow对此proposal的ack时，即向所有的follow发送commit消息，并在本地commit消息
.follow收到proposal即将commit写入磁盘，写入成功返回ack给leader
.每个proposal都有一个全局唯一单调递增的proposal id,即zxid

恢复模式：
进入恢复模式：当leader宕机或者丢失大多数的follow时
结束恢复模式：当leader被选举出来且大多数的follow与leader完成状态同步后，恢复模式结束，从而进入广播模式
恢复模式的意义：
1.发现集群中最大的zxid
2.建立新的epoch，从而保证新的leader不能再commit新的proposal
3.集群中大部分节点都commit过前一个leader commit过的消息，而新的leader被大多数节点支持，所以之前的leader commit的proposal不会丢失，至少被一个节点所保存
4.新leader与大多数节点通信，从而保证大部分节点具有新的数据

zxid：高32位，epoch 每一个leader的任期内，epoch不变，选举出新的leader后，epoch会增加
低32位 counter

zookeeper一致性的保证：
顺序一致性：从一个客户端发出的更新操作会按发送顺序，顺序执行
原子性 ：更新操作要不成功，要不失败，没有中间状态
单一系统镜像：一个客户端只能看到同一个view，无论连接到哪台服务器
可靠性：
1.一旦一个更新被应用，该更新将被持久化，直到有客户端更新该结果
2.如果一个客户端得到更新成功的状态码，该更新一定已经生效
3.任何一个被客户端通过读或者更新看到的结果，将不会回滚，即使从失败中恢复

实时性：保证客户端可以在一定时间内看到最新的视图

kafka中将topic存在在/config/topics路径下
get /config/topics/topic1

kafka的broker集群中leader(controller)的路径：
get /controller 可以看到当前的controller是哪个broker

broker的信息存放在/brokers/ids路径下面
get /brokers/ids/1 可以查看broker为1的具体信息

基于zk进行leader election
公平模式：
1.创建Leader父节点，如/chroot,并将其设置为persist
2.各客户端通过在/chroot下创建leader，如/chroot/leader来竞争leader，该节点被设置为ephemeral_sequential
3.客户端通过getChildren方法获取/chroot下所有的字节点，如果其注册的节点id在所有节点中最小，则当前客户端竞选leader成功
4.否则，在前一节点上注册watch事件，一旦前者被删除，则它收到通知，返回step 3(并不能认为自己成为新节点，有可能前面的节点只是宕机)
5.leader节点通过删除自己放弃leadership

kafka基于Controller的leader election
1.整个集群中选举出一个broker作为controller
2.controller为所有的topic的所有partition指定leader及follower
优点：
.缓解herd effect问题
.减轻zk负担
.controller与leader和follower间通过rpc通信，高效且实时
缺点：
.引入controller增加了复杂度
.需要考虑controller的failover

High level consumer rebalance:
1.high level consumer启动时将其id注册到consumer group下，在zk上的路径为/consumers/[consumer group]/ids/[consumer id]
2.在consumers/[consumer group]/ids上注册watch
3.在/brokers/ids上注册watch
4.如果consumer通过topic filter创建消息流，则它会同时在/brokers/topics上创建watch
5.强制自己在consumer group内启动rebalance流程

consumer rebalance算法：
1.将目标topic下的所有partition排序，存于pt
2.将consumer group下的所有consumer排序，存于cg，第i个consumer标记为ci
3.n=size(pt)/size(cg) 向上取整
4.解除ci对原来分配的partition的消费权，(i从0开始)
5.将第in到n(i+1)n-1个partition分配给ci

用命令启动消费者：

java -jar Demokafka-0.8.2.2.jar localhost:2181/kafka topic1 group1 consumer1

使用low level consumer(simple consumer)的主要原因是，用户希望比consumer group更好的控制数据的消费，如：
1.同一条消息重复读多次，方便replay
2.只消费某一个topic的部分partition
3.管理事务，从而确保每条消息被处理一次(exacatly one)

与high level consumer相比较，low level consumer要求用户做大量的额外工作
1.在应用程序中跟踪处理offset,并决定吓一条消费哪一条消息
2.获知每个partition的leader
3.处理leader的变化
4.处理多consumer的协作

high level consumer:
配置
1.自动管理offset
auto.commit.enable=true。
auto.commit.interval.ms=60*1000. //offset提交的间隔

2.手工管理offset
ConsumerConnector.commitOffset()

3.offset存储
offset.storage=zookeeper //默认是存储在zookeeper上，也可以把offset保存在kafka中，配置：offset.storage=kafka
dual.commit.enable=true。 //设置为true，则会同时把offset写在kafka和zk上

partition对应一个物理文件，每个partiotion中会有很多个segment，每个segment中会有很多的message，kafka会一次行删除整个segment的文件
log compaction:
1.一直保持消费log head的consumer可按顺序消费所有信息，并且offset连续
2.任何offset从0开始的读操作都至少可以读到每个key的最后一条消息
3.每条消息的offset保持不变，offset是消息的永久标示符
4.消费本身的顺序不会被改变

new api:
producer:
1.增加回调接口
2.重构partitioner接口

统一high level api 和low level api
1.从kafka.consumer和kafka.javaapi到java.clients.consumer
2.subscribe动态rebalance vs. assign手动分配
3.将offset存在与zk和kafka之外
4.ConsumerRebalancelisten
5.控制消费位置
6.消费流程控制

subscribe:实现high level api的接口
1.自动rebalance
2.自动分配partition给consumer
3.使用subscribe接口，并可注册consumerRebalanceListener

a.public void onPartitionsRevoked(Collection<TopicPartition> partitions) //partition原先被一个consumer消费，rebalance后，不被该consumer消费
b.public void onPartitionsAssigned(Collection<TopicPartition> partitions) //重新分配给该consumer消费的partition

assign:实现low level api接口
同一个group下的不同consumer可以指定消费相同的partition
1.使用assign
2.指定消费目标partition

1.0版本的offset管理
a.自动commit
b.手动commit
手动commit全部offset
手动commit部分offset
支持同步或异步commit，并直接commit回调

对特定partition进行commit:
stream api:

重新分配partition

1.可以针对一个topic进行reassign
2.也可以针对某几个partition进行数据迁移

计划plan
./kafka-reassign-partitions.sh --zookeeper localhost:2181,localhost:2182,localhost:2183 --topics-to-move-json-file /usr/local/kafka/bin/tmp/topics-to-move.json --broker-list "3,4,5" --generate
执行 execute
./kafka-reassign-partitions.sh --zookeeper localhost:2181,localhost:2182,localhost:2183 --reassignment-json-file /usr/local/kafka/bin/tmp/resign-plan.json --execute

验证
./kafka-reassign-partitions.sh --zookeeper localhost:2181,localhost:2182,localhost:2183 --verify --reassignment-json-file /usr/local/kafka/bin/tmp/resign-plan.json

preferred replica leader election:
1.在zk上创建/admin/preferred_replica_election节点，并存入需要调整preferred replica partition信息
2.controller 一直watch该节点，一旦该节点被创建，controller会收到通知，并获取该内容
3.controll读取preferred replica ,如果发现该replica并非leader并且不再isr中，controll会向该replica发送leaderAndRequest,使该replica成为leader,如果当前replica并非leader，并且不再isr中，controller为了保证数据不丢失，并不会将其设置为leader

./kafka-preferred-replica-election.sh --zookeeper localhost:2181,localhost:2182,localhost:2183 --path-to-json-file /usr/local/kafka/bin/tmp/prefer.json

prefer.json内容：
{"partitions":[{"topic":"topic1","partition":1}]} #设置topic1的partition 1的prefer leader

如何处理replica crash
1.leader crash后，isr中任何replica都可以成为leader
2.如果所有的isr都crash，则isr中第一个恢复的replica成为leader或第一个recover的replica成为leader
3.unclean.leader.election.enable