Skip to content

Commit c43288b

Browse files
author
centmeng
committed
kafka和storm
1 parent 702f500 commit c43288b

6 files changed

Lines changed: 278 additions & 0 deletions

File tree

‎src/com/msj/kafka/KafkaConsumer.java‎

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,52 @@
1+
package com.msj.kafka;
2+
3+
import kafka.consumer.ConsumerConfig;
4+
import kafka.consumer.ConsumerIterator;
5+
import kafka.consumer.KafkaStream;
6+
import kafka.javaapi.consumer.ConsumerConnector;
7+
import kafka.serializer.StringDecoder;
8+
import kafka.utils.VerifiableProperties;
9+
10+
import java.util.HashMap;
11+
import java.util.List;
12+
import java.util.Map;
13+
import java.util.Properties;
14+
15+
public class KafkaConsumer {
16+
17+
public static final String topic = "test";
18+
19+
public static void main(String[] args) {
20+
21+
Properties props = new Properties();
22+
props.put("zookeeper.connect", "192.168.1.114:2181,192.168.1.115:2181,192.168.1.116:2181");
23+
//group 代表一个消费组
24+
props.put("group.id", "group1");
25+
//zk连接超时
26+
props.put("zookeeper.session.timeout.ms", "4000");
27+
props.put("zookeeper.sync.time.ms", "200");
28+
props.put("auto.commit.interval.ms", "1000");
29+
props.put("auto.offset.reset", "smallest");
30+
//序列化类
31+
props.put("serializer.class", "kafka.serializer.StringEncoder");
32+
33+
ConsumerConfig config = new ConsumerConfig(props);
34+
ConsumerConnector consumer = kafka.consumer.Consumer.createJavaConsumerConnector(config);
35+
36+
Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
37+
38+
topicCountMap.put(topic, new Integer(1));
39+
40+
StringDecoder keyDecoder = new StringDecoder(new VerifiableProperties());
41+
StringDecoder valueDecoder = new StringDecoder(new VerifiableProperties());
42+
43+
Map<String, List<KafkaStream<String, String>>> consumerMap =
44+
consumer.createMessageStreams(topicCountMap,keyDecoder,valueDecoder);
45+
46+
KafkaStream<String, String> stream = consumerMap.get(topic).get(0);
47+
ConsumerIterator<String, String> it = stream.iterator();
48+
49+
while (it.hasNext())
50+
System.out.println(it.next().message());
51+
}
52+
}

‎src/com/msj/kafka/KafkaProducer.java‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,30 @@
1+
package com.msj.kafka;
2+
3+
import kafka.javaapi.producer.Producer;
4+
import kafka.producer.KeyedMessage;
5+
import kafka.producer.ProducerConfig;
6+
import kafka.serializer.StringEncoder;
7+
8+
import java.util.Properties;
9+
import java.util.concurrent.TimeUnit;
10+
11+
public class KafkaProducer {
12+
13+
public static final String topic = "test";
14+
15+
public static void main(String[] args) throws Exception {
16+
Properties properties = new Properties();
17+
properties.put("zookeeper.connect", "192.168.1.114:2181,192.168.1.115:2181,192.168.1.116:2181"); //声明zk
18+
properties.put("serializer.class", StringEncoder.class.getName());
19+
properties.put("metadata.broker.list", "192.168.1.114:9092"); // 声明kafka broker
20+
properties.put("request.required.acks", "1");
21+
Producer producer = new Producer<Integer, String>(new ProducerConfig(properties));
22+
for(int i=0; i < 10; i++){
23+
producer.send(new KeyedMessage<Integer, String>(topic, "hello kafka" + i));
24+
System.out.println("send message: " + "hello kafka" + i);
25+
TimeUnit.SECONDS.sleep(1);
26+
}
27+
producer.close();
28+
}
29+
30+
}

‎src/com/msj/storm/kafka/KafkaTopology.java‎

