频道栏目
首页 > 网络 > 云计算 > 正文

Kafka实践(三)java开发环境搭建

2016-09-07 09:38:22           
收藏   我要投稿

前面已经部署好了Kafka集群(伪分布式),下面要进入java开发环境搭建。

一、环境描述

1、win10下 eclipse (kepler)

2、本机建立了虚拟机 系统: centos 6.5 ip:192.168.136.134

3、在134上部署了zookeeper 伪分布式部署192.168.136.134:2181,192.168.136.134:2182,192.168.136.134:2183

4、在134上部署了kafka broker集群(伪分布式部署):192.168.136.134:9092,192.168.136.134:9093,192.168.136.134:9094

broker id 分别为 1,2,3

二、目标

1、在eclipse上建立工程,导入依赖库,可以正常编译, producter和consumer 都可以实现

2、可以在eclipse 调试程序, 本机 ----》 连接虚拟机的 zookeeper集群和 kafka的broker集群

3、打包后的程序,可以部署到虚拟机134上, 实现 虚机上的客户端 ---》连接虚拟机的zookeeper集群和 kafka的broker集群

三、搭建过程

整个过程其实走了一些弯路,网络上一些内容因为版本的问题,可能会用 maven方式,实际上现在也可以直接用常用的java工程开发方式,看各自喜欢。

另外,最新的api开发更简单而且据说效率更高,所以。。。。摸索都是泪。 希望本篇可以让大家少走弯路。

1) maven方式建立工程 (可行,个人不推荐, 因为工作场所无外网)

----------具体安装maven和配置的步骤见maven实践

在eclipse集成maven已经实现后,创建一个maven工程: 打开eclipse,File->new->other 选中maven project

 

\

 

然后,修改pom.xml,在文件中增加如下依赖

org.apache.kafka

kafka_2.10

0.8.2.0

完成后,保存。 保存会触发maven工程自己会从网络上下载依赖库,出现如下界面,就说明工程正在自动下载依赖库。

 

\

 

编写producter代码:

package com.newland.appkafka;

import java.util.Properties;

import java.util.concurrent.TimeUnit;

import kafka.javaapi.producer.Producer;

import kafka.producer.KeyedMessage;

import kafka.producer.ProducerConfig;

import kafka.serializer.StringEncoder;

public class kafkaProducer extends Thread{

private String topic;

public kafkaProducer(String topic){

super();

this.topic = topic;

}

@Override

public void run() {

Producer producer = createProducer();

int i=0;

while(true){

producer.send(new KeyedMessage(topic, "message: " + i++));

try {

TimeUnit.SECONDS.sleep(1);

} catch (InterruptedException e) {

e.printStackTrace();

}

}

}

private Producer createProducer() {

Properties properties = new Properties();

properties.put("zookeeper.connect", "192.168.136.134:2181,192.168.136.134:2182,192.168.136.134:2183");//声明zk

properties.put("serializer.class", StringEncoder.class.getName());

properties.put("metadata.broker.list", "192.168.136.134:9092,192.168.136.134:9093,192.168.136.134:9094");// 声明kafka broker

return new Producer(new ProducerConfig(properties));

}

public static void main(String[] args) {

new kafkaProducer("cwqsolotest").start();// 使用kafka集群中创建好的主题 test

}

}

类似的consumer代码:

package com.newland.appkafka;

import java.util.HashMap;

import java.util.List;

import java.util.Map;

import java.util.Properties;

import kafka.consumer.Consumer;

import kafka.consumer.ConsumerConfig;

import kafka.consumer.ConsumerIterator;

import kafka.consumer.KafkaStream;

import kafka.javaapi.consumer.ConsumerConnector;

/**

* 接收数据

* 接收到: message: 10

接收到: message: 11

接收到: message: 12

接收到: message: 13

接收到: message: 14

* @author zm

*

*/

