Apache Kafka 代码实例

214次阅读

共计 5984 个字符，预计需要花费 15 分钟才能阅读完成。

前提：

已经配置好 kafka。若未安装，可以参照【Apache Kafka 安装升级指南 http://www.linuxidc.com/Linux/2013-11/92753.htm】
已在 eclipse 里面安装 scala 插件。Eclipse Kepler 中在 Help->Eclipse Markectplace 中搜索 Scalar，然后安装即可。
使用 maven 构建 kafka 测试 project 在 eclipse 中。
创建 topic：在 kafka 的安装目录下执行 bin/kafka-create-topic.sh –zookeeper 192.168.20.99:2181 –replica 1 –partition 1 –topic test
启动 consumer：在 kafka 的安装目录下执行 bin/kafka-console-consumer.sh –zookeeper 192.168.20.99:2181 –topic test –from-beginning

pom.xml 文件如下 ：

所有 kafka 依赖的 jar 包都在 com.sksamuel.kafka 下面。其中 kafka 使用的版本是 0.8.0-beta1，kafka 是 2.10。

<groupId>com.iflytek.cpcloud.kafka</groupId>
<artifactId>kafkatest</artifactId>
<version>0.0.1-SNAPSHOT</version>
<packaging>jar</packaging>

<name>kafkatest</name>
<url>http://maven.apache.org</url>

<dependencies>
<dependency>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
<version>1.2.14</version>
</dependency>
<dependency>
<groupId>com.sksamuel.kafka</groupId>
<artifactId>kafka_2.10</artifactId>
<version>0.8.0-beta1</version>
</dependency>
</dependencies>
</project>

然后写一个 kafka producer 的测试程序如下：

package com.iflytek.cpcloud.kafka.kafkatest;

import Java.util.Properties;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

/**
* Test the Kafka Producer
* @author jcsong2
*
*/
public class ProducerTest {
public static void main(String[] args) {
Properties props = new Properties();
props.put(“zk.connect”, “192.168.20.99:2181”);
props.put(“serializer.class”, “kafka.serializer.StringEncoder”);
props.put(“metadata.broker.list”, “192.168.20.99:9092”);
ProducerConfig config = new ProducerConfig(props);
Producer<String, String> producer = new Producer<String, String>(config);
for (int i = 0; i < 10; i++)
producer.send(new KeyedMessage<String, String>(“test”, “test” + i));
}
}

在 consuemr 端可以看到 test0 到 test9 十行输出。

更多详情见请继续阅读下一页的精彩内容 ：http://www.linuxidc.com/Linux/2013-11/92754p2.htm

Kafka 的详细介绍 ：请点这里
Kafka 的下载地址 ：请点这里

相关阅读 ：

分布式发布订阅消息系统 Kafka 架构设计 http://www.linuxidc.com/Linux/2013-11/92751.htm

再写一个 kafka consumer 的测试程序如下：

package com.iflytek.cpcloud.kafka.kafkatest;

import Java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;

public class ConsumerTest extends Thread {
private final ConsumerConnector consumer;
private final String topic;

public static void main(String[] args) {
ConsumerTest consumerThread = new ConsumerTest(“test”);
consumerThread.start();
}

public ConsumerTest(String topic) {
consumer = kafka.consumer.Consumer
.createJavaConsumerConnector(createConsumerConfig());
this.topic = topic;
}

private static ConsumerConfig createConsumerConfig() {
Properties props = new Properties();
props.put(“zookeeper.connect”, “192.168.20.99:2181”);
props.put(“group.id”, “0”);
props.put(“zookeeper.session.timeout.ms”, “400000”);
props.put(“zookeeper.sync.time.ms”, “200”);
props.put(“auto.commit.interval.ms”, “1000”);

return new ConsumerConfig(props);

}

public void run() {
Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
topicCountMap.put(topic, new Integer(1));
Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer
.createMessageStreams(topicCountMap);
KafkaStream<byte[], byte[]> stream = consumerMap.get(topic).get(0);
ConsumerIterator<byte[], byte[]> it = stream.iterator();
while (it.hasNext())
System.out.println(new String(it.next().message()));
}
}

在 kafka-console-producer 端输入的数据会回显到 eclipse 的 console 中。

以上程序参考 kafka-0.8.0-bata1 中的 example。

前提：

已经配置好 kafka。若未安装，可以参照【Apache Kafka 安装升级指南 http://www.linuxidc.com/Linux/2013-11/92753.htm】
已在 eclipse 里面安装 scala 插件。Eclipse Kepler 中在 Help->Eclipse Markectplace 中搜索 Scalar，然后安装即可。
使用 maven 构建 kafka 测试 project 在 eclipse 中。
创建 topic：在 kafka 的安装目录下执行 bin/kafka-create-topic.sh –zookeeper 192.168.20.99:2181 –replica 1 –partition 1 –topic test
启动 consumer：在 kafka 的安装目录下执行 bin/kafka-console-consumer.sh –zookeeper 192.168.20.99:2181 –topic test –from-beginning

pom.xml 文件如下 ：

所有 kafka 依赖的 jar 包都在 com.sksamuel.kafka 下面。其中 kafka 使用的版本是 0.8.0-beta1，kafka 是 2.10。

<groupId>com.iflytek.cpcloud.kafka</groupId>
<artifactId>kafkatest</artifactId>
<version>0.0.1-SNAPSHOT</version>
<packaging>jar</packaging>

<name>kafkatest</name>
<url>http://maven.apache.org</url>

<dependencies>
<dependency>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
<version>1.2.14</version>
</dependency>
<dependency>
<groupId>com.sksamuel.kafka</groupId>
<artifactId>kafka_2.10</artifactId>
<version>0.8.0-beta1</version>
</dependency>
</dependencies>
</project>

然后写一个 kafka producer 的测试程序如下：

package com.iflytek.cpcloud.kafka.kafkatest;

import Java.util.Properties;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

/**
* Test the Kafka Producer
* @author jcsong2
*
*/
public class ProducerTest {
public static void main(String[] args) {
Properties props = new Properties();
props.put(“zk.connect”, “192.168.20.99:2181”);
props.put(“serializer.class”, “kafka.serializer.StringEncoder”);
props.put(“metadata.broker.list”, “192.168.20.99:9092”);
ProducerConfig config = new ProducerConfig(props);
Producer<String, String> producer = new Producer<String, String>(config);
for (int i = 0; i < 10; i++)
producer.send(new KeyedMessage<String, String>(“test”, “test” + i));
}
}

在 consuemr 端可以看到 test0 到 test9 十行输出。

更多详情见请继续阅读下一页的精彩内容 ：http://www.linuxidc.com/Linux/2013-11/92754p2.htm

Kafka 的详细介绍 ：请点这里
Kafka 的下载地址 ：请点这里

相关阅读 ：

分布式发布订阅消息系统 Kafka 架构设计 http://www.linuxidc.com/Linux/2013-11/92751.htm

正文完

星哥玩云-微信公众号