kafka集群部署以及java客户端测试

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方面进行测试，在正常使用中必然要使用线程去监听消费队列的，而在生产这方面也是要利用线程去提交数据，保障主线程的畅通等等，这点根据自己的业务需要进行完善了，本人这里也只是抛砖引玉罢了。最后，本文都是基于本人实际搭建过程的心得，若有错误欢迎指正。