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

Question

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

考虑到这一点,我设计了一个简单的拓扑结构,设计用于一个喷嘴和一个螺栓.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));
     }
}

按@肖的建议,试图初始化_serverSocket的open()方法和_clientSocket在运行nextTuple()方式收听请求.

不知道这个的表现形式,但它有效.. :-)

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生命周期被调用一次,所以它效率不高.