旧版本高级Api封装：
package xxxxxx;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import java.util.Properties;
public class KafkaProducerTest implements Runnable {
private final kafka.javaapi.producer.Producer producer;
private final String topic;
private final Properties props = new Properties();
private final String message;
public KafkaProducerTest(String topic, String message) {
props.put("serializer.class", "kafka.serializer.StringEncoder");
props.put("metadata.broker.list", "127.0.0.1:9092");
// Use random partitioner. Don't need the key type. Just set it to Integer.
// The message is of type String.
producer = new kafka.javaapi.producer.Producer(new ProducerConfig(props));
this.topic = topic;
this.message = message;
}
@Override
public void run() {
producer.send(new KeyedMessage(topic, message));
System.out.println("发送kafka消息:"+message);
}
public static void main(String[] arg){
for(int icount=0;icount<100;icount++){
KafkaProducerTest kafkaProducer = new KafkaProducerTest("myTopic", "hello world!");
Thread thread = new Thread(kafkaProducer);
thread.start();
}
}
}
package xxxxxxxxx;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
public class KafkaConsumerTest implements Runnable{
private final ConsumerConnector consumer;
private final String topic;
public KafkaConsumerTest(String topic)
{
consumer = kafka.consumer.Consumer.createJavaConsumerConnector(
createConsumerConfig());
this.topic = topic;
}
private static ConsumerConfig createConsumerConfig()
{
Properties props = new Properties();
props.put("zookeeper.connect", "127.0.0.1:2181");
props.put("group.id", "myGroup");
props.put("zookeeper.session.timeout.ms", "400");
props.put("zookeeper.sync.time.ms", "200");
props.put("auto.commit.interval.ms", "1000");
return new ConsumerConfig(props);
}
@Override
public void run() {
Map topicCountMap = new HashMap();
topicCountMap.put(topic, new Integer(1));
Map>> consumerMap = consumer.createMessageStreams(topicCountMap);
System.out.println("开始启动.....");
try {
KafkaStream stream = consumerMap.get(topic).get(0);
ConsumerIterator it = stream.iterator();
while (it.hasNext()) {
System.out.println("消费信息:" + new String(it.next().message()));
}
}catch (Exception e){
e.printStackTrace();
}
}
public static void main(String [] arg){
KafkaConsumerTest kafkaConsumer = new KafkaConsumerTest("myTopic");
Thread thread = new Thread(kafkaConsumer);
thread.start();
}
}
注意配置信息：
1、pom.xml文件中增加plugin:
org.mortbay.jetty
jetty-maven-plugin
7.6.10.v20130312
${basedir}/webapp
stop
55855
0
8085
60000
/
global.config.path
${basedir}/src/main/resources/config
APPID
spider.mySpider
2、读取配置文件信息的工具类：
package xxxxxx;
import com.alibaba.fastjson.JSON;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.util.Properties;
public class ReadConfigUtil {
private static Properties prop;
private static String globalConfigPath = System.getProperty("global.config.path");
static {
prop = new Properties();
// 遍历所有.properties文件
File cfgFiles = new File(globalConfigPath);
if (cfgFiles.exists()) {
File[] fs = cfgFiles.listFiles();
for (File confFile : fs) {
try {
String confName = confFile.getName();
if (confName.endsWith(".properties")) {
FileInputStream fis = new FileInputStream(confFile);
prop.load(fis);
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
public static Properties getProp() {
return prop;
}
public static String getValueByKey(String key) {
return getProp().getProperty(key);
}
}
kafka版本引用：
org.apache.kafka
kafka_2.11
0.8.2.1
slf4j-log4j12
org.slf4j
