Kafka简介及主要机制

一、简介
Kafka是一种分布式的，基于发布/订阅的消息系统
主要特性：
1）消息持久化
要从大数据中获取真正的价值，那么不能丢失任何信息。Apache Kafka设计上是时间复杂度O(1)的磁盘结构，它提供了常量时间的性能，即使是存储海量的信息（TB级）。
2）高吞吐
记住大数据，Kafka的设计是工作在标准硬件之上，支持每秒数百万的消息。
3）分布式
Kafka明确支持在Kafka服务器上的消息分区，以及在消费机器集群上的分发消费，维护每个分区的排序语义。
4）多客户端支持
Kafka系统支持与来自不同平台（如java、.NET、PHP、Ruby或Python等）的客户端相集成。
5）实时
生产者线程产生的消息对消费者线程应该立即可见，此特性对基于事件的系统（比如CEP系统）是至关重要的。

 

二、概念/相关原理、机制

Broker
-----------------------------------------------------
Kafka集群包含一个或多个服务器，这种服务器被称为broker(kafka实例)

**********************************************************************************************************
1.生产者---》Broker(Leader partition)
[主从复制了再返回ask?ask了再去主从复制?]
选择保存到哪一个partition(Key Hash)

2.Broker(Leader partition)-->>Broker(follower partition)
分布式存储partition
(主从复制)
[选举leader]
删除消息方式

3.Broker(Leader partition)-->>消费者
[处理了消息再返回commit?commit了再处理消息?]
offset
Consumers(消费者)/consumers group（订阅/广播）
**********************************************************************************************************
1.生产者---》Broker
Producers(生产者):负责发布消息到Kafka broker。

2.Broker-->>Broker
Topics/partition
-----------------------------------------------------
Topic在逻辑上可以被认为是一个queue，每条消费都必须指定它的topic，可以简单理解为必须指明把这条消息放进哪个queue里。
为了使得Kafka的吞吐率可以水平扩展，物理上把topic分成一个或多个partition，
每个partition在物理上对应一个文件夹，该文件夹下存储这个partition的所有消息和索引文件。

partition的数量可以在server.properties中指定。
在发送一条消息时，可以指定这条消息的key，producer根据这个key和partition机制来判断将这条消息发送到哪个parition。
paritition机制可以通过指定producer的paritition.class这一参数来指定，该class必须实现kafka.producer.Partitioner接口。

离散:
本例中如果key可以被解析为整数则将对应的整数与partition总数取余，该消息会被发送到该数对应的partition。（每个parition都会有个序号）

import kafka.producer.Partitioner;
import kafka.utils.VerifiableProperties;

public class JasonPartitioner<T> implements Partitioner {

    public JasonPartitioner(VerifiableProperties verifiableProperties) {}

    @Override
    public int partition(Object key, int numPartitions) {
        try {
            int partitionNum = Integer.parseInt((String) key);
            return Math.abs(Integer.parseInt((String) key) % numPartitions);
        } catch (Exception e) {
            return Math.abs(key.hashCode() % numPartitions);
        }
    }
}

partitions/Replication/Leader election(partition级别的概念)
-----------------------------------------------------
一个Topic的多个partitions,被分布在kafka集群中的多个server上;每个server(kafka实例)负责partitions中消息的读写操作;
此外kafka还可以配置partitions需要备份的个数(replicas),每个partition将会被备份到多台机器上,以提高可用性.

1.partitions
Kafka从0.8开始提供partition级别的replication(主从复制,同步)
replication对Kafka的吞吐率是有一定影响的，但极大的增强了可用性。

2.Replication(提高消息可用性，不会丢失)
Kakfa处理失败需要明确定义一个broker是否alive。对于Kafka而言，Kafka存活包含两个条件，
一是它必须维护与Zookeeper的session(这个通过Zookeeper的heartbeat机制来实现)。
二是follower必须能够及时将leader的writing复制过来，不能“落后太多”。

leader会track“in sync”的node list。
如果一个follower宕机，或者落后太多，leader将把它从”in sync”list中移除。
这里所描述的“落后太多”指follower复制的消息落后于leader后的条数超过预定值。

一条消息只有被“in sync” list里的所有follower都从leader复制过去才会被认为已提交。
这样就避免了部分数据被写进了leader，还没来得及被任何follower复制就宕机了，而造成数据丢失（consumer无法消费这些数据）。
而对于producer而言，它可以选择是否等待消息commit，这可以通过request.required.acks来设置。
这种机制确保了只要“in sync”list有一个或以上的flollower，一条被commit的消息就不会丢失。

这里的复制机制即不是同步复制，也不是单纯的异步复制。事实上，同步复制要求“活着的”follower都复制完，这条消息才会被认为commit，这种复制方式极大的影响了吞吐率。
而异步复制方式下，follower异步的从leader复制数据，数据只要被leader写入log就被认为已经commit，这种情况下如果follwer都落后于leader，而leader突然宕机，则会丢失数据。
而Kafka的这种使用“in sync” list的方式则很好的均衡了确保数据不丢失以及吞吐率。
follower可以批量的从leader复制数据，这样极大的提高复制性能（批量写磁盘），极大减少了follower与leader的差距。

3.Leader election
kafka集群，容忍N-1个replicas失效。

每个partition都有一个唯一的leader，所有的读写操作都在leader上完成(
    生产者push消息或消费者pull消息，都只和leader打交道，
        partition有多个follower,消费者保证消息不会从多个follower中得到；
    leader宕机了会重新选举出一个leader，保证消费者能得到消息。
)。

Leader的概念是相对于partition的备份(Replication)来说的。

