kafka 技术分享

一、什么是kafka：

            Kafka 是一种高吞吐量的分布式发布订阅消息系统。

            Scala编写的。

 

二、kafka的特点：

            *  通过O(1)的磁盘数据结构提供消息的持久化，这种结构对于即使数以TB的消息存储也能够保持长时间的稳定性能。

            * 高吞吐量[2] ：即使是非常普通的硬件Kafka也可以支持每秒数百万[2]  的消息。

            *  支持通过Kafka服务器和消费机集群来分区消息。

            *支持Hadoop并行数据加载

三、kafka的优缺点：

            优点 ：

                        1、分布式可高可扩展。Kafka 集群可以透明的扩展，增加新的服务器进集群。                              

                        2、高性能。Kafka 的性能大大超过传统的ActiveMQ、RabbitMQ等MQ 实现，尤其是Kafka 还支持batch 操作。
                        3、容错。Kafka每个Partition的数据都会复制到几台服务器上。当某个Broker故障失效时，ZooKeeper服务将通知生产者和消费者，生产者和消费者转而使用其它 Broker。
                       4、支持Scala、Java、C++、C#、Go、PHP、Python、Ruby，PHP需要5.3.3版本以上。
                       5、没有消费反馈，哪些消息已消费由consumer维护(通过记录一个offset值)，consumer也可以回滚到以前的位置，重新读取之前读取过的消息。
                       6. producer发送消息后，broker不会发送ACK给producer。
缺点：
                       1、  重复消息。Kafka 只保证每个消息至少会送达一次，虽然几率很小，但一条消息有可能会被送达多次。

                       2、消息乱序。虽然一个Partition 内部的消息是保证有序的，但是如果一个Topic 有多个Partition，Partition 之间的消息送达不保证有序。

                      3、复杂性。Kafka需要zookeeper 集群的支持，Topic通常需要人工来创建，部署和维护较一般消息队列成本更高

四、kafka的使用场景：

     1、Messaging   

4000

        对于一些常规的消息系统,kafka是个不错的选择;partitons/replication和容错,可以使kafka具有良好的扩展性和性能优势.不过到目前为止,我们应该很清楚认识到,kafka并没有提供JMS中的"事务性""消息传输担保(消息确认机制)""消息分组"等企业级特性;kafka只能使用作为"常规"的消息系统,在一定程度上,尚未确保消息的发送与接收绝对可靠(比如,消息重发,消息发送丢失等)
 
    2、Websitactivity tracking
       kafka可以作为"网站活性跟踪"的最佳工具;可以将网页/用户操作等信息发送到kafka中.并实时监控,或者离线统计分析等
 
    3、LogAggregation
       kafka的特性决定它非常适合作为"日志收集中心";application可以将操作日志"批量""异步"的发送到kafka集群中,而不是保存在本地或者DB中;kafka可以批量提交消息/压缩消息等,这对producer端而言,几乎感觉不到性能的开支.此时consumer端可以使hadoop等其他系统化的存储和分析系统.

四、kafka导读：

      Kafka 集群需要zookeeper支持来实现集群，最新的kafka 发行包中已经包含了zookeeper，部署的时候可以在一台服务器上同时启动一个zookeeper Server 和一个Kafka Server，也可以使用已有的其他zookeeper集群。

 

      Kafka 将消息流按Topic 组织，保存消息的服务器称为Broker，消费者可以订阅一个或者多个Topic。为了均衡负载，一个Topic 的消息又可以划分到多个分区(Partition)，分区越多，Kafka并行能力和吞吐量越高。

 

     和传统的MQ不同，消费者需要自己保留一个offset，从kafka 获取消息时，只拉去当前offset 以后的消息。Kafka 的scala/java 版的client 已经实现了这部分的逻辑，将offset 保存到zookeeper 上。每个消费者可以选择一个id，同样id 的消费者对于同一条消息只会收到一次。一个Topic 的消费者如果都使用相同的id，就是传统的 Queue；如果每个消费者都使用不同的id, 就是传统的pub-sub.

五、kafka相关概念：

