1.獲取所有topic
package com.example.demo;
import java.io.IOException;
import java.util.List;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
public class zookeeper {
public static void main(String[] args) {
String connectString = "172.16.10.211:2181";
int sessionTimeout = 4000;
Watcher watcher = new Watcher() {
public void process(WatchedEvent event) {
}
};
try {
ZooKeeper zooKeeper = new ZooKeeper(connectString, sessionTimeout, watcher);
List<String> list = zooKeeper.getChildren("/brokers/topics", false);
int len = list.size();
for(int i = 1;i < len;i++){
System.out.println(list.get(i));
//此處動態生成消費者
//JavaKafkaConsumerHighAPI example = new JavaKafkaConsumerHighAPI(list.get(i), 1);
//new Thread(example).start();
}
} catch (IOException e) {
e.printStackTrace();
} catch (KeeperException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
2.參考http://www.cnblogs.com/liuming1992/p/6432626.html生成消費者，這里進行了小小的改造
package com.example.demo;
import kafka.consumer.*;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.MessageAndMetadata;
import kafka.serializer.StringDecoder;
import kafka.utils.VerifiableProperties;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
* 自定義簡單Kafka消費者， 使用高級API
* Created by gerry on 12/21.
*/
public class JavaKafkaConsumerHighAPI implements Runnable {
/**
* Kafka數據消費對象
*/
private ConsumerConnector consumer;
/**
* Kafka Topic名稱
*/
private String topic;
/**
* 線程數量，一般就是Topic的分區數量
*/
private int numThreads;
/**
* 線程池
*/
private ExecutorService executorPool;
/**
* 構造函數
*
* @param topic Kafka消息Topic主題
* @param numThreads 處理數據的線程數/可以理解為Topic的分區數
* @param zookeeper Kafka的Zookeeper連接字符串
* @param groupId 該消費者所屬group ID的值
*/
public JavaKafkaConsumerHighAPI(String topic, int numThreads) {
// 1. 創建Kafka連接器
this.consumer = Consumer.createJavaConsumerConnector(createConsumerConfig("172.16.10.211:2181", "test-consumer-group"));
// 2. 數據賦值
this.topic = topic;
this.numThreads = numThreads;
}
@Override
public void run() {
// 1. 指定Topic
Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
topicCountMap.put(this.topic, this.numThreads);
// 2. 指定數據的解碼器
StringDecoder keyDecoder = new StringDecoder(new VerifiableProperties());
StringDecoder valueDecoder = new StringDecoder(new VerifiableProperties());
// 3. 獲取連接數據的迭代器對象集合
/**
* Key: Topic主題
* Value: 對應Topic的數據流讀取器，大小是topicCountMap中指定的topic大小
*/
Map<String, List<KafkaStream<String, String>>> consumerMap = this.consumer.createMessageStreams(topicCountMap, keyDecoder, valueDecoder);
// 4. 從返回結果中獲取對應topic的數據流處理器
List<KafkaStream<String, String>> streams = consumerMap.get(this.topic);
// 5. 創建線程池
this.executorPool = Executors.newFixedThreadPool(this.numThreads);
// 6. 構建數據輸出對象
int threadNumber = 0;
for (final KafkaStream<String, String> stream : streams) {
this.executorPool.submit(new ConsumerKafkaStreamProcesser(stream, threadNumber,topic));
threadNumber++;
}
}
public void shutdown() {
// 1. 關閉和Kafka的連接，這樣會導致stream.hashNext返回false
if (this.consumer != null) {
this.consumer.shutdown();
}
// 2. 關閉線程池，會等待線程的執行完成
if (this.executorPool != null) {
// 2.1 關閉線程池
this.executorPool.shutdown();
// 2.2. 等待關閉完成, 等待五秒
try {
if (!this.executorPool.awaitTermination(5, TimeUnit.SECONDS)) {
System.out.println("Timed out waiting for consumer threads to shut down, exiting uncleanly!!");
}
} catch (InterruptedException e) {
System.out.println("Interrupted during shutdown, exiting uncleanly!!");
}
}
}
/**
* 根據傳入的zk的連接信息和groupID的值創建對應的ConsumerConfig對象
*
* @param zookeeper zk的連接信息，類似於：<br/>
* hadoop-senior01.ibeifeng.com:2181,hadoop-senior02.ibeifeng.com:2181/kafka
* @param groupId 該kafka consumer所屬的group id的值， group id值一樣的kafka consumer會進行負載均衡
* @return Kafka連接信息
*/
private ConsumerConfig createConsumerConfig(String zookeeper, String groupId) {
// 1. 構建屬性對象
Properties prop = new Properties();
// 2. 添加相關屬性
prop.put("group.id", groupId); // 指定分組id
prop.put("zookeeper.connect", zookeeper); // 指定zk的連接url
prop.put("zookeeper.session.timeout.ms", "400"); //
prop.put("zookeeper.sync.time.ms", "200");
prop.put("auto.commit.interval.ms", "1000");
// 3. 構建ConsumerConfig對象
return new ConsumerConfig(prop);
}
/**
* Kafka消費者數據處理線程
*/
public static class ConsumerKafkaStreamProcesser implements Runnable {
// Kafka數據流
private KafkaStream<String, String> stream;
// 線程ID編號
private int threadNumber;
private String topic;
public ConsumerKafkaStreamProcesser(KafkaStream<String, String> stream, int threadNumber,String topic) {
this.stream = stream;
this.threadNumber = threadNumber;
this.topic = topic;
}
@Override
public void run() {
// 1. 獲取數據迭代器
ConsumerIterator<String, String> iter = this.stream.iterator();
// 2. 迭代輸出數據
while (iter.hasNext()) {
// 2.1 獲取數據值
MessageAndMetadata value = iter.next();
// 2.2 輸出
System.out.println(this.threadNumber + "____" + value.offset() +"_____"+ topic + "____" + value.message());
}
// 3. 表示當前線程執行完成
System.out.println("Shutdown Thread:" + this.threadNumber);
}
}
}
3.pom
<dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka_2.11</artifactId> <version>0.8.2.1</version> </dependency>
4.
package com.example.text;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.example.es.Es;
import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
public class KafkaConsumer implements Runnable {
private static Logger logger = LoggerFactory.getLogger(KafkaConsumer.class);
private Map<String, Integer> topicCountMap;
private Properties props;
public KafkaConsumer(Map<String, Integer> topicCountMap, Properties props) {
this.topicCountMap = topicCountMap;
this.props = props;
}
@Override
public void run() {
ConsumerConnector consumer = null;
ExecutorService executor = null;
try {
consumer = Consumer.createJavaConsumerConnector(new ConsumerConfig(props));
Map<String, List<KafkaStream<byte[], byte[]>>> msgStreams = consumer.createMessageStreams(topicCountMap);
for (String topic : topicCountMap.keySet()) {
List<KafkaStream<byte[], byte[]>> msgStreamList = msgStreams.get(topic);
// 使用ExecutorService來調度線程
executor = Executors.newFixedThreadPool(topicCountMap.get(topic));
for (int i = 0; i < msgStreamList.size(); i++) {
KafkaStream<byte[], byte[]> kafkaStream = msgStreamList.get(i);
executor.submit(new HanldMessageThread(kafkaStream, i, topic));
}
}
} catch (Exception e) {
if (consumer != null) {
consumer.shutdown();
}
if (executor != null) {
executor.shutdown();
}
try {
if (!executor.awaitTermination(5000, TimeUnit.MILLISECONDS)) {
logger.error("Timed out waiting for consumer threads to shutdown, exiting uncleanly");
}
} catch (InterruptedException e1) {
logger.error("Interrupted during shutdown, exiting uncleanly");
}
logger.error(e.getMessage());
}
}
}
/**
* 具體處理message的線程
*
* @author Administrator
*
*/
class HanldMessageThread implements Runnable {
private KafkaStream<byte[], byte[]> kafkaStream = null;
private int num = 0;
private String topic;
public HanldMessageThread(KafkaStream<byte[], byte[]> kafkaStream, int num, String topic) {
super();
this.kafkaStream = kafkaStream;
this.num = num;
this.topic = topic;
}
public void run() {
ConsumerIterator<byte[], byte[]> iterator = kafkaStream.iterator();
while (iterator.hasNext()) {
String message = new String(iterator.next().message());
// System.out.println(Thread.currentThread().getName());
// System.out.println(this.num + "____" + topic + "____" + message);
// System.out.println(Thread.currentThread().getId());
// System.out.println("Thread no: " + num + ", message: " + message);
if (topic.startsWith("xrs") || topic.startsWith("meitan") || topic.startsWith("qiyexinxi")) {
Es.setData(message, "xrs_db", topic);
} else if (topic.startsWith("search")) {
Es.setData(message, "pholcus_news_v1", topic);
} else {
Es.setData(message, "pholcus_db", topic);
}
}
}
}
private static Properties props;
static {
props = new Properties();
props.put("zookeeper.connect", "172.16.10.211:2181");
props.put("group.id", "test-consumer-group");
props.put("zookeeper.session.timeout.ms", "400");
props.put("zookeeper.sync.time.ms", "200");
props.put("auto.commit.interval.ms", "1000");
}
Map<String, Integer> topicCountMap = new HashMap<String, Integer>();