我们为这个问题奋斗了很长时间。简言之，我们的风暴拓扑在一段时间后停止以随机方式从喷口发出消息。我们有一个自动化的脚本，在主数据刷新活动完成后，它会在每天6:00 utc重新部署拓扑。
在过去的两周内，我们的拓扑在utc晚些时候（22:00到02:00之间）停止发送消息3次。只有当我们重新启动它时，它才会联机，大约是协调世界时06:00。
我已经搜索了很多答案和博客，但找不到这里发生了什么。我们有一个未锚定的拓扑结构，这是我们3-4年前做出的选择。我们从0.9.2开始，现在是1.1.0。
我查过所有的日志，我百分之百肯定 nextTuple() 没有调用控制器的方法，并且系统中没有可能导致此问题的异常发生。我也检查了所有我们积累的日志，甚至没有一个错误或警告日志解释突然停止。信息日志也没那么有用。在工作者日志、主管日志或nimbus日志中，没有与此问题相关的内容。
这就是我们的spout类的外观：controller.java

public class Controller implements IRichSpout {

    SpoutOutputCollector _collector;
    Calendar LAST_RUN = null;
    List<ControllerMessage> msgList;

    /**
     * It is to open the spout
     */
      public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
        _collector = collector;
        msgList= new ArrayList<ControllerMessage>();

        MongoIndexingHandler mongoIndexingHandler = new MongoIndexingHandler();
        mongoIndexingHandler.createMongoIndexes();

      }

      /**
       * It executes the next tuple
       */

    @Override
    public void nextTuple() {
           Map<String, Object> logMap = new HashMap<>();
            logMap.put("BEGIN", new Date());

        try {
            TriggerHandler thandler = new TriggerHandler();
            if (msgList.size() == 0) {
                List<ControllerMessage> mList = thandler.getControllerMessage(new Date());
                msgList = mList;
            }

            if (msgList.size() > 0) {
                ControllerMessage message = msgList.get(0);
                if(thandler.fire(message.getFireTime())) {
                    Util.log(message, "CONTROLLER_LOGS", message.getTime(), new Date());
                    msgList.remove(0);
                    _collector.emit(new Values(message));
                }

            }
            else{
                Utils.sleep(1000);
            }

        } catch (Exception e) {
            _collector.reportError(e);

            Util.exLog(e, "EXECUTOR_ERROR", new Date(), "nextTuple()",Controller.class);
        } 
    }

      /**
       * It acknowledges the messages
       */
      @Override
      public void ack(Object id) {

      }
      /**
       * It tells failed messages
       */
      @Override
      public void fail(Object id) {

      }
     /**
      * It declares the message name
      */
      @Override
      public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("SPOUT_MESSAGE"));
      }

    @Override
    public void activate() {

    }

    @Override
    public void close() {

    }

    @Override
    public void deactivate() {

    }

    @Override
    public Map<String, Object> getComponentConfiguration() {
        return null;
    }

}

这是拓扑类：diagnostictopology.java

public class DiagnosticTopology {

    public static void main(String[] args) throws Exception {
        int gSize = (null != args && args.length > 0) ? Integer.parseInt(args[0]) : 2;
        int sSize = (null != args && args.length > 1) ? Integer.parseInt(args[1]) : 128;
        int sMSize = (null != args && args.length > 2) ? Integer.parseInt(args[2]) : 16;
        int aGSize = (null != args && args.length > 3) ? Integer.parseInt(args[3]) : 16;
        int rSize = (null != args && args.length > 4) ? Integer.parseInt(args[4]) : 64;
        int rMSize = (null != args && args.length > 5) ? Integer.parseInt(args[5]) : 16;
        int dMSize = (null != args && args.length > 6) ? Integer.parseInt(args[6]) : 8;
        int wSize = (null != args && args.length > 7) ? Integer.parseInt(args[7]) : 16;
        String topologyName = (null != args && args.length > 8) ? args[8] : "DIAGNOSTIC";

        TopologyBuilder builder = new TopologyBuilder();

        builder.setSpout("controller", new Controller(), 1);
        builder.setBolt("generator", new GeneratorBolt(), gSize).shuffleGrouping("controller");
        builder.setBolt("scraping", new ScrapingBolt(), sSize).shuffleGrouping("generator");
        builder.setBolt("smongo", new MongoBolt(), sMSize).shuffleGrouping("scraping");
        builder.setBolt("aggregation", new AggregationBolt(), aGSize).shuffleGrouping("scraping");
        builder.setBolt("rule", new RuleBolt(), rSize).shuffleGrouping("smongo");
        builder.setBolt("rmongo", new RMongoBolt(), rMSize).shuffleGrouping("rule");
        builder.setBolt("dstatus", new DeviceStatusBolt(), dMSize).shuffleGrouping("rule");

        builder.setSpout("trigger", new TriggerSpout(), 1);
        builder.setBolt("job", new JobTriggerBolt(), 4).shuffleGrouping("trigger");

        Config conf = new Config();
        conf.setDebug(false);
        conf.setNumWorkers(wSize);

        StormSubmitter.submitTopologyWithProgressBar(topologyName, conf, builder.createTopology());
    }
}

我们为生产和测试环境准备了相当好的服务器（xeon、8核、32 gb和闪存驱动器），并且没有外部因素会导致此问题，因为异常处理在代码中无处不在。
当这件事发生的时候，似乎一切都突然停止了，也没有发生原因的痕迹。
非常感谢您的帮助！