Kafka 之 入门

摘要: 最近研究采集层，对Kafka作了一个研究。分为入门，中级，高级步步进阶。本篇主要介绍基本概念，适用场景。

            

                    

1、入门

1.    简介

Kafka is a distributed, partitioned, replicated commit log service。它提供了相似于JMS的特性，可是在设计实现上彻底不一样，此外它并非JMS规范的实现。kafka对消息保存时根据Topic进行归类，发送消息者成为Producer,消息接受者成为Consumer,此外kafka集群有多个kafka实例组成，每一个实例(server)成为broker。不管是kafka集群，仍是producer和consumer都依赖于zookeeper来保证系统可用性集群保存一些meta信息。

下面这张图描述更准确。

主要特性：

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

 

2.    概念

Topics/logs

一个Topic能够认为是一类消息，每一个topic将被分红多个partition(区),每一个partition在存储层面是append log文件。任何发布到此partition的消息都会被直接追加到log文件的尾部，每条消息在文件中的位置称为offset（偏移量），offset为一个long型数字，它是惟一标记一条消息。它惟一的标记一条消息。kafka并无提供其余额外的索引机制来存储offset，由于在kafka中几乎不容许对消息进行“随机读写”。

    kafka和JMS实现(activeMQ)不一样的是:即便消息被消费,消息仍然不会被当即删除.日志文件将会根据broker中的配置要求,保留必定的时间以后删除;好比log文件保留2天,那么两天后,文件会被清除,不管其中的消息是否被消费.kafka经过这种简单的手段,来释放磁盘空间,以及减小消息消费以后对文件内容改动的磁盘IO开支.

 

    对于consumer而言,它须要保存消费消息的offset,对于offset的保存和使用,有consumer来控制;当consumer正常消费消息时,offset将会"线性"的向前驱动,即消息将依次顺序被消费.事实上consumer可使用任意顺序消费消息,它只须要将offset重置为任意值..(offset将会保存在zookeeper中,参见下文)

    kafka集群几乎不须要维护任何consumer和producer状态信息,这些信息有zookeeper保存;所以producer和consumer的客户端实现很是轻量级,它们能够随意离开,而不会对集群形成额外的影响.

    partitions的设计目的有多个.最根本缘由是kafka基于文件存储.经过分区,能够将日志内容分散到多个server上,来避免文件尺寸达到单机磁盘的上限,每一个partiton都会被当前server(kafka实例)保存;能够将一个topic切分多任意多个partitions,来消息保存/消费的效率.此外越多的partitions意味着能够容纳更多的consumer,有效提高并发消费的能力.(具体原理参见下文).

 

Distribution

一个Topic的多个partitions,被分布在kafka集群中的多个server上;每一个server(kafka实例)负责partitions中消息的读写操做;此外kafka还能够配置partitions须要备份的个数(replicas),每一个partition将会被备份到多台机器上,以提升可用性.

 

    基于replicated方案,那么就意味着须要对多个备份进行调度;每一个partition都有一个server为"leader";leader负责全部的读写操做,若是leader失效,那么将会有其余follower来接管(成为新的leader);follower只是单调的和leader跟进,同步消息便可..因而可知做为leader的server承载了所有的请求压力,所以从集群的总体考虑,有多少个partitions就意味着有多少个"leader",kafka会将"leader"均衡的分散在每一个实例上,来确保总体的性能稳定.

 

Producers

    Producer将消息发布到指定的Topic中,同时Producer也能决定将此消息归属于哪一个partition;好比基于"round-robin"方式或者经过其余的一些算法等.

 

Consumers

    本质上kafka只支持Topic.每一个consumer属于一个consumer group;反过来讲,每一个group中能够有多个consumer.发送到Topic的消息,只会被订阅此Topic的每一个group中的一个consumer消费.

    若是全部的consumer都具备相同的group,这种状况和queue模式很像;消息将会在consumers之间负载均衡.

    若是全部的consumer都具备不一样的group,那这就是"发布-订阅";消息将会广播给全部的消费者.

    在kafka中,一个partition中的消息只会被group中的一个consumer消费;每一个group中consumer消息消费互相独立;咱们能够认为一个group是一个"订阅"者,一个Topic中的每一个partions,只会被一个"订阅者"中的一个consumer消费,不过一个consumer能够消费多个partitions中的消息.kafka只能保证一个partition中的消息被某个consumer消费时,消息是顺序的.事实上,从Topic角度来讲,消息仍不是有序的.

 

    kafka的设计原理决定,对于一个topic,同一个group中不能有多于partitions个数的consumer同时消费,不然将意味着某些consumer将没法获得消息.

 

Guarantees

    1) 发送到partitions中的消息将会按照它接收的顺序追加到日志中

    2) 对于消费者而言,它们消费消息的顺序和日志中消息顺序一致.

    3) 若是Topic的"replication factor"为N,那么容许N-1个kafka实例失效.

 

 

 

3.    适用场景

1、Messaging

    对于一些常规的消息系统,kafka是个不错的选择;partitons/replication和容错,可使kafka具备良好的扩展性和性能优点.不过到目前为止,咱们应该很清楚认识到,kafka并无提供JMS中的"事务性""消息传输担保(消息确认机制)""消息分组"等企业级特性;kafka只能使用做为"常规"的消息系统,在必定程度上,还没有确保消息的发送与接收绝对可靠(好比,消息重发,消息发送丢失等)

2、Websit activity tracking

    kafka能够做为"网站活性跟踪"的最佳工具;能够将网页/用户操做等信息发送到kafka中.并实时监控,或者离线统计分析等

3、Metrics

         Kafka一般被用于可操做的监控数据。这包括从分布式应用程序来的聚合统计用来生产集中的运营数据提要。

4、Log Aggregation

    kafka的特性决定它很是适合做为"日志收集中心";application能够将操做日志"批量""异步"的发送到kafka集群中,而不是保存在本地或者DB中;kafka能够批量提交消息/压缩消息等,这对producer端而言,几乎感受不到性能的开支.此时consumer端可使hadoop等其余系统化的存储和分析系统.

 

4.    命令

1.  启动Server

Kafka 依赖 ZK 服务

nohup bin/kafka-server-start.sh config/server.properties &

 

2.  建立Topic

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic page_visits

 

3.  查看命令

bin/kafka-topics.sh --list --zookeeper localhost:2181

4.  发送消息

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic page_visits

5.  消费消息

bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic page_visits --from-beginning

6.  多 Broker 方式

 

bin/kafka-server-start.sh config/server-1.properties &

bin/kafka-server-start.sh config/server-2.properties &

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic visits

bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic visits

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic visits

my message test1

my message test2

 

bin/kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic visits

 

7.  中止服务

pkill -9 -f config/server.properties

 

 

8.       删除无用的topic

bin/kafka-run-class.sh kafka.admin.DeleteTopicCommand --topic visits --zookeeper sjxt-hd02:2181,sjxt-hd03:2181,sjxt-hd04:2181

 

beta in 0.8.1

bin/kafka-topics.sh --zookeeper zk_host:port --delete --topic my_topic_name