Lines changed: 74 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,74 @@
1+
package com.msj.storm.kafka;
2+
3+
import backtype.storm.Config;
4+
import backtype.storm.LocalCluster;
5+
import backtype.storm.generated.AlreadyAliveException;
6+
import backtype.storm.generated.InvalidTopologyException;
7+
import backtype.storm.topology.TopologyBuilder;
8+
9+
/**
10+
* @author Vincent.M mengshaojie@188.com
11+
* @date 2017/5/11 下午2:57
12+
* @copyright ©2017 孟少杰 All Rights Reserved
13+
* @desc kafka与storm，缺少jar包，所以需要jar包地方屏蔽掉
14+
*/
15+
public class KafkaTopology {
16+
public static void main(String[] args) throws
17+
AlreadyAliveException, InvalidTopologyException {
18+
// zookeeper hosts for the Kafka cluster
19+
//缺jar包
20+
// ZkHosts zkHosts = new ZkHosts("192.168.1.114:2181");
21+
22+
// Create the KafkaSpout configuartion
23+
// Second argument is the topic name
24+
// Third argument is the zookeeper root for Kafka
25+
// Fourth argument is consumer group id
26+
//缺jar包
27+
// SpoutConfig kafkaConfig = new SpoutConfig(zkHosts,
28+
// "words_topic", "", "id7");
29+
30+
// Specify that the kafka messages are String
31+
//缺jar包
32+
// kafkaConfig.scheme = new SchemeAsMultiScheme(new StringScheme());
33+
34+
// We want to consume all the first messages in the topic everytime
35+
// we run the topology to help in debugging. In production, this
36+
// property should be false
37+
//缺jar包
38+
// kafkaConfig.forceFromStart = true;
39+
40+
// Now we create the topology
41+
TopologyBuilder builder = new TopologyBuilder();
42+
43+
// set the kafka spout class
44+
//缺jar包
45+
// builder.setSpout("KafkaSpout", new KafkaSpout(kafkaConfig), 1);
46+
47+
// configure the bolts
48+
builder.setBolt("SentenceBolt", new com.msj.storm.kafka.SentenceBolt(), 1).globalGrouping("KafkaSpout");
49+
builder.setBolt("PrinterBolt", new com.msj.storm.kafka.PrinterBolt(), 1).globalGrouping("SentenceBolt");
50+
51+
52+
// create an instance of LocalCluster class for executing topology in local mode.
53+
LocalCluster cluster = new LocalCluster();
54+
Config conf = new Config();
55+
56+
// Submit topology for execution
57+
cluster.submitTopology("KafkaToplogy", conf, builder.createTopology());
58+
59+
60+
try {
61+
// Wait for some time before exiting
62+
System.out.println("Waiting to consume from kafka");
63+
Thread.sleep(10000);
64+
} catch (Exception exception) {
65+
System.out.println("Thread interrupted exception : " + exception);
66+
}
67+
68+
// kill the KafkaTopology
69+
cluster.killTopology("KafkaToplogy");
70+
71+
// shut down the storm test cluster
72+
cluster.shutdown();
73+
}
74+
}

‎src/com/msj/storm/kafka/PrinterBolt.java‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,19 @@
1+
package com.msj.storm.kafka;
2+
3+
import backtype.storm.topology.BasicOutputCollector;
4+
import backtype.storm.topology.OutputFieldsDeclarer;
5+
import backtype.storm.topology.base.BaseBasicBolt;
6+
import backtype.storm.tuple.Tuple;
7+
8+
public class PrinterBolt extends BaseBasicBolt {
9+
10+
public void execute(Tuple input, BasicOutputCollector collector) {
11+
// get the sentence from the tuple and print it
12+
String sentence = input.getString(0);
13+
System.out.println("Received Sentence:" + sentence);
14+
}
15+
16+
public void declareOutputFields(OutputFieldsDeclarer declarer) {
17+
// we don't emit anything
18+
}
19+
}

