java - 在实际集群环境中是否可以在Storm Bolts/Spout中使用全局静态变量

Question

我有一个Storm项目，拓扑类有提交拓扑的主要方法，Spout1 --> Bolt1，Spout2 --> Bolt2。我有另一个最终的 util 类，它有一个静态变量 --> 字符串列表。现在 Bolt1 使用这个静态列表并打印出内容，而 Bolt2 在列表中添加 String。 Spout1 和 Spout2 都分别以 1000 毫秒和 500 毫秒的间隔向各自的 Bolt 发送消息。

我将worker数量设置为4，即Config.setNumWorkers(4)。

并行度提示:Spout1 - 1、Bolt1 - 100、Spout2 - 1、Bolt2 - 100。

此代码设置在我的 Window 计算机中以 LocalCluster 模式运行。

但我不确定这是否适用于 Linux 中的实际集群环境，有多台服务器为 Supervisor 守护进程提供服务。在实际的集群环境中，我认为工作人员将在具有不同 JVM 进程的不同机器上运行。 那么 Bolt 是否可以访问全局静态变量，即字符串列表，因为它们可以在本地集群中执行此操作？

下面是一些代码供引用:

public static void main(String[] args) {

    System.out.println("Starting Topology....");
    TopologyBuilder builder = new TopologyBuilder();

    builder.setSpout("spout12", new TestSpout2(), 1);
    builder.setBolt("bolt12", new TestBolt2(), 100).shuffleGrouping("spout12", "spout12Stream");
    builder.setSpout("spout11", new TestSpout1(), 1);
    builder.setBolt("bolt11", new TestBolt1(), 100).shuffleGrouping("spout11", "spout11Stream");

    Config conf = new Config();
    conf.setDebug(false);
    conf.setNumWorkers(4);
    LocalCluster cluster = new LocalCluster();

    cluster.submitTopology("TestTopology3", conf, builder.createTopology());

}


Spout1 :
public void nextTuple() {
    int a = (int) (((Math.random() * 10)+1)*((Math.random() * 10)+1));
    String str = String.valueOf(a);
    Utils.sleep(1000);
    collector.emit("spout11Stream", new Values(str), str);
}


Bolt1 : 
public void execute(Tuple tuple) {
    System.out.println("########## In execute of TestBolt11....\t Value 0 : " + tuple.getString(0) + "\t\t List : "
            + CommonUtils.list);
    _collector.ack(tuple);
}


Spout2 : 
public void nextTuple() {
    int a = (int) (((Math.random() * 10)+1)*((Math.random() * 10)+1));
    String str = String.valueOf(a);
    Utils.sleep(500);
    collector.emit("spout12Stream", new Values(str), str);
}


Bolt2 : 
public void execute(Tuple tuple) {
    System.out.println("!!!!!!!!!!!!! In execute of TestBolt12....\t Value 0 : " + tuple.getString(0));
    CommonUtils.list.add(tuple.getString(0)+"gb");
    _collector.ack(tuple);
}


CommonUtilis class : 
public final class CommonUtils {
public static List<String> list = new ArrayList<String>();
}

系统输出成功运行:

 !!!!!!!!!!!!! In execute of TestBolt12....  Value 0 : 31
 !!!!!!!!!!!!! In execute of TestBolt12....  Value 0 : 9
 !!!!!!!!!!!!! In execute of TestBolt12....  Value 0 : 68
 !!!!!!!!!!!!! In execute of TestBolt12....  Value 0 : 24
 ########## In execute of TestBolt11....     Value 0 : 39        List : [31gb, 9gb, 68gb, 24gb]
 !!!!!!!!!!!!! In execute of TestBolt12....  Value 0 : 60
 !!!!!!!!!!!!! In execute of TestBolt12....  Value 0 : 30
 ########## In execute of TestBolt11....     Value 0 : 26        List : [31gb, 9gb, 68gb, 24gb, 60gb, 30gb]
 !!!!!!!!!!!!! In execute of TestBolt12....  Value 0 : 9
 !!!!!!!!!!!!! In execute of TestBolt12....  Value 0 : 15
 ########## In execute of TestBolt11....     Value 0 : 11        List : [31gb, 9gb, 68gb, 24gb, 60gb, 30gb, 9gb, 15gb]

Answer 1

简短的回答是否定的，静态变量访问的范围受 JVM 限制，正如您所指出的，您将拥有多个 JVM。

还有一些解决方法:

1. 您可以使用并行度提示为 1 的 Bolt 来访问静态变量。由于只有一个 Bolt，因此访问仅发生在一个 JVM 中，但话又说回来，您也可以只在该 Bolt 上使用实例变量。
2. (1) 的一种变体是在 Storm 中使用调度程序实现来拥有 Bolt 的多个实例，但全部分配给同一个工作线程。我没有尝试过这种方法，但我想无论如何我都会把它扔掉。
3. 您可以使用关系数据库或 NoSQL 数据库作为分布在集群中的数据的读/写存储
4. 您可以将列表存储在zookeeper中。我会将 Zookeeper 访问包装在一个单例类中，以便使您的 Bolt 实现更简单。