13518219792
storm-kafka-client使用的示例分析，相信很多没有经验的人对此束手无策，为此本文总结了问题出现的原因和解决方法，通过这篇文章希望你能解决这个问题。
巴林左旗ssl适用于网站、小程序/APP、API接口等需要进行数据传输应用场景，ssl证书未来市场广阔！成为成都创新互联的ssl证书销售渠道，可以享受市场价格4-6折优惠！如果有意向欢迎电话联系或者加微信：18982081108（备注：SSL证书合作）期待与您的合作！
package hgs.core.sk;
import java.util.Map;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.kafka.spout.ByTopicRecordTranslator;
import org.apache.storm.kafka.spout.KafkaSpout;
import org.apache.storm.kafka.spout.KafkaSpoutConfig;
import org.apache.storm.kafka.spout.KafkaSpoutConfig.FirstPollOffsetStrategy;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.TopologyBuilder;
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;
//参考如下
//https://community.hortonworks.com/articles/87597/how-to-write-topology-with-the-new-kafka-spout-cli.html
//https://github.com/apache/storm/blob/master/examples/storm-kafka-client-examples/src/main/java/org/apache/storm/kafka/spout/KafkaSpoutTopologyMainNamedTopics.java#L52
public class StormKafkaMainTest {
public static void main(String[] args) {
TopologyBuilder builder = new TopologyBuilder();
//该类将传入的kafka记录转换为storm的tuple
ByTopicRecordTranslator brt =
new ByTopicRecordTranslator<>( (r) -> new Values(r.value(),r.topic()),new Fields("values","test7"));
//设置要消费的topic即test7
brt.forTopic("test7", (r) -> new Values(r.value(),r.topic()), new Fields("values","test7"));
//类似之前的SpoutConfig
KafkaSpoutConfig ksc = KafkaSpoutConfig
//bootstrapServers 以及topic(test7)
.builder("bigdata01:9092,bigdata02:9092,bigdata03:9092", "test7")
//设置group.id
.setProp(ConsumerConfig.GROUP_ID_CONFIG, "skc-test")
//设置开始消费的气势位置
.setFirstPollOffsetStrategy(FirstPollOffsetStrategy.LATEST)
//设置提交消费边界的时长间隔
.setOffsetCommitPeriodMs(10_000)
//Translator
.setRecordTranslator(brt)
.build();
builder.setSpout("kafkaspout", new KafkaSpout<>(ksc), 2);
builder.setBolt("mybolt1", new MyboltO(), 4).shuffleGrouping("kafkaspout");
Config config = new Config();
config.setNumWorkers(2);
config.setNumAckers(0);
try {
StormSubmitter.submitTopology("storm-kafka-clients", config, builder.createTopology());
} catch (Exception e) {
e.printStackTrace();
}
/* LocalCluster cu = new LocalCluster();
cu.submitTopology("test", config, builder.createTopology());*/
}
}
class MyboltO extends BaseRichBolt{
private static final long serialVersionUID = 1L;
OutputCollector collector = null;
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
this.collector = collector;
}
public void execute(Tuple input) {
//这里把消息大一出来，在对应的woker下面的日志可以找到打印的内容
String out = input.getString(0);
System.out.println(out);
//collector.ack(input);
}
public void declareOutputFields(OutputFieldsDeclarer declarer) {
}
} pom.xml文件
