Kafka集群搭建

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

![title](/api/file/getImage?fileId=592398660f93bd0545000151)

## Topics/logs/consumer/Producer

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

![title](/api/file/getImage?fileId=5923990d0f93bd0545000153)

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

kafka集群几乎不需要维护任何consumer和producer状态信息,这些信息有zookeeper保存;因此producer和consumer的客户端实现非常轻量级,它们可以随意离开,而不会对集群造成额外的影响.

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

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

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

![title](/api/file/getImage?fileId=592398db0f93bd0545000152)

## consumer/consumer-group

本质上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角度来说,消息仍不是有序的.

## log

如果一个topic的名称为"my_topic",它有2个partitions,那么日志将会保存在my_topic_0和my_topic_1两个目录中;日志文件中保存了一序列"log entries"(日志条目),每个log entry格式为"4个字节的数字N表示消息的长度" + "N个字节的消息内容";每个日志都有一个offset来唯一的标记一条消息,offset的值为8个字节的数字,表示此消息在此partition中所处的起始位置..每个partition在物理存储层面,有多个log file组成(称为segment).segmentfile的命名为"最小offset".kafka.例如"00000000000.kafka";其中"最小offset"表示此segment中起始消息的offset.

![title](/api/file/getImage?fileId=59239a1c0f93bd0545000154)

# 安装部署

## 下载并且解压

tar zxvf kafka-0.9.0.0-src.tgz 
    mv kafka-0.9.0.0-src ~/programs/kafka
    
## 启动zookeeper集群

[参考zookeeper集群搭建](https://note.louyj.com/blog/post/louyj/zookeeper-%E9%9B%86%E7%BE%A4%E6%90%AD%E5%BB%BA)

## 修改server.properties配置

broker.id=1
    port=9092
    log.dirs=/opt/kafka/tmp
    host.name=HDP125
    zookeeper.connect=HDP125:2181,HDP126:2181,HDP127:2181
     
    delete.topic.enable=true #允许删除topic

## 修改各主机hosts文件,添加hostname

## 启动kafka(修改kafka-server-start.sh内存大小)

./kafka-server-start.sh -daemon ../config/server.properties JMX_PORT=19093
    ./kafka-server-start.sh -daemon ../config/server.properties
    
# kafka 相关操作

## 创建topic

./kafka-topics.sh --create --zookeeper localhost:2181/kafka_090 --replication-factor 1 --partitions 1 --topic test
    
## 删除topic(delete.topic.enable=true)

./kafka-topics.sh --delete --topic test --zookeeper localhost:2181/kafka_090
    
## 查看topic

./kafka-topics.sh --list --zookeeper localhost:2181/kafka_090 --topic test
    
## 发送消息

./kafka-console-producer.sh --broker-list localhost:9092,xxxxx --topic test 
    
## 消费消息

./kafka-console-consumer.sh --zookeeper localhost:2181/kafka_090 --topic test --from-beginning

## 查看消费情况

#后续版本已弃用
    ./kafka-consumer-offset-checker.sh --zookeeper localhost:2181/kafka_090 --group test_group  --topic gn_http
    
    ./kafka-consumer-groups.sh --zookeeper localhost:2181/kafka_090  --list
    ./kafka-consumer-groups.sh --zookeeper localhost:2181/kafka_090  --describe --group test_group
    
    #new consumer消费
    ./kafka-consumer-groups.sh --new-consumer  --list --bootstrap-server localhost:9093 
    ./kafka-consumer-groups.sh --new-consumer  --describe --bootstrap-server localhost:9093 --group test_group

## 查看topic分区offset

./kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic tp_test --time -2
    ./kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic tp_test --time -1
    ./kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic tp_test --time -1 --partitions 0
    # --time timestamp/-1(latest)/-2(earliest)

Navigation

Recent Posts

Friend Links