‎src/com/msj/storm/kafka/SentenceBolt.java‎

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,49 @@
1+
package com.msj.storm.kafka;
2+
3+
import backtype.storm.topology.BasicOutputCollector;
4+
import backtype.storm.topology.OutputFieldsDeclarer;
5+
import backtype.storm.topology.base.BaseBasicBolt;
6+
import backtype.storm.tuple.Fields;
7+
import backtype.storm.tuple.Tuple;
8+
import com.google.common.collect.ImmutableList;
9+
import org.apache.commons.lang.StringUtils;
10+
11+
import java.util.ArrayList;
12+
import java.util.List;
13+
14+
public class SentenceBolt extends BaseBasicBolt {
15+
16+
// list used for aggregating the words
17+
private List<String> words = new ArrayList<String>();
18+
19+
public void execute(Tuple input, BasicOutputCollector collector) {
20+
// Get the word from the tuple
21+
String word = input.getString(0);
22+
23+
if(StringUtils.isBlank(word)){
24+
// ignore blank lines
25+
return;
26+
}
27+
28+
System.out.println("Received Word:" + word);
29+
30+
// add word to current list of words
31+
words.add(word);
32+
33+
if (word.endsWith(".")) {
34+
// word ends with '.' which means this is the end of
35+
// the sentence publishes a sentence tuple
36+
collector.emit(ImmutableList.of(
37+
(Object) StringUtils.join(words, ' ')));
38+
39+
// and reset the words list.
40+
words.clear();
41+
}
42+
}
43+
44+
public void declareOutputFields(OutputFieldsDeclarer declarer) {
45+
// here we declare we will be emitting tuples with
46+
// a single field called "sentence"
47+
declarer.declare(new Fields("sentence"));
48+
}
49+
}

‎src/com/msj/storm/kafka/WordsProducer.java‎

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,54 @@
1+
package com.msj.storm.kafka;
2+
3+
import kafka.javaapi.producer.Producer;
4+
import kafka.producer.KeyedMessage;
5+
import kafka.producer.ProducerConfig;
6+
7+
import java.util.Properties;
8+
9+
public class WordsProducer {
10+
11+
public static void main(String[] args) {
12+
// Build the configuration required for connecting to Kafka
13+
Properties props = new Properties();
14+
15+
// List of Kafka brokers. Complete list of brokers is not
16+
// required as the producer will auto discover the rest of
17+
// the brokers. Change this to suit your deployment.
18+
props.put("metadata.broker.list", "192.168.1.114:9092");
19+
20+
// Serializer used for sending data to kafka. Since we are sending string,
21+
// we are using StringEncoder.
22+
props.put("serializer.class", "kafka.serializer.StringEncoder");
23+
24+
// We want acks from Kafka that messages are properly received.
25+
props.put("request.required.acks", "1");
26+
27+
// Create the producer instance
28+
ProducerConfig config = new ProducerConfig(props);
29+
Producer<String, String> producer = new Producer<String, String>(config);
30+
31+
// Now we break each word from the paragraph
32+
for (String word : METAMORPHOSIS_OPENING_PARA.split("\\s")) {
33+
// Create message to be sent to "words_topic" topic with the word
34+
KeyedMessage<String, String> data =
35+
new KeyedMessage<String, String>("words_topic", word);
36+
37+
// Send the message
38+
producer.send(data);
39+
}
40+
41+
System.out.println("Produced data");
42+
43+
// close the producer
44+
producer.close();
45+
}
46+
47+
// First paragraph from Franz Kafka's Metamorphosis
48+
private static String METAMORPHOSIS_OPENING_PARA =
49+
"One morning, when Gregor Samsa woke from troubled dreams, "
50+
+ "he found himself transformed in his bed into a horrible "
51+
+ "vermin. He lay on his armour-like back, and if he lifted "
52+
+ "his head a little he could see his brown belly, slightly "
53+
+ "domed and divided by arches into stiff sections.";
54+
}

0 commit comments

Comments
 (0)