时间戳

首页 > 解决方案 > Apache Storm 2.1.0 本地 DRPC 不返回任何响应，尽管最后一个螺栓很好地向收集器发出了一个元组

java - Apache Storm 2.1.0 本地 DRPC 不返回任何响应，尽管最后一个螺栓很好地向收集器发出了一个元组

问题描述

我在尝试运行包含一个螺栓的 DRPC 拓扑并通过本地集群对其进行查询时遇到问题。使用 IntelliJ 调试后，确实执行了螺栓，但 JCQueue 在螺栓执行后陷入了无限循环，直到向服务器发送超时。

以下是用于构建拓扑构建器的代码：

public static LinearDRPCTopologyBuilder createBuilder()
{
    var bolt = new MRedisLookupBolt(createRedisConfiguration(), new RedisTurnoverMapper());
    var builder = new LinearDRPCTopologyBuilder("sales");
    builder.addBolt(bolt, 1).localOrShuffleGrouping();
    return builder;
}

MRedisLookupBolt 只是 IBasicBolt 对 Jedis 执行 hget 命令的一个非常简单的实现。MRedisLookupBolt的execute方法只是发出一个Values包含两个字段值的实例，这些字段的声明如下：

declarer.declare(new Fields("id", "Value"));

拓扑是在单元测试中构建和查询的，如下所示：

Config conf = new Config();
conf.setDebug(true);
conf.setNumWorkers(1);

try(LocalDRPC drpc = new LocalDRPC())
{
       LocalCluster cluster = new LocalCluster();
       var builder = BasicRedisRPCTopology.createBuilder();
       LocalCluster.LocalTopology topo = cluster.submitTopology(
              "Sales-fetch", conf, builder.createLocalTopology(drpc));
       var result = drpc.execute("sales", "XXXXX");
       System.out.println("################ Result: " + result);
}
catch (Exception e)
{
       e.printStackTrace();
}

阅读日志时，我确信数据是红色的，并且所有内容都已发出 在此处输入图像描述

但最后，我用我的测试方法轻轻打印了这个堆栈跟踪。当然，没有为结果变量分配任何值，并且进程永远不会到达最后的打印指令：

在此处输入图像描述

我在这里缺少一些东西。我的理解是：BoltExecutor 用于检索要执行的螺栓的 id 的 JCQueue 永无止境，尽管只有一个参数发送到本地 DRPC 并且只有一个螺栓声明到拓扑中。我已经尝试向拓扑添加更多螺栓或更改用于创建它的构建器实现，但没有成功。

标签： javaconfigurationapache-stormdrpc

解决方案

我使用 Apache Storm 2.1.0 找到了适合我用例的解决方案。

似乎submitTopology按照文档中的建议调用本地集群的方法并没有使用 2.1.0 版本正确地结束执行LinearDRPCTopologyBuilder程序来构建拓扑。

通过查看源代码，可以理解如何将LinearDRPCTopologyBuilder逻辑TopologyBuilder直接应用于。

这是应用于该createBuilder方法的更改：

    public static TopologyBuilder createBuilder(ILocalDRPC localDRPC)
    {
        var spout = Optional.ofNullable(localDRPC)
                .map(drpc -> new DRPCSpout("sales", drpc))
                .orElse(new DRPCSpout("sales"));
        var bolt = new MRedisLookupBolt(createRedisConfiguration(), new RedisTurnoverMapper());
        var builder = new TopologyBuilder();
        builder.setSpout("drpc", spout);
        builder.setBolt("redisLookup", bolt, 1)
               .shuffleGrouping("drpc");
        builder.setBolt("return", new ReturnResults())
               .shuffleGrouping("redisLookup");
        return builder;
    }

这是一个执行示例：

        Config conf = new Config();
        conf.setDebug(true);
        conf.setNumWorkers(1);

        try(LocalDRPC drpc = new LocalDRPC())
        {
            LocalCluster cluster = new LocalCluster();
            var builder = BasicRedisRPCTopology.createBuilder(drpc);
            cluster.submitTopology("Sales-fetch", conf, builder.createTopology());
            var result = drpc.execute("sales", "XXXXX");
            System.out.println("################ Result: " + result);
        }
        catch (Exception e)
        {
            e.printStackTrace();
        }

不幸的是，该解决方案不允许使用所有嵌入式工具，LinearDRPCTopologyBuilder并且意味着“手动”构建所有拓扑流。有必要将映射器行为更改为，因为字段未以与以前相同的顺序公开。

推荐阅读