我正在编写一个大数据应用程序，其中两个线程同时运行。线程a从网络接收数据，并将它们作为jsonobject放入blockingqueue。线程b，一个风暴喷口，然后从阻塞队列中读取并处理它们。我将blockingqueue对象传递给类构造函数中的spout类。我发现的问题是喷口中的阻塞队列是空的。你能告诉我怎么解决这个问题吗？
ctzwtxfj1#
通过运行一些类来启动storm应用程序，这些类将拓扑构建和配置为一组对象，然后将对象集合（连同jar文件）提交给nimbus服务器。其中一些对象是作为拓扑提交的一部分序列化的喷口和螺栓的示例。集群上的bolt和spout的每个示例都是这些反序列化对象之一。因此，所有螺栓和喷口都是在您第一次启动拓扑（通常在边缘节点上）时构造的，而不是在集群上。这对您意味着，在类初始化和对象构造期间由spout引用的任何对象都将与spout示例一起序列化。这将包括blockingqueue。你的blockingqueue正在被序列化并分发到集群中，听起来好像它无法在旅途中幸存下来。您要做的是在构造函数中保留阻塞队列变量null，而是在open（）方法中设置该变量。创建实际的队列对象时，可以将其存储在某个公共静态变量中，以便spout的open（）方法可以使用它。
1条答案
按热度按时间ctzwtxfj1#
通过运行一些类来启动storm应用程序，这些类将拓扑构建和配置为一组对象，然后将对象集合（连同jar文件）提交给nimbus服务器。其中一些对象是作为拓扑提交的一部分序列化的喷口和螺栓的示例。集群上的bolt和spout的每个示例都是这些反序列化对象之一。因此，所有螺栓和喷口都是在您第一次启动拓扑（通常在边缘节点上）时构造的，而不是在集群上。
这对您意味着，在类初始化和对象构造期间由spout引用的任何对象都将与spout示例一起序列化。这将包括blockingqueue。你的blockingqueue正在被序列化并分发到集群中，听起来好像它无法在旅途中幸存下来。
您要做的是在构造函数中保留阻塞队列变量null，而是在open（）方法中设置该变量。创建实际的队列对象时，可以将其存储在某个公共静态变量中，以便spout的open（）方法可以使用它。