KAFKA学习总结

一、 简介

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

KAFKA用途广泛,可以用作消息队列,也可以用作日志系统,还有其他一些应用,在此不作详细介绍;

二、 组成

按照不同模块的职责来区分,一个正常运行的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来协调生产和消费的过程;

三、 术语

除了上一节中提到的四个部分之外,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,大概是这么个样子:

四、 使用

配置

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

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

? 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

kafka学习总结