六、kafka集群
(利用上面server1,server2,server3,下面以server1为实例)
1)下载kafka0.8(http://kafka.apache.org/downloads.html),保存到服务器/home/wwb目录下kafka-0.8.0-beta1-src.tgz(kafka_2.8.0-0.8.0-beta1.tgz)
2)解压 tar -zxvf kafka-0.8.0-beta1-src.tgz,产生文件夹kafka-0.8.0-beta1-src更改为kafka01
3)配置
修改kafka01/config/server.properties,其中broker.id,log.dirs,zookeeper.connect必须根据实际情况进行修改,其他项根据需要自行斟酌。大致如下:
broker.id=1
port=9091
log.dir=./logs
num.partitions=2
zookeeper.connect=192.168.0.1:2181,192.168.0.2:2182,192.168.0.3:2183
kafka.csv.metrics.dir=/tmp/kafka_metrics
4)初始化因为kafka用scala语言编写,因此运行kafka需要首先准备scala相关环境。
> cd kafka01
> ./sbt update
> ./sbt package
> ./sbt assembly-package-dependency
在第二个命令时可能需要一定时间,由于要下载更新一些依赖包。所以请大家 耐心点。
5) 启动kafka01
>JMX_PORT=9997 bin/kafka-server-start.sh config/server.properties &
a)kafka02操作步骤与kafka01雷同,不同的地方如下
修改kafka02/config/server.properties
broker.id=2
port=9092
##其他配置和kafka-0保持一致
启动kafka02
JMX_PORT=9998 bin/kafka-server-start.shconfig/server.properties &
b)kafka03操作步骤与kafka01雷同,不同的地方如下
修改kafka03/config/server.properties
broker.id=3
port=9093
##其他配置和kafka-0保持一致
启动kafka02
JMX_PORT=9999 bin/kafka-server-start.shconfig/server.properties &
6)创建Topic(包含一个分区,三个副本)
>bin/kafka-create-topic.sh--zookeeper 192.168.0.1:2181 --replica 3 --partition 1 --topicmy-replicated-topic
7)查看topic情况
>bin/kafka-list-top.sh --zookeeper 192.168.0.1:2181
topic: my-replicated-topic partition: 0 leader: 1 replicas: 1,2,0 isr: 1,2,0
8)创建发送者
>bin/kafka-console-producer.sh--broker-list 192.168.0.1:9091 --topic my-replicated-topic
my test message1
my test message2
^C
9)创建消费者
>bin/kafka-console-consumer.sh --zookeeper127.0.0.1:2181 --from-beginning --topic my-replicated-topic
...
my test message1
my test message2
^C
10)杀掉server1上的broker
>pkill -9 -f config/server.properties
11)查看topic
>bin/kafka-list-top.sh --zookeeper192.168.0.1:2181
topic: my-replicated-topic partition: 0 leader: 1 replicas: 1,2,0 isr: 1,2,0
发现topic还正常的存在
11)创建消费者,看是否能查询到消息
>bin/kafka-console-consumer.sh --zookeeper192.168.0.1:2181 --from-beginning --topic my-replicated-topic
...
my test message 1
my test message 2
^C
说明一切都是正常的。
OK,以上就是对Kafka个人的理解,不对之处请大家及时指出。
补充说明:
1、public Map>> createMessageStreams(Map topicCountMap),其中该方法的参数Map的key为topic名称,value为topic对应的分区数,譬如说如果在kafka中不存在相应的topic时,则会创建一个topic,分区数为value,如果存在的话,该处的value则不起什么作用
2、关于生产者向指定的分区发送数据,通过设置partitioner.class的属性来指定向那个分区发送数据,如果自己指定必须编写相应的程序,默认是kafka.producer.DefaultPartitioner,分区程序是基于散列的键。
3、在多个消费者读取同一个topic的数据,为了保证每个消费者读取数据的唯一性,必须将这些消费者group_id定义为同一个值,这样就构建了一个类似队列的数据结构,如果定义不同,则类似一种广播结构的。
4、在consumerapi中,参数设计到数字部分,类似Map,
numStream,指的都是在topic不存在的时,会创建一个topic,并且分区个数为Integer,numStream,注意如果数字大于broker的配置中num.partitions属性,会以num.partitions为依据创建分区个数的。
5、producerapi,调用send时,如果不存在topic,也会创建topic,在该方法中没有提供分区个数的参数,在这里分区个数是由服务端broker的配置中num.partitions属性决定的
欢迎关注,一起学习大数据!
领取专属 10元无门槛券
私享最新 技术干货