kafka集群部署以及java客户端测试
2016-12-08 16:37
429 查看
kafka集群部署以及java客户端测试
本文主要讲述本人的集群部署kafka过程以及遇到的问题:其中:kafka版本为:kafka_2.10,zookeeper版本为:zookeeper-3.4.8,jdk-8u101-linux-x64
一 kafka以及zookeeper安装以及环境配置
二 zookeeper集群配置以及测试
三 kafka集群配置以及测试
四 java客户端生产与消费测试
一 kafka以及zookeeper安装以及环境配置*
系统环境:CentOS 6.5 服务器三台:192.168.1.1,192.168.1.2,192.168.1.3 jdk下载地址:http://www.oracle.com/technetwork/java/javase/downloads/index.html zookeeper下载地址:http://www.apache.org/dyn/closer.cgi/zookeeper/ kafka下载地址:http://kafka.apache.org/downloads 1.jdk安装 cmd:rpm -ivh jdk-8u101-linux-x64.rpm 安装完毕:java -version显示如下信息即安装成功: java version "1.8.0_101" Java(TM) SE Runtime Environment (build 1.8.0_101-b13) Java HotSpot(TM) 64-Bit Server VM (build 25.101-b13, mixed mode) 2.zookeeper安装 将下载的zookeeper-3.4.8.tar.gz解压,同时将zookeeper放入usr/local下面 cmd:unzip zookeeper-3.4.8.tar.gz cmd:mv zookeeper-3.4.8.tar.gz zookeeper cmd:mv zookeeper /usr/local 3.kafka安装 将下载的kafka_2.10-0.10.0.0.tgz解压安装,同时将zookeeper放入usr/local下面 cmd:unzip kafka_2.10-0.10.0.0.tgz cmd:mv kafka_2.10-0.10.0.0.tgz kafka cmd:mv kafka /usr/local
二 zookeeper集群配置以及测试
1.由于kafka必须依赖zookeeper进行管理自身的broker,offset,producer,comsumer等,所以在安装部署kafka之前必须先部署zookeeeper,虽然kafka也自带了zookeeper,不过本人这里还是自己独立使用的zookeeper.首先进入上面已经解压好的zookeeper目录: cmd:cd /usr/local/zookeeper 修改配置文件:cd conf cmd:cp zoo_sample.cfg zoo.cfg cmd:vi zoo.cfg # The number of milliseconds of each ticktickTime=2000 # synchronization phase can take # The number of ticks that can pass between syncLimit=5 # the directory where the snapshot is stored. dataDir=/usr/local/zookeeper/data dataLogDir=/usr/local/zookeeper/logs # the port at which the clients will connect clientPort=2181 server.0=192.168.1.1:4001:4002 server.1=192.168.1.2:4001:4002 server.2=192.168.1.3:4001:4002
2.以上就是zookeeper的配置文件,server.0,1,2位三个服务器的地址以及端口,首先得确保这三台服务器之间是互通的,同时4001以及4002端口也都是开放的。另外4001表示这三台服务器之间通信使用的端口,4002则是服务器之间选举使用的端口,当有一台服务器down以后就会时候此端口进行选取新的leader。
3.最重要的一步:
分别在三台服务器的dataDir路径下创建一个myid的文件(本例的dataDir地址为:/usr/local/zookeeper/data),文件的内容为zookeeper阶段的编号,例如:
192.168.1.1这台服务器上的文件内容为0,其余的分别是1,2。
cmd:cd /usr/local/zookeeper/data
cmd:echo 0 > myid
cmd:cat myid
若显示0即成功,同时另外两台服务器上配置也如此,只是修改myid里的内容分别为1与2即可。
4.启动集群服务器:
启动三台服务器上的zookeeper服务:
cmd:bin/zkService.sh start
全都启动以后查看状况:
cmd:bin/zkService.sh status
显示如下:
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg
Mode: follower
其中Mode表示node的状态,会有两个follower以及一个leader,此时的server1位leader,为了测试选举,将server1的服务停止:
cmd:bin/zkService.sh stop
再次查看服务器状态,就可以看到新的leader已经更换。
此时zookeeper集群已经搭建OK
三 kafka集群配置以及测试
1.由于是三个服务器,因此这里会存在三个broker,分别是0,1,2,对应的服务器为:192.168.1.1,192.168.1.2,192.168.1.3,因此在创建分区的时候会实用三个分区,如下是安装配置步骤:1.首先要修改kakfa的配置文件
cmd:cd /use/local/kafka/config
cmd:vi server.properties
#此Broker的ID,集群中每个Broker的ID不可相同 broker.id=0(此处三个服务器分别填写0,1,2 请不要写重了) #监听器,端口号与port一致即可 listeners=PLAINTEXT://:9092 #Broker的Hostname,填主机IP即可 advertised.host.name=192.168.1.1(如果要kafka远程访问,此处必填,否则只能局域网访问kafka) #进行IO的线程数,应大于主机磁盘数 num.io.threads=8 #消息文件存储的路径 log.dirs=/usr/local/kafka/kafka-logs #消息文件清理周期,即清理x小时前的消息记录 log.retention.hours=168 #每个Topic默认的分区数,一般在创建Topic时都会指定分区数,所以这个配成1就行了 num.partitions=1 #Zookeeper连接串,此处填写上一节中安装的三个zk节点的ip和端口即可 zookeeper.connect=192.168.1.1:2181,192.168.1.2:2181,192.168.1.3:2181
2.接下来分别启动三个服务器上的kafka:
cmd:bin/kafka-server-start.sh config/server.properties &
没有报错即启动成功
3.然后创建topic
bin/kafka-topics.sh –create –zookeeper 192.168.1.1:2181,192.168.1.2:2181,192.168.1.3:2181 –replication-factor 3 –partitions 3 –topic test
查看Topic的状态:
bin/kafka-topics.sh –describe –zookeeper 192.168.1.1:2181,192.168.1.2:2181,192.168.1.3:2181 –topic test
输出:
Topic:test PartitionCount:3 ReplicationFactor:3 Configs:
Topic: test Partition: 0 Leader: 1 Replicas: 1,2,0 Isr: 2,0,1
Topic: test Partition: 1 Leader: 2 Replicas: 2,0,1 Isr: 2,0,1
Topic: test Partition: 2 Leader: 3 Replicas: 0,1,2 Isr: 2,0,1
简单测试:
生产者:
bin/kafka-console-producer.sh –broker-list 192.168.1.1:9092,192.168.1.2:9092,192.168.1.3:9092 –topic test
消费者:
bin/kafka-console-consumer.sh –zookeeper 192.168.1.1:2181,192.168.1.2:2181,192.168.1.3:2181 –topic test –from-beginning
此时在生产者控制台输入:hello world
消费者正常显示:hello world
如此集群就搭建完毕,剩下的就是利用java客户端进行远程测试。
四 java客户端生产与消费测试
本人直接使用的lib包的方式进行测试的,并非maven。如下是本人的lib包这些lib包都是可以再kafka根目录下的libs里面找到的。
1.首先利用kafka api创建一个生产者:
package com.KafkaService; import java 4000 .util.Properties; import org.apache.kafka.clients.producer.Callback; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata; import org.slf4j.Logger; import org.slf4j.LoggerFactory; public class KafkaProducerService { private static Logger LOG = LoggerFactory .getLogger(KafkaProducerService.class); public static void main(String[] args) { Properties props = new Properties(); props.put("bootstrap.servers", "192.168.1.1:9092,192.168.1.2:9092,192.168.1.3:9092"); props.put("retries", 3); props.put("linger.ms", 1); props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); KafkaProducer<String, String> producer = new KafkaProducer<String, String>( props); for (int i = 0; i < 1; i++) { ProducerRecord<String, String> record = new ProducerRecord<String, String>( "tests", "11", "今天天气不错哟yoyo=======>" + i); producer.send(record, new Callback() { @Override public void onCompletion(RecordMetadata metadata, Exception e) { // TODO Auto-generated method stub if (e != null) System.out.println("the producer has a error:" + e.getMessage()); else { System.out .println("The offset of the record we just sent is: " + metadata.offset()); System.out .println("The partition of the record we just sent is: " + metadata.partition()); } } }); try { Thread.sleep(1000); // producer.close(); } catch (InterruptedException e1) { // TODO Auto-generated catch block e1.printStackTrace(); } } } }
其次,创建消费者代码:
package com; import java.util.ArrayList; import java.util.Arrays; import java.util.List; import java.util.Properties; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** * @author Joker * 自己控制偏移量提交 * 很多时候,我们是希望在获得消息并经过一些逻辑处理后,才认为该消息已被消费,这可以通过自己控制偏移量提交来实现。 */ public class ManualOffsetConsumer { private static Logger LOG = LoggerFactory.getLogger(ManualOffsetConsumer.class); public static void main(String[] args) { // TODO Auto-generated method stub Properties props = new Properties(); //设置brokerServer(kafka)ip地址 props.put("bootstrap.servers", "192.168.1.1:9092,192.168.1.2:9092,192.168.1.3:9092"); //设置consumer group name props.put("group.id","mygroup11"); props.put("enable.auto.commit", "false"); //设置使用最开始的offset偏移量为该group.id的最早。如果不设置,则会是latest即该topic最新一个消息的offset //如果采用latest,消费者只能得道其启动后,生产者生产的消息 props.put("auto.offset.reset", "earliest"); //设置心跳时间 props.put("session.timeout.ms", "30000"); props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); KafkaConsumer<String ,String> consumer = new KafkaConsumer<String ,String>(props); consumer.subscribe(Arrays.asList("test")); final int minBatchSize = 5; //批量提交数量 List<ConsumerRecord<String, String>> buffer = new ArrayList<>(); while (true) { ConsumerRecords<String, String> records = consumer.poll(100); for (ConsumerRecord<String, String> record : records) { System.out.println("consumer message values is "+record.value()+" and the offset is "+ record.offset()); buffer.add(record); } if (buffer.size() >= minBatchSize) { System.out.println("now commit offset"+buffer.size()); consumer.commitSync(); buffer.clear(); } } } }
如此便是java客户端的一个生产者以及消费者的例子,例子中消费者只是利用了一个简单的main方面进行测试,在正常使用中必然要使用线程去监听消费队列的,而在生产这方面也是要利用线程去提交数据,保障主线程的畅通等等,这点根据自己的业务需要进行完善了,本人这里也只是抛砖引玉罢了。最后,本文都是基于本人实际搭建过程的心得,若有错误欢迎指正。
相关文章推荐
- kafka集群部署以及java客户端测试
- kafka集群部署以及java客户端测试
- kafka集群部署以及java客户端测试
- kafka集群部署以及java客户端测试
- kafka集群部署以及java客户端测试
- kafka集群部署以及java客户端测试
- kafka集群部署以及java客户端测试
- kafka集群部署以及java客户端测试
- kafka集群部署以及java客户端测试
- kafka集群部署以及java客户端测试
- kafka集群部署以及java客户端测试
- kafka集群部署以及java客户端测试
- kafka集群部署以及java客户端测试
- kafka集群部署以及java客户端测试
- kafka集群部署以及java客户端测试
- kafka集群部署以及java客户端测试
- kafka集群部署以及java客户端测试
- kafka集群部署以及java客户端测试
- kafka集群部署以及java客户端测试
- kafka集群部署以及java客户端测试