我需要写一个风暴喷口来从端口读取数据.想知道这在逻辑上是否可行.
考虑到这一点,我设计了一个简单的拓扑结构,设计用于一个喷嘴和一个螺栓. 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：.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()方法中运行以侦听请求.
不知道这个的表现形式,但它有效.. 🙂
