我的风暴螺栓不能反序列化集群模式

vwkv1x7d  于 2021-06-24  发布在  Storm
关注(0)|答案(1)|浏览(266)

我使用springboot和storm做了一个演示，它在本地模式下工作，但是在提交jar时在集群模式下报告一个错误

./storm jar storm-demo3-0.0.1-SNAPSHOT.jar org.springframework.boot.loader.JarLauncher simpleBoot

当我移动springboot并打包maven编译器插件时，它就可以很好地工作了

<build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <source>8</source>
                    <target>8</target>
                </configuration>
            </plugin>
        </plugins>
    </build>

这是主管的错误

java.lang.RuntimeException: java.lang.ClassNotFoundException: com.fosung.share.stormdemo3.bolt.FilterBolt
        at org.apache.storm.utils.Utils.javaDeserialize(Utils.java:259) ~[storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.utils.Utils.getSetComponentObject(Utils.java:507) ~[storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.daemon.task$get_task_object.invoke(task.clj:76) ~[storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.daemon.task$mk_task_data$fn__6524.invoke(task.clj:180) ~[storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.util$assoc_apply_self.invoke(util.clj:931) ~[storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.daemon.task$mk_task_data.invoke(task.clj:172) ~[storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.daemon.task$mk_task.invoke(task.clj:184) ~[storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.daemon.executor$mk_executor$fn__10662.invoke(executor.clj:379) ~[storm-core-1.2.2.jar:1.2.2]
        at clojure.core$map$fn__4553.invoke(core.clj:2622) ~[clojure-1.7.0.jar:?]
        at clojure.lang.LazySeq.sval(LazySeq.java:40) ~[clojure-1.7.0.jar:?]
        at clojure.lang.LazySeq.seq(LazySeq.java:49) ~[clojure-1.7.0.jar:?]
        at clojure.lang.RT.seq(RT.java:507) ~[clojure-1.7.0.jar:?]
        at clojure.core$seq__4128.invoke(core.clj:137) ~[clojure-1.7.0.jar:?]
        at clojure.core.protocols$seq_reduce.invoke(protocols.clj:30) ~[clojure-1.7.0.jar:?]
        at clojure.core.protocols$fn__6506.invoke(protocols.clj:101) ~[clojure-1.7.0.jar:?]
        at clojure.core.protocols$fn__6452$G__6447__6465.invoke(protocols.clj:13) ~[clojure-1.7.0.jar:?]
        at clojure.core$reduce.invoke(core.clj:6519) ~[clojure-1.7.0.jar:?]
        at clojure.core$into.invoke(core.clj:6600) ~[clojure-1.7.0.jar:?]
        at org.apache.storm.daemon.executor$mk_executor.invoke(executor.clj:380) ~[storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.daemon.worker$fn__11300$exec_fn__2470__auto__$reify__11302$iter__11307__11311$fn__11312.invoke(worker.clj:663) ~[storm-core-1.2.2.jar:1.2.2]
        at clojure.lang.LazySeq.sval(LazySeq.java:40) ~[clojure-1.7.0.jar:?]
        at clojure.lang.LazySeq.seq(LazySeq.java:49) ~[clojure-1.7.0.jar:?]
        at clojure.lang.RT.seq(RT.java:507) ~[clojure-1.7.0.jar:?]
        at clojure.core$seq__4128.invoke(core.clj:137) ~[clojure-1.7.0.jar:?]
        at clojure.core$dorun.invoke(core.clj:3009) ~[clojure-1.7.0.jar:?]
        at clojure.core$doall.invoke(core.clj:3025) ~[clojure-1.7.0.jar:?]
        at org.apache.storm.daemon.worker$fn__11300$exec_fn__2470__auto__$reify__11302.run(worker.clj:663) ~[storm-core-1.2.2.jar:1.2.2]
        at java.security.AccessController.doPrivileged(Native Method) ~[?:1.8.0_152]
        at javax.security.auth.Subject.doAs(Subject.java:422) ~[?:1.8.0_152]
        at org.apache.storm.daemon.worker$fn__11300$exec_fn__2470__auto____11301.invoke(worker.clj:633) ~[storm-core-1.2.2.jar:1.2.2]
        at clojure.lang.AFn.applyToHelper(AFn.java:178) ~[clojure-1.7.0.jar:?]
        at clojure.lang.AFn.applyTo(AFn.java:144) ~[clojure-1.7.0.jar:?]
        at clojure.core$apply.invoke(core.clj:630) ~[clojure-1.7.0.jar:?]
        at org.apache.storm.daemon.worker$fn__11300$mk_worker__11391.doInvoke(worker.clj:605) [storm-core-1.2.2.jar:1.2.2]
        at clojure.lang.RestFn.invoke(RestFn.java:512) [clojure-1.7.0.jar:?]
        at org.apache.storm.daemon.worker$_main.invoke(worker.clj:798) [storm-core-1.2.2.jar:1.2.2]
        at clojure.lang.AFn.applyToHelper(AFn.java:165) [clojure-1.7.0.jar:?]
        at clojure.lang.AFn.applyTo(AFn.java:144) [clojure-1.7.0.jar:?]
        at org.apache.storm.daemon.worker.main(Unknown Source) [storm-core-1.2.2.jar:1.2.2]
Caused by: java.lang.ClassNotFoundException: com.fosung.share.stormdemo3.bolt.FilterBolt
        at java.net.URLClassLoader.findClass(URLClassLoader.java:381) ~[?:1.8.0_152]
        at java.lang.ClassLoader.loadClass(ClassLoader.java:424) ~[?:1.8.0_152]
        at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:338) ~[?:1.8.0_152]
        at java.lang.ClassLoader.loadClass(ClassLoader.java:357) ~[?:1.8.0_152]
        at java.lang.Class.forName0(Native Method) ~[?:1.8.0_152]
        at java.lang.Class.forName(Class.java:348) ~[?:1.8.0_152]
        at java.io.ObjectInputStream.resolveClass(ObjectInputStream.java:683) ~[?:1.8.0_152]
        at java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1863) ~[?:1.8.0_152]
        at java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1746) ~[?:1.8.0_152]
        at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2037) ~[?:1.8.0_152]
        at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1568) ~[?:1.8.0_152]
        at java.io.ObjectInputStream.readObject(ObjectInputStream.java:428) ~[?:1.8.0_152]
        at org.apache.storm.utils.Utils.javaDeserialize(Utils.java:253) ~[storm-core-1.2.2.jar:1.2.2]
        ... 38 more
2019-05-22 11:09:14.684 o.a.s.util main [ERROR] Halting process: ("Error on initialization")
java.lang.RuntimeException: ("Error on initialization")
        at org.apache.storm.util$exit_process_BANG_.doInvoke(util.clj:341) [storm-core-1.2.2.jar:1.2.2]
        at clojure.lang.RestFn.invoke(RestFn.java:423) [clojure-1.7.0.jar:?]
        at org.apache.storm.daemon.worker$fn__11300$mk_worker__11391.doInvoke(worker.clj:605) [storm-core-1.2.2.jar:1.2.2]
        at clojure.lang.RestFn.invoke(RestFn.java:512) [clojure-1.7.0.jar:?]
        at org.apache.storm.daemon.worker$_main.invoke(worker.clj:798) [storm-core-1.2.2.jar:1.2.2]
        at clojure.lang.AFn.applyToHelper(AFn.java:165) [clojure-1.7.0.jar:?]
        at clojure.lang.AFn.applyTo(AFn.java:144) [clojure-1.7.0.jar:?]
        at org.apache.storm.daemon.worker.main(Unknown Source) [storm-core-1.2.2.jar:1.2.2]

我的pom.xml

<properties>
        <java.version>1.8</java.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.apache.storm</groupId>
            <artifactId>storm-core</artifactId>
            <version>1.2.2</version>
            <!--<scope>provided</scope>-->
            <exclusions>
                <exclusion>
                    <groupId>org.apache.logging.log4j</groupId>
                    <artifactId>log4j-slf4j-impl</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.apache.logging.log4j</groupId>
                    <artifactId>log4j-1.2-api</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.apache.logging.log4j</groupId>
                    <artifactId>log4j-web</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>org.slf4j</groupId>
                    <artifactId>slf4j-log4j12</artifactId>
                </exclusion>
                <exclusion>
                    <artifactId>ring-cors</artifactId>
                    <groupId>ring-cors</groupId>
                </exclusion>
            </exclusions>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <source>8</source>
                    <target>8</target>
                </configuration>
            </plugin>
        </plugins>
    </build>

我的拓扑

public class MyTopology {

    public  static void main(String[] args) {
        System.out.println("MyTopology main start");
        // 定义一个拓扑

        TopologyBuilder builder = new TopologyBuilder();
        // 设置1个Executeor(线程)，默认一个
        DataSpout dataSpout = new DataSpout();
        builder.setSpout("spoutId", dataSpout);
        // shuffleGrouping:表示是随机分组
        // 设置1个Executeor(线程)，和两个task
        FilterBolt filterBolt = new FilterBolt();
        InsertBolt insertBolt = new InsertBolt();

        builder.setBolt("filterBolt", filterBolt).setNumTasks(1).allGrouping("spoutId", "spoutId");
        builder.setBolt("insertBolt", insertBolt).setNumTasks(1).allGrouping("filterBolt", "spoutId");
        Config conf = new Config();

        try {
            // 有参数时，表示向集群提交作业，并把第一个参数当做topology名称
            // 没有参数时，本地提交
            if (args != null && args.length > 0) {
                System.out.println("运行远程模式");
                StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
            } else {
                // 启动本地模式
                System.out.println("运行本地模式");
                LocalCluster cluster = new LocalCluster();
                cluster.submitTopology("TopologyApp", conf, builder.createTopology());
            }
        } catch (Exception e) {
            System.out.println("storm启动失败!程序退出!");
            System.exit(1);
            e.printStackTrace();
        }
//        System.out.println("storm启动成功...");
    }
}

我的嘴

public class DataSpout extends BaseRichSpout {

    SpoutOutputCollector collector;

    @Override
    public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
        this.collector = collector;
        System.out.println("spout open");
    }

    @Override
    public void nextTuple() {
        /*try {
            Thread.sleep(1000);
            return;
        } catch (InterruptedException e) {
            e.printStackTrace();
        }*/
        System.out.println("spout nextTuple start");
        int rndomn = (int)Math.random() * 1000;

        collector.emit("spoutId", new Values(rndomn));

        try {
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declareStream("spoutId", new Fields("spoutId"));
    }
}

我的门闩

public class FilterBolt extends BaseRichBolt  {
    OutputCollector collector;
    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
    }

    @Override
    public void execute(Tuple input) {
        System.out.println("filter bolt start");
        Integer o = (Integer) input.getValues().get(0);
        if (o>10){
            collector.emit("spoutId", new Values(o));
        }

    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {

        //定义下个bolt接收streamId
        declarer.declareStream("spoutId", new Fields("spoutId"));

    }
}
spring-bootapache-storm

1条答案

按热度按时间
byqmnocz

byqmnocz1#

Spring（ Boot ）不适合风暴。storm是一个框架，这意味着它负责管理某些类（如bolt）的生命周期。由于storm对spring一无所知，所以spring的依赖注入并不是现成的。可以将spring设置为在storm应用程序的某些部分上工作，例如使用任务和工作程序挂钩，这样就可以在storm工作程序中创建spring上下文。我不认为我会推荐它，除非你有充分的理由需要Spring。
关于您得到的错误，storm在您提交的jar中找不到您的类之一。因为您没有为spring配置发布pom.xml，所以很难说，但也许您使用的是一个可以移动类的插件。向storm提交拓扑时，storm会运行几个阶段，您应该了解：
首先是你 storm jar com.yourcompany.yourMain . 这将在本地机器（或运行命令的任何地方）上启动一个jvm，在您的情况下，它将运行拓扑设置 MyTopology.main . 然后，安装程序序列化了喷口和螺栓，并将jar和序列化的拓扑发送到nimbus（一个单独的jvm），后者又将其发送给主管（另一个单独的jvm）。在supervisor上，supervisor jvm启动许多工作jvm来运行拓扑。每个worker jvm都以如下命令开始 java -cp your-topology.jar org.apache.storm.Worker . 工作jvm加载序列化的拓扑和拓扑jar中的类，并启动线程来运行喷口和螺栓。
这些阶段很可能是你失败的原因。运行拓扑设置代码时，您使用的是spring boot命令，因此spring boot有机会运行。当拓扑在工作机上启动时，jvm是通过对非spring主方法的常规旧调用启动的，因此spring没有机会运行。
如果您决定不使用spring，您可以在这里找到一个pom的工作示例。
其他可能感兴趣的链接还有一个早期的答案和一个为storm进行spring集成的项目。

赞(0）分享  回复(0）举报  2021-06-24
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备2022004421号-2 NULL123 https://www.null123.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com