1.producer：
　　消息生产者，发布消息到 kafka 集群的终端或服务。
2.broker：
　　kafka 集群中包含的服务器。
3.topic：
　　每条发布到 kafka 集群的消息属于的类别，即kafka 是面向 topic的。
4.partition：
　　partition是物理上的概念，每个 topic包含一个或多个partition。kafka 分配的单位是 partition。
5.consumer：
　　从 kafka 集群中消费消息的终端或服务。
6.Consumergroup：
　　high-levelconsumer API 中，每个consumer 都属于一个consumer group，每条消息只能被consumer group 中的一个Consumer 消费，但可以被多个consumer group 消费。
7.replica：
　　partition的副本，保障partition 的高可用。
8.leader：
　　replica中的一个角色，producer 和consumer 只跟leader 交互。
9.follower：
　　replica中的一个角色，从leader 中复制数据。
10.controller：
　　kafka 集群中的其中一个服务器，用来进行 leader election 以及 各种failover。
12.zookeeper：
　　kafka 通过 zookeeper 来存储集群的 meta 信息。

六、 kafka一个topic的多个partition，数据是怎么存放的？kafka为什么吞吐量高？

    

答：因为每条消息都被append到该partition中，是顺序写磁盘，因此效率非常高（经验证，顺序写磁盘效率比随机写内存还要高，这是Kafka高吞吐率的一个很重要的保证）。

Kafka就是使用了分区partition，通过将topic的消息打散到多个分区并分布保存在不同的broker上实现了消息处理不管是producer还是consumer的高吞吐量。

七、问题

    

1.  topic满了怎么办？

首先topic是有属性的，比如设置了2个参数max message size 和 flush rate.

分别表示消息的最大消息数，写出频率。

当达到max message size阈值的时候，就会flush到磁盘上。不会溢出丢失或者消息存不进来。

Kafka的topic属性，还设置了删除策略，都是通过参数属性实现的。

参考：http://blog.csdn.net/dly1580854879/article/details/71404135

2.  为啥要删除过去很久的topic数据？

Kafka消费完数据后，数据还保留在自盘上，每次消息消费在哪里了，是用offset来记录的，offset由zookeeper管理。

3 . kafka消费者的消息确认。

Kafka消费者没有消息确认机制。这是他和其他mq的区别。

但是kafka生产者的生产却不同。只有Leader负责数据读写，Follower只向Leader顺序Fetch数据（N条通路），系统更加简单且高效。zookeeper是将信息leader信息同步到Follows中。

kafka在引入Replication之后，同一个Partition可能会有多个Replica，而这时需要在这些Replication之间选出一个Leader，Producer和Consumer只与这个Leader交互，其它Replica作为Follower从Leader中复制数据。因为需要保证同一个Partition的多个Replica之间的数据一致性（其中一个宕机后其它Replica必须要能继续服务并且即不能造成数据重复也不能造成数据丢失）。如果没有一个Leader，所有Replica都可同时读/写数据，那就需要保证多个Replica之间互相（N×N条通路）同步数据，数据的一致性和有序性非常难保证，大大增加了Replication实现的复杂性，同时也增加了出现异常的几率。而引入Leader后，只有Leader负责数据读写，Follower只向Leader顺序Fetch数据（N条通路），系统更加简单且高效。zookeeper是将信息leader信息同步到Follows中。

九、zookeeper在kafka中的作用？

    http://blog.csdn.net/dly1580854879/article/details/71403778
       


下面这段，我感觉对kafka的理解很有帮助：