一个基本的原则就是，如果leader不在了，新的leader必须拥有原来的leader commit的所有消息。
这就需要作一个折衷，如果leader在标明一条消息被commit前等待更多的follower确认，
那在它die之后就有更多的follower可以作为新的leader，但这也会造成吞吐率的下降。

1.生产者---》Broker(Leader partition)[主从复制了再返回ask?ask了再去主从复制?]
2.Broker(Leader partition)-->>Broker(follower partition)(主从复制)[选举leader]分布式存储partition
3.Broker(Leader partition)-->>消费者[处理了消息再返回commit?commit了再处理消息?]

如何选举Leader
-----------------------------------------------------
最简单最直观的方案是，所有Follower都在ZooKeeper上设置一个Watch，一旦Leader宕机，
其对应的ephemeral znode会自动删除，此时所有Follower都尝试创建该节点，
而创建成功者（ZooKeeper保证只有一个能创建成功）即是新的Leader，其它Replica即为Follower。

(还没看懂)
但是该方法会有3个问题：
split-brain 这是由ZooKeeper的特性引起的，虽然ZooKeeper能保证所有Watch按顺序触发，但并不能保证同一时刻所有Replica“看”到的状态是一样的，
        这就可能造成不同Replica的响应不一致
herd effect 如果宕机的那个Broker上的Partition比较多，会造成多个Watch被触发，造成集群内大量的调整
ZooKeeper负载过重 每个Replica都要为此在ZooKeeper上注册一个Watch，当集群规模增加到几千个Partition时ZooKeeper负载会过重。

Kafka 0.8.*的Leader Election方案解决了上述问题，它在所有broker中选出一个controller，
所有Partition的Leader选举都由controller决定。controller会将Leader的改变直接通过RPC的方式（比ZooKeeper Queue的方式更高效）
通知需为为此作为响应的Broker。同时controller也负责增删Topic以及Replica的重新分配。

消息Deliver guarantee
-----------------------------------------------------
At most once 消息可能会丢，但绝不会重复传输
At least one 消息绝不会丢，但可能会重复传输
Exactly once 每条消息肯定会被传输一次且仅传输一次，很多时候这是用户所想要的。

读完消息先commit再处理消息:这种模式下，如果consumer在commit后还没来得及处理消息就crash了，下次重新开始工作后就无法读到刚刚已提交而未处理的消息，这就对应于At most once
读完消息先处理再commit:这种模式下，如果处理完了消息在commit之前consumer crash了，下次重新开始工作时还会处理刚刚未commit的消息，实际上该消息已经被处理过了。这就对应于At least once。

如果一定要做到Exactly once，就需要协调offset和实际操作的输出。
比如，consumer拿到数据后可能把数据放到HDFS，如果把最新的offset和数据本身一起写到HDFS，那就可以保证数据的输出和offset的更新要么都完成，要么都不完成，间接实现Exactly once。
（目前就high level API而言，offset是存于Zookeeper中的，无法存于HDFS，而low level API的offset是由自己去维护的，可以将之存于HDFS中）

总之，Kafka默认保证At least once，并且允许通过设置producer异步提交来实现At most once。
而Exactly once要求与目标存储系统协作，幸运的是Kafka提供的offset可以使用这种方式非常直接非常容易。

Persistence/文件存储Log
-----------------------------------------------------
kafka使用文件存储消息(append only log),这就直接决定kafka在性能上严重依赖文件系统的本身特性.
且无论任何OS下,对文件系统本身的优化是非常艰难的.文件缓存/直接内存映射等是常用的手段.
因为kafka是对日志文件进行append操作,因此磁盘检索的开支是较小的;同时为了减少磁盘写入的次数,
broker会将消息暂时buffer起来,当消息的个数(或尺寸)达到一定阀值时,再flush到磁盘,
这样减少了磁盘IO调用的次数.对于kafka而言,较高性能的磁盘,将会带来更加直接的性能提升.

需要考虑的影响性能点很多,除磁盘IO之外,我们还需要考虑网络IO,这直接关系到kafka的吞吐量问题.
对于producer端,可以将消息buffer起来,当消息的条数达到一定阀值时,批量发送给broker;
对于consumer端也是一样,批量fetch多条消息.不过消息量的大小可以通过配置文件来指定.

kafka支持gzip/snappy等多种压缩方式.

Kafka提供两种策略去删除旧数据。
一是基于时间，二是基于partition文件大小。
例如:让Kafka删除一周前的数据，也可通过配置让Kafka在partition文件超过1GB时删除旧数据。

***************************************
3.Consumers(消费者)/consumers group
消费消息
每个consumer属于一个特定的consuer group（可为每个consumer指定group name，若不指定group name则属于默认的group）。
同一topic的一条消息只能被同一个consumer group内的一个consumer消费，但多个consumer group可同时消费这一消息。
如果每一个consumer都属于一个特定的group，则每一个consumer都能得到这条消息。

offset
-----------------------------------------------------
当前消费的消息的position
这个offset由consumer控制。正常情况下consumer会在消费完一条消息后线性增加这个offset。
当然，consumer也可将offset设成一个较小的值，重新消费一些消息。

Consumer Rebalance(后续引入Zookeeper)（Consumer加入或减少）（Broker加入或减少）待续
-----------------------------------------------------
如果某consumer group中consumer数量少于partition数量，则至少有一个consumer会消费多个partition的数据，
如果consumer的数量与partition数量相同，则正好一个consumer消费一个partition的数据，

而如果consumer的数量多于partition的数量时，会有部分consumer无法消费该topic下任何一条消息。

参考:
http://www.linuxidc.com/Linux/2014-09/107388.htm
http://blog.csdn.NET/qqqq724/article/details/43228863
http://my.oschina.Net/frankwu/blog/303745