kafka学习总结

KAFKA学习总结

1、 简介

KAFKA是Apache基金会的一个开源项目,是一个分布式的发布-订阅的消息系统;

KAFKA用途普遍,能够用做消息队列,也能够用做日志系统,还有其余一些应用,在此不做详细介绍;

2、 组成

按照不一样模块的职责来区分,一个正常运行的KAFKA共有四个部分:Zookeeper,Broker,Producer和Consumer;

Producer

消息产生者,产生消息,发送到Broker;

Consumer

消息Consumer,从Broker获得消息;

Broker

亦即KAFKA的Server;中间人,消息中转站;一堆Broker组成的集群叫作cluster

Zookeeper

管理链接信息,包括各个节点的IP,端口等;Producer和Consumer须要到Zookeeper请求Broker的信息,从而进行消息的收发;一个新的Broker的启动也须要到Zookeeper来注册; zookeeper也能够配集群。目的是防止某一台挂了;

 producer和consumer经过zookeeper去发现topic，而且经过zookeeper来协调生产和消费的过程;

3、 术语

除了上一节中提到的四个部分以外,KAFKA还包括一些其余概念,现介绍以下:

Topic

Topic,是KAFKA对消息分类的依据;一条消息,必须有一个与之对应的Topic;

好比如今又两个Topic,分别是Love和Hate,Producer向Love发送一个消息XiJinping,而后向Hate发送一个消息Obama;那么,订阅Love的Consumer就会收到消息XiJinping,订阅Hate的Consumer就会收到消息Obama;(每一个Consumer能够同时订阅多个Topic,也便是说,同时订阅Love和Hate的Consumer能够收到XiJinping和Obama);

Message

Message就是咱们所说的消息,是KAfKA操做的对象,消息是按照Topic存储的;

KAFKA中按照必定的期限保存着全部发布过的Message,无论这些Message是否被消费过;例如这些Message的保存期限被这只为两天,那么一条Message从发布开始的两天时间内是可用的,超过保存期限的消息会被清空以释放存储空间;

Partition

每个Topic能够有多个Partition,这样作是为了提升KAFKA系统的并发能力;

每一个Partition中按照消息发送的顺序保存着Producer发来的消息,每一个消息用ID标识,表明这个消息在改Partition中的偏移量,这样,知道了ID,就能够方便的定位一个消息了;每一个新提交过来的消息,被追加到Partition的尾部;若是一个Partition被写满了,就再也不追加;(注意,KAFKA不保证不一样Partition之间的消息有序保存)

Leader

Partition中负责消息读写的节点;Leader是从Partition的节点中随机选取的;

ReplicationFactor

一个Partition中复制数据的全部节点,包括已经挂了的;

isr

ReplicationFactor的子集,存活的且和Leader保持同步的节点;

Consumer Group

传统的消息系统提供两种使用方式:队列和发布-订阅;

队列,是一个池中有若干个Consumer,一条消息发出来之后,被其中的一个Consumer消费;

发布-订阅,是一个消息被广播出去,以后被全部订阅该主题的Consumer消费;

KAFKA提供的使用方式能够达到以上两种方式的效果:Consumer Group;

每个Consumer用Consumer Group Name标识本身,当一条消息产生后,改消息被订阅了其Topic的Consumer Group收到,以后被这个Consumer Group中的一个Consumer消费;

若是全部的Consumer都在同一个Consumer Group中,那么这就和传统的队列形式的消息系统同样了;

若是每个Consumer都在一个不一样的Consumer Group中,那么就和传统的发布-订阅的形式同样了;

上个图片:

一个Topic,两个Broker,五个Partition,大概是这么个样子:

 

4、 使用

配置

根据前面提到的四个部分再加上日志,在配置文件中分别有五类配置文件;

下面简要说几项比较基本的配置:

Ÿ zookeeper.properties

dataDir:用于存放zookeeper生成的数据文件,默认放在/tmp/zookeeper路径下;

clientPort:zookeeper监听的端口;

Ÿ server.properties

broker.id:broker的惟一标识,整数;

port:broker监听的端口;

host.name:broker绑定到的IP,若是不设置,将绑定到全部的接口;(官网中的配置文件是这么说的,我也不知道什么叫绑定到全部的接口,多是和端口的说明写混了)

log.dirs:broker存放数据文件的地方,默认在/tmp/kafuka-logs/

zookeeper.connect:

Ÿ producer.properties

metadata.broker.list:broker的ip和端口(我看的文档里说producer和consumer都是从zookeeper得到broker的信息,可是这里又配置producer的信息,不知道是什么意思);

Ÿ consumer.properties

zookeeper.connect:zookeeper的ip和端口

group.id:consumer所属的consumer group的id

producer和consumer的配置文件,官网上的demo没用到这两个文件,直接在命令行里输入了参数,

Ÿ 日志配置文件,目前没用到,不作介绍;

操做步骤

1) 初始化zookeeper;

2) 初始化broker;

3) 建立Topic(若是不显示建立Topic,Producer在发送Message的时候回自动建立,可是诸如Partition等属性就没法自定义了,失去了灵活性,因此不建议不建立Topic);

4) 初始化producer和consumer,这两步没有前后顺序;

5) 产生Message,消费Message,这两部也没有前后顺序;

关于producer中配置broker的问题:

在KAFKA官网上的java代码和命令行demo中,都有在Producer中直接配置broke的地址信息,而我看的一篇介绍文档中,java代码里没有出现props.put("metadata.broker.list", "xxx.xxx.xxx.xxx:xxxx");

而是

props.put("zk.connect", "xxx.xxx.xxx.xxx:xxxx");

可是这种配置方式在个人eclipse里抛出异常了,说是没有发现"metadata.broker.list"的配置;

猜想大概是版本变化的缘由;

应用于java项目中

具体java编码的方式,网上一堆,我就不copy了,简单说一下producer和consumer端都须要哪几个部分:

Ÿ producer:

主类,发送Message的逻辑放在里边;

实现Partitioner的类:根据业务逻辑,将Message发送到不一样的Partition中,若是不实现此接口,KAFKA有一个默认的类;

实现kafka.serializer.Encoder的类,用于封装Message给KAFKA进行解析和传送;

Ÿ consumer:

没有必须实现的接口,不过在官网的demo中将部分接收Message的逻辑抽取出来单独弄了个类而且实现了Runnable接口,这种作法,根据具体的业务逻辑进行变换就好了;

参考文献: 

http://kafka.apache.org/documentation.html

http://my.oschina.net/ielts0909/blog/117489