溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
storm-kafka-client使用的示例分析，相信很多沒有經驗的人對此束手無策，為此本文總結了問題出現的原因和解決方法，通過這篇文章希望你能解決這個問題。
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<String,String> 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<String,String> 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文件
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>hgs</groupId> <artifactId>core.sk</artifactId> <version>1.0.0-SNAPSHOT</version> <packaging>jar</packaging> <name>core.sk</name> <url>http://maven.apache.org</url> <properties> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> </properties> <dependencies> <dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> <version>3.8.1</version> <scope>test</scope> </dependency> <!-- <dependency> <groupId>org.apache.storm</groupId> <artifactId>storm-kafka</artifactId> <version>1.1.3</version> </dependency> --> <dependency> <groupId>org.apache.storm</groupId> <artifactId>storm-kafka-client</artifactId> <version>1.1.3</version> </dependency> <dependency> <groupId>org.apache.storm</groupId> <artifactId>storm-core</artifactId> <version>1.1.3</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka_2.11</artifactId> <version>1.0.0</version> <exclusions> <exclusion> <groupId>org.slf4j</groupId> <artifactId>slf4j-log4j12</artifactId> </exclusion> <exclusion> <groupId>org.apache.zookeeper</groupId> <artifactId>zookeeper</artifactId> </exclusion> </exclusions> </dependency> <!-- <dependency> <groupId>org.apache.storm</groupId> <artifactId>storm-kafka-monitor</artifactId> <version>1.2.2</version> </dependency> --> <!-- <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>0.8.2.1</version> </dependency> --> <dependency> <groupId>org.clojure</groupId> <artifactId>clojure</artifactId> <version>1.7.0</version> </dependency> <!-- 嘗試了很多次 都會有這個錯誤： java.lang.NullPointerException at org.apache.storm.kafka.monitor.KafkaOffsetLagUtil.getOffsetLags(KafkaOffsetLagUtil.java:272) 最后修改為kafka相應的kafka-clients版本后問題得到解決，應該是該出的問題 --> <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>1.0.0</version> </dependency> </dependencies> <build> <plugins> <plugin> <artifactId>maven-assembly-plugin</artifactId> <version>2.2</version> <configuration> <archive> <manifest> <!-- 我運行這個jar所運行的主類 --> <mainClass>hgs.core.sk.StormKafkaMainTest</mainClass> </manifest> </archive> <descriptorRefs> <descriptorRef> <!-- 必須是這樣寫 --> jar-with-dependencies </descriptorRef> </descriptorRefs> </configuration> <executions> <execution> <id>make-assembly</id> <phase>package</phase> <goals> <goal>single</goal> </goals> </execution> </executions> </plugin> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-compiler-plugin</artifactId> <configuration> <source>1.8</source> <target>1.8</target> </configuration> </plugin> </plugins> </build> </project>
//以下為lambda表達式，因為在上面用大了，所以在這兒記錄一下，以免以后看不懂
import java.util.UUID;
import org.junit.jupiter.api.Test;
public class TEst {
@Test
public void sysConfig() {
String[] ags = {"his is my first storm program so i hope it will success",
"i love bascketball",
"the day of my birthday i was alone"};
String uuid = UUID.randomUUID().toString();
String nexttuple= ags[new Random().nextInt(ags.length)];
System.out.println(nexttuple);
}
@Test
public void lambdaTest() {
int b = 100;
//該出返回10*a的值、
//"(a) -> 10*a" 相當于 new testinter<T>();
printPerson((a) -> 10*a) ;
}
void printPerson( testinter<Integer> t) {
//穿過來的t需要一個參數a 即下面借口中定義的方法sysoutitems(int a )
System.out.println(t.sysoutitems(100));
};
}
//定義接口，在lambda表達式運用中，必須為借口，并且借口只能有一個方法
interface testinter<T>{
T sysoutitems(int a );
//void aAndb(int a, int b );
}看完上述內容，你們掌握storm-kafka-client使用的示例分析的方法了嗎？如果還想學到更多技能或想了解更多相關內容，歡迎關注億速云行業資訊頻道，感謝各位的閱讀！
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
