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"));
}
}