storm和kafka整合

storm和kafka整合

依赖

<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-kafka-client</artifactId>
<version>1.2.2</version>
<scope>provided</scope>
</dependency> <dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.1.0</version>
</dependency> <dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-core</artifactId>
<version>1.2.2</version>
<scope>provided</scope>
</dependency>

App

package test;

import java.util.List;
import java.util.concurrent.TimeUnit; import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.storm.kafka.spout.Func;
import org.apache.storm.kafka.spout.KafkaSpout;
import org.apache.storm.kafka.spout.KafkaSpoutConfig;
import org.apache.storm.kafka.spout.KafkaSpoutRetryExponentialBackoff;
import org.apache.storm.kafka.spout.KafkaSpoutConfig.FirstPollOffsetStrategy;
import org.apache.storm.kafka.spout.KafkaSpoutRetryExponentialBackoff.TimeInterval;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values; public class App
{
public static void main( String[] args ) throws Exception{ KafkaSpoutConfig<String, String> conf = KafkaSpoutConfig
.builder("worker1:9092,worker2:9092,worker3:9092", "test") // 你的kafka集群地址和topic
.setProp(ConsumerConfig.GROUP_ID_CONFIG, "consumer") // 设置消费者组,随便写
.setProp(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 1024 * 1024 * 4)
// .setRecordTranslator(new MyRecordTranslator())
.setRecordTranslator( // 翻译函数,就是将消息过滤下,具体操作自己玩
new MyRecordTranslator(),
new Fields("word")
)
.setRetry( // 某条消息处理失败的策略
new KafkaSpoutRetryExponentialBackoff(
new TimeInterval(500L, TimeUnit.MICROSECONDS),
TimeInterval.milliSeconds(2),
Integer.MAX_VALUE,
TimeInterval.seconds(10)
)
)
.setOffsetCommitPeriodMs(10000)
.setFirstPollOffsetStrategy(FirstPollOffsetStrategy.LATEST)
.setMaxUncommittedOffsets(250)
.build(); TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("KafkaSpout", new KafkaSpout<String, String>(conf), 1);
builder.setBolt("Recieve", new RecieveBolt(), 1).globalGrouping("KafkaSpout");
builder.setBolt("Consume", new ConsumeBolt(), 1).globalGrouping("Recieve");
builder.createTopology(); // 集群运行
// Config config = new Config();
// config.setNumWorkers(3);
// config.setDebug(true);
// StormSubmitter.submitTopology("teststorm", config, builder.createTopology()); // 本地测试
// Config config = new Config();
// config.setNumWorkers(3);
// config.setDebug(true);
// config.setMaxTaskParallelism(20);
// LocalCluster cluster = new LocalCluster();
// cluster.submitTopology("teststorm", config, builder.createTopology());
// Utils.sleep(60000);
// // 执行完毕,关闭cluster
// cluster.shutdown();
}
} class MyRecordTranslator implements Func<ConsumerRecord<String, String>, List<Object>> { private static final long serialVersionUID = 1L; @Override
public List<Object> apply(ConsumerRecord<String, String> record) {
return new Values(record.value());
} }

ConsumeBolt

package test;

import java.io.FileWriter;
import java.io.IOException;
import java.util.Map;
import java.util.UUID; import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple; public class ConsumeBolt extends BaseRichBolt { private static final long serialVersionUID = -7114915627898482737L; private FileWriter fileWriter = null; private OutputCollector collector; public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) { this.collector = collector; try {
fileWriter = new FileWriter("/usr/local/tmpdata/" + UUID.randomUUID());
// fileWriter = new FileWriter("C:\\Users\\26401\\Desktop\\test\\" + UUID.randomUUID());
} catch (IOException e) {
throw new RuntimeException(e);
}
} public void execute(Tuple tuple) { try {
String word = tuple.getStringByField("word") + "......." + "\n";
fileWriter.write(word);
fileWriter.flush();
System.out.println(word);
} catch (IOException e) {
throw new RuntimeException(e);
}
} public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) { }
}

RecieveBolt

package test;

import java.util.Map;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values; public class RecieveBolt extends BaseRichBolt { private static final long serialVersionUID = -4758047349803579486L; private OutputCollector collector; public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
this.collector = collector;
} public void execute(Tuple tuple) {
// 将spout传递过来的tuple值进行转换
this.collector.emit(new Values(tuple.getStringByField("word") + "!!!"));
} // 声明发送消息的字段名
public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
outputFieldsDeclarer.declare(new Fields("word"));
}
}
上一篇:Jade 模板引擎使用


下一篇:Jade模板引擎使用详解