每一条消息被发送到broker时，会根据paritition规则选择被存储到哪一个partition。如果partition规则设置的合理，所有消息可以均匀分布到不同的partition里，这样就实现了水平扩展。（如果一个topic对应一个文件，那这个文件所在的机器I/O将会成为这个topic的性能瓶颈，而partition解决了这个问题）。在创建topic时可以在server.properties中指定这个partition的数量(如下所示)，当然也可以在topic创建之后去修改parition数量。num.partitions=3。
并且：Kafka就是使用了分区partition，通过将topic的消息打散到多个分区并分布保存在不同的broker上实现了消息处理不管是producer还是consumer的高吞吐量。Kafka的生产者和消费者都可以多线程地并行操作，而每个线程处理的是一个分区的数据。因此分区实际上是调优Kafka并行度的最小单元。对于producer而言，它实际上是用多个线程并发地向不同分区所在的broker发起Socket连接同时给这些分区发送消息；而consumer呢，同一个消费组内的所有consumer线程都被指定topic的某一个分区进行消费具体如何确定consumer线程数目我们后面会详细说明。所以说，如果一个topic分区越多，理论上整个集群所能达到的吞吐量就越大。但分区是否越多越好呢？显然也不是，因为每个分区都有自己的开销：一、客户端/服务器端需要使用的内存就越多先说说客户端的情况。Kafka
082之后推出了Java版的全新的producer，这个producer有个参数batchsize，默认是16KB。它会为每个分区缓存消息，一旦满了就打包将消息批量发出。看上去这是个能够提升性能的设计。不过很显然，因为这个参数是分区级别的，如果分区数越多，这部分缓存所需的内存占用也会更多。假设你有10000个分区，按照默认设置，这部分缓存需要占用约157MB的内存。而consumer端呢？我们抛开获取数据所需的内存不说，只说线程的开销。如果还是假设有10000个分区，同时consumer线程数要匹配分区数大部分情况下是最佳的消费吞吐量配置的话，那么在consumer
client就要创建10000个线程，也需要创建大约10000个Socket去获取分区数据。这里面的线程切换的开销本身已经不容小觑了。服务器端的开销也不小，如果阅读Kafka源码的话可以发现，服务器端的很多组件都在内存中维护了分区级别的缓存，比如controller，FetcherManager等，因此分区数越多，这种缓存的成本越久越大。二、文件句柄的开销每个分区在底层文件系统都有属于自己的一个目录。该目录下通常会有两个文件： base_offsetlog和base_offsetindex。Kafak的controller和ReplicaManager会为每个broker都保存这两个文件句柄file
handler。很明显，如果分区数越多，所需要保持打开状态的文件句柄数也就越多，最终可能会突破你的ulimit -n的限制。三、降低高可用性Kafka通过副本replica机制来保证高可用。具体做法就是为每个分区保存若干个副本replica_factor指定副本数。每个副本保存在不同的broker上。期中的一个副本充当leader 副本，负责处理producer和consumer请求。其他副本充当follower角色，由Kafka controller负责保证与leader的同步。如果leader所在的broker挂掉了，contorller会检测到然后在zookeeper的帮助下重选出新的leader——这中间会有短暂的不可用时间窗口，虽然大部分情况下可能只是几毫秒级别。但如果你有10000个分区，10个broker，也就是说平均每个broker上有1000个分区。此时这个broker挂掉了，那么zookeeper和controller需要立即对这1000个分区进行leader选举。比起很少的分区leader选举而言，这必然要花更长的时间，并且通常不是线性累加的。如果这个broker还同时是controller情况就更糟了。说了这么多“废话”，很多人肯定已经不耐烦了。那你说到底要怎么确定分区数呢？答案就是：视情况而定。基本上你还是需要通过一系列实验和测试来确定。当然测试的依据应该是吞吐量。虽然LinkedIn这篇文章做了Kafka的基准测试，但它的结果其实对你意义不大，因为不同的硬件、软件、负载情况测试出来的结果必然不一样。我经常碰到的问题类似于，官网说每秒能到10MB，为什么我的producer每秒才1MB？
—— 且不说硬件条件，最后发现他使用的消息体有1KB，而官网的基准测试是用100B测出来的，因此根本没有可比性。不过你依然可以遵循一定的步骤来尝试确定分区数：创建一个只有1个分区的topic，然后测试这个topic的producer吞吐量和consumer吞吐量。假设它们的值分别是Tp和Tc，单位可以是MB/s。然后假设总的目标吞吐量是Tt，那么分区数 = Tt / maxTp,TcTp表示producer的吞吐量。测试producer通常是很容易的，因为它的逻辑非常简单，就是直接发送消息到Kafka就好了。Tc表示consumer的吞吐量。测试Tc通常与应用的关系更大，因为Tc的值取决于你拿到消息之后执行什么操作，因此Tc的测试通常也要麻烦一些。另外，Kafka并不能真正地做到线性扩展其实任何系统都不能，所以你在规划你的分区数的时候最好多规划一下，这样未来扩展时候也更加方便。消息-分区的分配默认情况下，Kafka根据传递消息的key来进行分区的分配，即hashkey