public class kafkaConsumer extends Thread{

private String topic;

public kafkaConsumer(String topic){

super();

this.topic = topic;

}

@Override

public void run() {

ConsumerConnector consumer = createConsumer();

Map topicCountMap = new HashMap();

topicCountMap.put(topic, 1); // 一次从主题中获取一个数据

Map>> messageStreams = consumer.createMessageStreams(topicCountMap);

KafkaStream stream = messageStreams.get(topic).get(0);// 获取每次接收到的这个数据

ConsumerIterator iterator = stream.iterator();

while(iterator.hasNext()){

String message = new String(iterator.next().message());

System.out.println("接收到: " + message);

}

}

private ConsumerConnector createConsumer() {

Properties properties = new Properties();

properties.put("zookeeper.connect", "192.168.136.134:2181,192.168.136.134:2182,192.168.136.134:2183");//声明zk

properties.put("group.id", "group5");// 必须要使用别的组名称, 如果生产者和消费者都在同一组,则不能访问同一组内的topic数据

return Consumer.createJavaConsumerConnector(new ConsumerConfig(properties));

}

public static void main(String[] args) {

new kafkaConsumer("cwqsolotest").start();// 使用kafka集群中创建好的主题 test

}

}

然后确保没有错误后,在需要测试的类上右键-》run as java application

正常情况下,就可以连接上kafka,producter向消息队列推送消息,consumer可以从消息队列接受消息。

接收到: message: 11

接收到: message: 12

接收到: message: 13

接收到: message: 14

本人在运行producter时,并不顺利,eclipse控制台很不争气的报错了,

 

\

 

一度怀疑自己的写法,或者依赖包有问题,后来试着打包,将生成后的jar包,放到虚拟机上运行,一切正常,才确定是因为部署在本机和虚拟机两台机器的原因。

通过查找资料,大致确定了原因,就是本机和kafka服务端所在虚拟机,不是同一台机器,如果kafka服务端,设置的ip用的localhost,会发生此类问题。

于是修改kafka的server1.properties, server2.properties,server3.properties

 

\

 

 

\

 

上面是通过maven方式建立工程。

2) 下面说明一下如何建立 java 工程。

与普通的java工程一样,建立后,将producter和consumer类放到工程里, 在lib 目录下,拷贝原kafka lib目录下的jar包

 

\

 

红色框内为需要从kafka的libs里拷贝过来的jar包。 其中标蓝绿色的jar包,是写producter需要的, 如果只写consumer的应用并不需要。后面调试顺利。

在此工程中,又发现了一个问题,代码很多地方出现了划掉的情况,这个说明api有更新。

 

\

 

向同事请教,同事说最新的api,已经简化了开发, 标蓝色的jar包无需加入,要修改一下写法。

package com.newland.appkafka;

import org.apache.kafka.clients.producer.KafkaProducer;

import org.apache.kafka.clients.producer.Producer;

import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class ProducerNewApi {

public static void main(String[] args) {

Properties props = new Properties();

props.put("bootstrap.servers", "192.168.136.134:9092,192.168.136.134:9093,192.168.136.134:9094");

props.put("acks", "all");

props.put("retries", 0);

props.put("batch.size", 16384);

props.put("linger.ms", 1);

props.put("buffer.memory", 33554432);

props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");

props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

Producer producer = new KafkaProducer(props);

for(int i = 0; i < 100; i++)

producer.send(new ProducerRecord("cwqsolotest", "message: " + i++));

producer.close();

}

}

新的api写法,在import 这里就有显著不同,引入的jar包,也只需要kafka的客户端。

在虚拟机上开一个consumer接受消息,执行结果如下:

 

\
上一篇:容器云平台使用体验:DaoCloud
下一篇:Solr入门之官方文档6.0阅读笔记系列(十)
相关文章
图文推荐

关于我们 | 联系我们 | 广告服务 | 投资合作 | 版权申明 | 在线帮助 | 网站地图 | 作品发布 | Vip技术培训 | 举报中心

版权所有: 红黑联盟--致力于做实用的IT技术学习网站