Apache Storm拓扑中的依赖注入

如何解决Apache Storm拓扑中的依赖注入

背景很少：我正在使用Apache Storm开发拓扑，我想为什么不在其中使用依赖项注入，但是我不确定当拓扑部署到集群时它将如何在集群环境中运行。我开始寻找关于DI是否可以在Storm拓扑中使用的好选择的答案，我遇到了有关Apache Spark的一些线索，其中提到了序列化将是一个问题，并且看到了针对Apache Storm的一些回应。所以最后我决定用google guice编写一个示例拓扑，看看会发生什么。

我用两个螺栓编写了一个示例拓扑，并使用google guice注入了依赖项。第一个螺栓发出一个刻度元组，然后第一个螺栓创建消息，螺栓在日志上打印该消息并调用一些相同的类。然后，此消息将发送到第二个螺栓，并且在那里也有相同的打印逻辑。

第一螺栓

public class FirstBolt extends BaseRichBolt {
    private OutputCollector collector;
    private static int count = 0;
    private FirstInjectClass firstInjectClass;

    @Override
    public void prepare(Map map,TopologyContext topologyContext,OutputCollector outputCollector) {
        collector = outputCollector;
        Injector injector = Guice.createInjector(new Module());
        firstInjectClass = injector.getInstance(FirstInjectClass.class);
    }

    @Override
    public void execute(Tuple tuple) {
        count++;
        String message = "Message count "+count;
        firstInjectClass.printMessage(message);
        log.error(message);

        collector.emit("TO_SECOND_BOLT",new Values(message));

        collector.ack(tuple);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declareStream("TO_SECOND_BOLT",new Fields("MESSAGE"));
    }

    @Override
    public Map<String,Object> getComponentConfiguration() {
        Config conf = new Config();
        conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,10);
        return conf;
    }
}

第二螺栓

public class SecondBolt extends BaseRichBolt {
    private OutputCollector collector;
    private SecondInjectClass secondInjectClass;

    @Override
    public void prepare(Map map,OutputCollector outputCollector) {
        collector = outputCollector;
        Injector injector = Guice.createInjector(new Module());
        secondInjectClass = injector.getInstance(SecondInjectClass.class);
    }

    @Override
    public void execute(Tuple tuple) {
        String message = (String) tuple.getValue(0);
        secondInjectClass.printMessage(message);
        log.error("SecondBolt {}",message);

        collector.ack(tuple);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
    }
}

其中注入依赖项的类

public class FirstInjectClass {
    FirstInterface firstInterface;

    private final String prepend = "FirstInjectClass";

    @Inject
    public FirstInjectClass(FirstInterface firstInterface) {
        this.firstInterface = firstInterface;
    }

    public void printMessage(String message){
        log.error("{} {}",prepend,message);
        firstInterface.printMethod(message);
    }
}

用于绑定的接口

public interface FirstInterface {
    void printMethod(String message);
}

接口的实现

public class FirstInterfaceImpl implements FirstInterface{
    private final String prepend = "FirstInterfaceImpl";
    public void printMethod(String message){
        log.error("{} {}",message);
    }
}

以同样的方式，另一个类通过DI接收依赖项

public class SecondInjectClass {
    SecondInterface secondInterface;
    private final String prepend = "SecondInjectClass";

    @Inject
    public SecondInjectClass(SecondInterface secondInterface) {
        this.secondInterface = secondInterface;
    }

    public void printMessage(String message){
        log.error("{} {}",message);
        secondInterface.printMethod(message);
    }
}

另一个绑定界面

public interface SecondInterface {
    void printMethod(String message);
}

第二个界面的实现

public class SecondInterfaceImpl implements SecondInterface{
    private final String prepend = "SecondInterfaceImpl";
    public void printMethod(String message){
        log.error("{} {}",message);
    }
}

模块类

public class Module extends AbstractModule {
    @Override
    protected void configure() {
        bind(FirstInterface.class).to(FirstInterfaceImpl.class);
        bind(SecondInterface.class).to(SecondInterfaceImpl.class);
    }
}

这里没有什么幻想，只有两个螺栓和几个用于DI的类。我将其部署在服务器上，并且工作正常。但是，问题是我必须在每个螺栓中初始化Injector，这使我怀疑它的副作用是什么？

此实现很简单，只需2个螺栓。如果我有更多螺栓，该怎么办？如果必须在所有螺栓中初始化喷油器，它将对拓扑产生什么影响？

如果我尝试在prepare方法之外初始化Injector，则会出现序列化错误。