Kafka简介及基本原理与使用场景

Apache Kafka是分布式发布-订阅消息系统。它最初由LinkedIn公司开发，之后成为Apache项目的一部分。Kafka是一种快速、可扩展的、设计内在就是分布式的，分区的和可复制的提交日志服务。

现在很多开源分布式系统，例如Flume(数据实时分析)，Storm(数据实时处理)，Spark(内存数据处理)，elasticsearch(全文检索)
几种分布式系统的对比

2017-05-05_095912.png

上图介绍到的动态扩容，kafka目前是通过zookeeper来实现动态扩容的。zookeeper：一个提供分布式状态管理，分布式配置管理，分布式锁服务的集群。

AMQP协议

kafka借鉴AMQP协议进行开发
基本概念

 

消费者(Consumer)：从消息队列中请求消息的客户端应用程序

生产者(Producer)：向Broker发布消息的客户端应用程序。

AMQP服务器端(Broker)：用于接收生产者发送的消息并将消息路由给服务器中队列。

话题（Topic）：是特定类型的消息流。消息是字节的有效负载（Payload），话题是消息的分类名或种子（Feed）名。类似新闻中的体育，娱乐，教育等概念。实际应用中往往一个业务一个主题。

分区(Partition)：topic中的消息按照分区来进行组织。其是kafka消息队列组织的最小单位，一个分区可以看作一个FIFO队列。

 

备份(Replication)：为了保证分布式高可靠性，kafka0.8开始对每个分区数据进行备份，防止一个Broker宕机导致分区数据中数据不可用

zookeeper配置

配置zookeeper需要先配置Java_HOME，注意下JAVA_HOME的配置方法如下：

JAVA_HOME=/usr/local/jdk1.8 export PATH=$JAVA_HOME/bin:$PATH

一般将系统的环境变量PATH写在后面，因为系统读取环境变量是从前往后找的，如果PATH中本来已经配置了JAVA_HOME，那么将其放在后面可以让我们配置的JAVA_HOME优先被读取到。
配置完java的环境变量后，需要配置zookeeper的conf/目录下的zoo.cfg文件。将zoo_sample.cfg拷贝为zoo.cfg后，修改内容为下：

# The number of milliseconds of each tick # 即下面计时方式的单位 tickTime=2000 # The number of ticks that the initial # synchronization phase can take # 20S，即这个时间内，集群中的机器都要启动 initLimit=10 # The number of ticks that can pass between # sending a request and getting an acknowledgement # leader发送给follower的心跳超时时间 syncLimit=5 # the directory where the snapshot is stored. # do not use /tmp for storage, /tmp here is just # example sakes. # 设置了zookeeper的dataDir以及DataLogDir，注意下这两个目录不要设 # 置为同一个，那样的话会影响到zookeeper的性能。 dataDir=/home/eversilver/silverTest/kafka/zkData # 这里存放的是zookeeper的事务日志，一般很多，需要定期的去清理，否 # 则产生很多垃圾，拖慢响应速度。官方文档只给出了具体的清理方法 dataLogDir=/home/eversilver/silverTest/kafka/zookeeper/zkLog # the port at which the clients will connect clientPort=2181 # the maximum number of client connections. # increase this if you need to handle more clients #maxClientCnxns=60 # # Be sure to read the maintenance section of the # administrator guide before turning on autopurge. # # #sc_maintenance # # The number of snapshots to retain in dataDir #autopurge.snapRetainCount=3 # Purge task interval in hours # Set to "0" to disable auto purge feature #autopurge.purgeInterval=1 # 设置zookeeper的服务器集群，集群一般设置为基数台，当前机器对应 # 的集群id为1，有几台机器就配置几行，这里有三台机器。 # 机器名称=机器IP:leader与follower之间通信端口:leader选举的端口 # 注意，所以集群中的机器上面的端口应该对应相同 server.1=192.168.142.133:12888:13888 server.2=192.168.142.134:12888:13888 server.3=192.168.142.135:12888:13888

cfg文件配置完成后，在cfg文件中的dataDir目录下新建一个myid文件，代表当前机器的id，这个id与上面server.后面的值相同即可：

eversilver@debian:/usr/local/zookeeper$ cat ~/silverTest/kafka/zookeeper/zkData/myid 1

然后即可启动本机，有几台机器即启动几台：

eversilver@debian:/usr/local/zookeeper$ ./bin/zkServer.sh start ZooKeeper JMX enabled by default Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg Starting zookeeper ... STARTED eversilver@debian:/usr/local/zookeeper$ jps 25846 Jps 25822 QuorumPeerMain eversilver@debian:/usr/local/zookeeper$