Kafka-2.11学习笔记（三）JavaApi访问kafka

鲁春利的工作笔记，[b]好记性不如烂笔头[/b]

Kafka底层是基于Scala语言实现的，但是也提供了Java的API接口。
Java实现的消息生产者

package com.lucl.kafka.simple;

import java.util.Properties;

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

import org.apache.log4j.Logger;

/**
* <p> Copyright: Copyright (c) 2015 </p>
*
* <p> Date : 2015-11-17 21:42:50 </p>
*
* <p> Description : JavaApi for kafka producer </p>
*
* @author luchunli
*
* @version 1.0
*
*/
public class SimpleKafkaProducer {
private static final Logger logger = Logger.getLogger(SimpleKafkaProducer.class);
/**
*
*/
private void execMsgSend() {
Properties props = new Properties();
props.put("metadata.broker.list", "192.168.137.117:9092");
props.put("serializer.class", "kafka.serializer.StringEncoder");
props.put("key.serializer.class", "kafka.serializer.StringEncoder");
props.put("request.required.acks", "0");

ProducerConfig config = new ProducerConfig(props);

logger.info("set config info(" + config + ") ok.");

Producer<String, String> procuder = new Producer<>(config);

String topic = "mytopic";
for (int i = 1; i <= 10; i++) {
String value = "value_" + i;
KeyedMessage<String, String> msg = new KeyedMessage<String, String>(topic, value);
procuder.send(msg);
}
logger.info("send message over.");

procuder.close();
}

/**
* @param args
*/
public static void main(String[] args) {
SimpleKafkaProducer simpleProducer = new SimpleKafkaProducer();
simpleProducer.execMsgSend();
}

}

此时通过控制台方式启动的消费者，可以看到生产者生产的数据被消费者消费了：

[hadoop@nnode kafka0.8.2.1]$ bin/kafka-console-consumer.sh --zookeeper nnode:2181,dnode1:2181,dnode2:2181 --topic mytopic --from-beginning
hello world
this is my first message
value_1
value_2
value_3
value_4
value_5
value_6
value_7
value_8
value_9
value_10

Java实现的消息消费者

package com.lucl.kafka.simple;

import java.util.HashMap;
import java.util.Iterator;
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;
import kafka.message.MessageAndMetadata;
import kafka.serializer.Decoder;
import kafka.serializer.StringDecoder;
import kafka.utils.VerifiableProperties;

import org.apache.log4j.Logger;

/**
* <p> Copyright: Copyright (c) 2015 </p>
*
* <p> Date : 2015-11-17 21:42:50 </p>
*
* <p> Description : JavaApi for kafka consumer </p>
*
* @author luchunli
*
* @version 1.0
*
*/
public class SimpleKafkaConsumer {
private static final Logger logger = Logger.getLogger(SimpleKafkaConsumer.class);

/**
*
*/
private void execMsgConsume() {
Properties props = new Properties();
props.put("zookeeper.connect", "nnode:2181,dnode1:2181,dnode2:2181");
props.put("group.id", "group-1");
props.put("serializer.class", "kafka.serializer.StringEncoder");

ConsumerConfig config = new ConsumerConfig(props);
ConsumerConnector consumer = Consumer.createJavaConsumerConnector(config);

Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
topicCountMap.put("mytopic", 1);
Decoder<String> keyDecoder = new StringDecoder(new VerifiableProperties());
Decoder<String> valueDecoder = new StringDecoder(new VerifiableProperties());
Map<String, List<KafkaStream<String, String>>> createMessageStreams = consumer.createMessageStreams(topicCountMap, keyDecoder, valueDecoder);
for (Iterator<String> it = createMessageStreams.keySet().iterator(); it.hasNext(); ) {
String key = it.next();
logger.info("The key of the createMessageStreams is " + key);
List<KafkaStream<String, String>> values = createMessageStreams.get(key);
for (KafkaStream<String, String> value : values) {
ConsumerIterator<String, String> consumerIt = value.iterator();
while (consumerIt.hasNext()) {
MessageAndMetadata<String, String> data = consumerIt.next();
logger.info("The message got by consuer is " + data.message());
}
}
}

}

/**
* @param args
*/
public static void main(String[] args) {
SimpleKafkaConsumer simpleConsumer = new SimpleKafkaConsumer();
simpleConsumer.execMsgConsume();
}

}

启动Consumer程序，然后再启动Producer程序，此时Consumer端输出内容如下：

23:37:30,411  INFO SimpleKafkaConsumer:55 - The key of the createMessageStreams is mytopic
23:37:30,433  INFO VerifiableProperties:68 - Verifying properties
23:37:30,433  INFO VerifiableProperties:68 - Property client.id is overridden to group-1
23:37:30,433  INFO VerifiableProperties:68 - Property metadata.broker.list is overridden to nnode:9092
23:37:30,433  INFO VerifiableProperties:68 - Property request.timeout.ms is overridden to 30000
23:37:30,451  INFO ClientUtils$:68 - Fetching metadata from broker id:117,host:nnode,port:9092 with correlation id 0 for 1 topic(s) Set(mytopic)
23:37:30,453  INFO SyncProducer:68 - Connected to nnode:9092 for producing
23:37:30,486  INFO SyncProducer:68 - Disconnecting from nnode:9092
23:37:30,528  INFO ConsumerFetcherThread:68 - [ConsumerFetcherThread-group-1_LuchunliPC-1447947448911-f949268d-0-117], Starting
23:37:30,546  INFO ConsumerFetcherManager:68 - [ConsumerFetcherManager-1447947449115] Added fetcher for partitions ArrayBuffer([[mytopic,0], initOffset -1 to broker id:117,host:nnode,port:9092] )
23:37:52,466  INFO SimpleKafkaConsumer:61 - The message got by consuer is value_1
23:37:52,466  INFO SimpleKafkaConsumer:61 - The message got by consuer is value_2
23:37:52,466  INFO SimpleKafkaConsumer:61 - The message got by consuer is value_3
23:37:52,466  INFO SimpleKafkaConsumer:61 - The message got by consuer is value_4
23:37:52,466  INFO SimpleKafkaConsumer:61 - The message got by consuer is value_5
23:37:52,466  INFO SimpleKafkaConsumer:61 - The message got by consuer is value_6
23:37:52,466  INFO SimpleKafkaConsumer:61 - The message got by consuer is value_7
23:37:52,469  INFO SimpleKafkaConsumer:61 - The message got by consuer is value_8
23:37:52,469  INFO SimpleKafkaConsumer:61 - The message got by consuer is value_9
23:37:52,469  INFO SimpleKafkaConsumer:61 - The message got by consuer is value_10
23:39:11,351  INFO ClientCnxn:1096 - Client session timed out, have not heard from server in 4000ms for sessionid 0x3512026596f0001, closing socket connection and attempting reconnect
23:39:11,452  INFO ZkClient:449 - zookeeper state changed (Disconnected)

本文出自 “闷葫芦的世界” 博客，请务必保留此出处http://luchunli.blog.51cto.com/2368057/1714857