Storm：用于从端口读取数据的Spout

我需要写一个风暴喷口来从端口读取数据。 想知道这在逻辑上是否可行。

考虑到这一点，我设计了一个简单的拓扑结构，设计用于一个喷嘴和一个螺栓。 spout会收集使用wget发送的HTTP请求，而bolt会显示请求 – 就是这样。

我的喷口结构如下：

public class ProxySpout extends BaseRichSpout{ //The O/P collector SpoutOutputCollector sc; //The socket Socket clientSocket; //The server socket ServerSocket sc; public ProxySpout(int port){ this.sc=new ServerSocket(port); try{ clientSocket=sc.accept(); }catch(IOException ex){ //Handle it } } public void nextTuple(){ try{ InputStream ic=clientSocket.getInputStream(); byte b=new byte[8196]; int len=ic.read(b); sc.emit(new Values(b)); ic.close(); }catch(//){ //Handle it }finally{ clientSocket.close(); } } } 

我也实现了其余的方法。

当我将其转换为拓扑并运行它时，我发送第一个请求时出错：

了java.lang.RuntimeException：java.io.NotSerializableException：java.net.Socket中

只需要知道我实施这个喷口的方式是否有问题。 鲸鱼喷水器甚至可以从端口收集数据吗？ 或者鲸鱼喷水充当代理的实例？

编辑

搞定了。

代码是：

  public class ProxySpout extends BaseRichSpout{ //The O/P collector static SpoutOutputCollector _collector; //The socket static Socket _clientSocket; static ServerSocket _serverSocket; static int _port; public ProxySpout(int port){ _port=port; } public void open(Map conf,TopologyContext context, SpoutOutputCollector collector){ _collector=collector; _serverSocket=new ServerSocket(_port); } public void nextTuple(){ _clientSocket=_serverSocket.accept(); InputStream incomingIS=_clientSocket.getInputStream(); byte[] b=new byte[8196]; int len=b.incomingIS.read(b); _collector.emit(new Values(b)); } } 

根据@ Shaw的建议，尝试在open()方法中初始化_serverSocket ， _clientSocket在nextTuple()方法中运行以侦听请求。

不知道这个的表现forms，但它有效.. 🙂

Answer 1

在构造函数中只需分配变量。 尝试在prepare方法中实例化ServerSocket，不要在构造函数中编写任何新的… 并重命名变量，你有两个sc变量。

 public class ProxySpout extends BaseRichSpout{ int port; public ProxySpout(int port){ this.port=port; } @Override public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) { //new ServerSocket } @Override public void nextTuple() { } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { } } 

如果你把它放在prepare方法中，那么只有在已经部署了spout之后才会调用它，所以它不需要被序列化，并且它只会在每个spout生命周期被调用一次，所以它不是低效的。