思享工具箱

storm实时计算实例（socket实时接入)

摘要：
本文介绍了一个简单的过程，即从实时日志文件中侦听，将其写入套接字服务器，然后访问Storm计算。
介绍

实现了一个简单的从实时日志文件监听，写入socket服务器，再接入Storm计算的一个流程。

源码

日志监听实时写入socket服务器

 
[java] view plain copy
 
  1. package socket;  
  2.   
  3. import java.io.BufferedReader;  
  4. import java.io.File;       
  5. import java.io.IOException;       
  6. import java.io.InputStreamReader;  
  7. import java.io.PrintWriter;  
  8. import java.io.RandomAccessFile;       
  9. import java.net.Socket;  
  10. import java.util.concurrent.Executors;       
  11. import java.util.concurrent.ScheduledExecutorService;       
  12. import java.util.concurrent.TimeUnit;       
  13. /* 
  14.  * 监测数据，通过socket远程发送到另外服务器 ，见MyServerMulti 
  15.  * ClientRead再通过服务器从socket里读 
  16.  *  
  17.  */  
  18.       
  19. public class LogViewToSocket {       
  20.     private long lastTimeFileSize = 0;  //上次文件大小       
  21.     /**    
  22.      * 实时输出日志信息    
  23.      * @param logFile 日志文件    
  24.      * @throws IOException    
  25.      */      
  26.       
  27.     public String getNewFile(File file)  
  28.     {  
  29.         File[] fs=file.listFiles();  
  30.         long maxtime=0;  
  31.         String newfilename="";  
  32.         for (int i=0;i<fs.length;i++)  
  33.         {  
  34.             if (fs[i].lastModified()>maxtime)  
  35.             {  
  36.                 maxtime=fs[i].lastModified();  
  37.                 newfilename=fs[i].getAbsolutePath();  
  38.                   
  39.             }  
  40.         }  
  41.         return newfilename;  
  42.     }  
  43.      RandomAccessFile randomFile=null;  
  44.      String newfile=null;  
  45.      String thisfile=null;  
  46.     public void realtimeShowLog(final File logFile,final PrintWriter out) throws IOException{       
  47.            newfile=getNewFile(logFile);  
  48.         //指定文件可读可写       
  49.             randomFile = new RandomAccessFile(new File(newfile),"r");       
  50.         //启动一个线程每1秒钟读取新增的日志信息       
  51.         ScheduledExecutorService exec =        
  52.             Executors.newScheduledThreadPool(1);       
  53.         exec.scheduleWithFixedDelay(new Runnable(){       
  54.             public void run() {       
  55.                 try {       
  56.                     //获得变化部分的       
  57.                     randomFile.seek(lastTimeFileSize);       
  58.                     String tmp = "";       
  59.                     while( (tmp = randomFile.readLine())!= null) {       
  60.                         System.out.println(new String(tmp.getBytes("ISO8859-1")));   
  61.                         out.println(new String(tmp.getBytes("ISO8859-1")));  
  62.                         out.flush();   
  63.                     }     
  64.                    thisfile=getNewFile(logFile);  
  65.                    if(!thisfile.equals(newfile))  
  66.                      
  67.                    {  
  68.                        randomFile = new RandomAccessFile(new File(newfile),"r");  
  69.                        lastTimeFileSize=0;  
  70.                    }  
  71.                    else  
  72.                          
  73.                     lastTimeFileSize = randomFile.length();       
  74.                      
  75.                 } catch (IOException e) {       
  76.                     throw new RuntimeException(e);       
  77.                 }       
  78.             }       
  79.         }, 0, 1, TimeUnit.SECONDS);       
  80.     }       
  81.            
  82.     public static void main(String[] args) throws Exception {       
  83.         LogViewToSocket view = new LogViewToSocket();       
  84.   
  85.             Socket socket=new Socket("192.168.27.100",5678);   
  86.      
  87.         PrintWriter out=new PrintWriter(socket.getOutputStream());      
  88.            
  89.             
  90.   
  91.         final File tmpLogFile = new File("/home/hadoop/test");       
  92.         view.realtimeShowLog(tmpLogFile,out);   
  93.        // socket.close();  
  94.           
  95.     }       
  96.       
  97. }      
 

socket服务器处理

[java] view plain copy
 
  1. import java.io.BufferedReader;    
  2. import java.io.IOException;    
  3. import java.io.InputStreamReader;    
  4. import java.io.PrintWriter;    
  5. import java.net.ServerSocket;    
  6. import java.net.Socket;    
  7. import java.net.SocketAddress;  
  8. import java.util.*;  
  9.     
  10. public class MyServerMulti {    
  11.     private static Socket socket1;  
  12.   
  13.     public static void main(String[] args) throws IOException {    
  14.         ServerSocket server = new ServerSocket(5678);    
  15.           int i=0;  
  16.           ArrayList<PrintWriter> outs=new ArrayList<PrintWriter>();  
  17.             
  18.           /* 
  19.            * 一个client socket发送数据过来， server端再发到其他client socket端 
  20.            *  
  21.            */  
  22.           Socket socket1=null;  
  23.         while (true) {  
  24.               
  25.             Socket socket = server.accept();    
  26.              i++;  
  27.              System.out.println(i);  
  28.              System.out.println(socket.getInetAddress());  
  29.                  PrintWriter out= new PrintWriter(socket.getOutputStream());  
  30.                  outs.add(out);  
  31.                  if(i==1)  
  32.                       socket1=socket;  
  33.                  if(i==2)  
  34.                        
  35.                  invoke(socket1,outs);  
  36.                    
  37.               
  38.         }    
  39.     }    
  40.         
  41.     private static void invoke(final Socket client, final ArrayList<PrintWriter> outs) throws IOException {    
  42.         new Thread(new Runnable() {    
  43.             public void run() {    
  44.                 BufferedReader in = null;    
  45.                 PrintWriter out = null;    
  46.                 PrintWriter out1 = null;  
  47.                 try {    
  48.                     in = new BufferedReader(new InputStreamReader(client.getInputStream()));    
  49.                     out = new PrintWriter(client.getOutputStream());    
  50.     
  51.                     while (true) {    
  52.                         String msg = in.readLine();    
  53.                         System.out.println(msg);    
  54.                         out.println("Server received " + msg);    
  55.                         out.flush();    
  56.                           
  57.                         /*数据转发送到多个client*/  
  58.                         for(int i=0;i<outs.size();i++)  
  59.                         {  
  60.                             out1=outs.get(i);  
  61.                             System.out.println(i);  
  62.                             System.out.println("send msg:"+msg);  
  63.                              out1.println(msg);  
  64.                             out1.flush();  
  65.                         }  
  66.                           
  67.                         System.out.println(client.getInetAddress());  
  68.                         if (msg.equals("bye")) {    
  69.                             break;    
  70.                         }    
  71.                     }    
  72.                 } catch(IOException ex) {    
  73.                     ex.printStackTrace();    
  74.                 } finally {    
  75.                     try {    
  76.                         in.close();    
  77.                     } catch (Exception e) {}    
  78.                     try {    
  79.                         out.close();    
  80.                     } catch (Exception e) {}    
  81.                     try {    
  82.                         client.close();    
  83.                     } catch (Exception e) {}    
  84.                 }    
  85.             }    
  86.         }).start();    
  87.     }    
  88. }    

storm topology

[java] view plain copy
 
  1. import java.io.BufferedReader;  
  2. import java.io.BufferedWriter;  
  3. import java.io.File;  
  4. import java.io.FileNotFoundException;  
  5. import java.io.FileOutputStream;  
  6. import java.io.FileReader;  
  7. import java.io.FileWriter;  
  8. import java.io.IOException;  
  9. import java.io.InputStreamReader;  
  10. import java.io.OutputStreamWriter;  
  11. import java.io.PrintWriter;  
  12. import java.io.RandomAccessFile;  
  13. import java.net.Socket;  
  14. import java.net.UnknownHostException;  
  15. import java.util.Map;  
  16.    
  17. //import mytest.ThroughputTest.GenSpout;  
  18.    
  19. import backtype.storm.Config;  
  20. import backtype.storm.LocalCluster;  
  21. import backtype.storm.StormSubmitter;  
  22. import backtype.storm.generated.AlreadyAliveException;  
  23. import backtype.storm.generated.InvalidTopologyException;  
  24. import backtype.storm.spout.SpoutOutputCollector;  
  25. import backtype.storm.task.OutputCollector;  
  26. import backtype.storm.task.TopologyContext;  
  27. import backtype.storm.topology.BasicOutputCollector;  
  28. import backtype.storm.topology.OutputFieldsDeclarer;  
  29. import backtype.storm.topology.TopologyBuilder;  
  30. import backtype.storm.topology.base.BaseBasicBolt;  
  31. import backtype.storm.topology.base.BaseRichBolt;  
  32. import backtype.storm.topology.base.BaseRichSpout;  
  33. import backtype.storm.tuple.Fields;  
  34. import backtype.storm.tuple.Tuple;  
  35. import backtype.storm.tuple.Values;  
  36. import backtype.storm.utils.Utils;  
  37. /* 
  38.  *  
  39.  * 
  40.  *  storm jar stormtest.jar socket.SocketProcess /home/hadoop/out_socket.txt true 
  41.  *  
  42.  */  
  43.    
  44. public class SocketProcess {  
  45.          public static class  SocketSpout extends BaseRichSpout {  
  46.    
  47.                    /** 
  48.                     */  
  49.               static Socket sock=null;  
  50.               static BufferedReader in=null;  
  51.               String str=null;  
  52.                    private static final long serialVersionUID = 1L;  
  53.                    private SpoutOutputCollector _collector;  
  54.                    private BufferedReader br;  
  55.                    private String dataFile;  
  56.                    private BufferedWriter bw2;  
  57.                     RandomAccessFile randomFile;  
  58.                     private long lastTimeFileSize = 0;   
  59.                     int cnt=0;  
  60.                    //定义spout文件  
  61.                     SocketSpout(){  
  62.                        
  63.                    }  
  64.    
  65.                    //定义如何读取spout文件  
  66.                    @Override  
  67.                    public void open(Map conf, TopologyContext context,  
  68.                                      SpoutOutputCollector collector) {  
  69.                             // TODO Auto-generated method stub  
  70.                             _collector = collector;  
  71.                             try {  
  72.                                 sock=new Socket("192.168.27.100",5678);  
  73.                                  in=     
  74.                                     new BufferedReader(new InputStreamReader(sock.getInputStream()));     
  75.                             } catch (UnknownHostException e) {  
  76.                                 // TODO Auto-generated catch block  
  77.                                 e.printStackTrace();  
  78.                             } catch (IOException e) {  
  79.                                 // TODO Auto-generated catch block  
  80.                                 e.printStackTrace();  
  81.                             }  
  82.                          
  83.                    }  
  84.    
  85.                    //获取下一个tuple的方法  
  86.                    @Override  
  87.                    public void nextTuple() {  
  88.                             // TODO Auto-generated method stub  
  89.                        if(sock==null){  
  90.                              try {  
  91.                                 sock=new Socket("192.168.27.100",5678);  
  92.                                  in=     
  93.                                         new BufferedReader(new InputStreamReader(sock.getInputStream()));    
  94.                             } catch (UnknownHostException e) {  
  95.                                 // TODO Auto-generated catch block  
  96.                                 e.printStackTrace();  
  97.                             } catch (IOException e) {  
  98.                                 // TODO Auto-generated catch block  
  99.                                 e.printStackTrace();  
  100.                             }   
  101.                        }  
  102.                          
  103.                          
  104.                        while(true){      
  105.                             
  106.                         try {  
  107.                             str = in.readLine();  
  108.                         } catch (IOException e) {  
  109.                             // TODO Auto-generated catch block  
  110.                             e.printStackTrace();  
  111.                         }  
  112.                         System.out.println(str);    
  113.                         _collector.emit(new Values(str));  
  114.                         if(str.equals("end")){      
  115.                             break;      
  116.                             }   
  117.                         }  
  118.                          
  119.                          
  120.                          
  121.                          
  122.                          
  123.                          
  124.                          
  125.                               
  126.                               
  127.                    }  
  128.    
  129.    
  130.                    @Override  
  131.                    public void declareOutputFields(OutputFieldsDeclarer declarer) {  
  132.                             // TODO Auto-generated method stub  
  133.                             declarer.declare(new Fields("line"));  
  134.                    }  
  135.                     
  136.          }  
  137.           
  138.    
  139.          public static class Process extends BaseRichBolt{  
  140.    
  141.                    private String _seperator;  
  142.                    private String _outFile;  
  143.                    PrintWriter pw;  
  144.                    private OutputCollector _collector;  
  145.                    private BufferedWriter bw;  
  146.                     
  147.                    public Process(String outFile) {  
  148.                              
  149.                             this._outFile   = outFile;  
  150.                              
  151.                    }  
  152.                     
  153.                    //把输出结果保存到外部文件里面。  
  154.                    @Override  
  155.                    public void prepare(Map stormConf, TopologyContext context,  
  156.                                      OutputCollector collector) {  
  157.                             // TODO Auto-generated method stub  
  158.                             this._collector = collector;  
  159.                             File out = new File(_outFile);  
  160.                             try {  
  161. //                                  br = new BufferedWriter(new FileWriter(out));  
  162.                                      bw = new BufferedWriter(new OutputStreamWriter(   
  163.                              new FileOutputStream(out, true)));   
  164.                             } catch (IOException e1) {  
  165.                                      // TODO Auto-generated catch block  
  166.                                      e1.printStackTrace();  
  167.                             }                  
  168.                    }  
  169.                     
  170.                    //blot计算单元，把tuple中的数据添加一个bkeep和回车。然后保存到outfile指定的文件中。  
  171.                    @Override  
  172.                    public void execute(Tuple input) {  
  173.                             // TODO Auto-generated method stub  
  174.                             String line = input.getString(0);  
  175. //                         System.out.println(line);  
  176.                        //     String[] str = line.split(_seperator);  
  177.                          //   System.out.println(str[2]);  
  178.                             try {  
  179.                                      bw.write(line+",bkeep"+" ");  
  180.                                      bw.flush();  
  181.                             } catch (IOException e) {  
  182.                                      // TODO Auto-generated catch block  
  183.                                      e.printStackTrace();  
  184.                             }  
  185.                              
  186.                             _collector.emit(new Values(line));  
  187.                    }  
  188.    
  189.                    @Override  
  190.                    public void declareOutputFields(OutputFieldsDeclarer declarer) {  
  191.                             // TODO Auto-generated method stub  
  192.                             declarer.declare(new Fields("line"));  
  193.                    }  
  194.                     
  195.          }  
  196.           
  197.          public static void main(String[] argv) throws AlreadyAliveException, InvalidTopologyException{  
  198.                   
  199.                    String outFile   = argv[0]; //输出文件  
  200.                    boolean distribute = Boolean.valueOf(argv[1]);       //本地模式还是集群模式  
  201.                    TopologyBuilder builder = new TopologyBuilder();  //build一个topology  
  202.         builder.setSpout("spout", new  SocketSpout(), 1);   //指定spout  
  203.         builder.setBolt("bolt", new Process(outFile),1).shuffleGrouping("spout");  //指定bolt，包括bolt、process和grouping  
  204.         Config conf = new Config();  
  205.         if(distribute){  
  206.             StormSubmitter.submitTopology("SocketProcess", conf, builder.createTopology());  
  207.         }else{  
  208.                  LocalCluster cluster = new LocalCluster();  
  209.                  cluster.submitTopology("SocketProcess", conf, builder.createTopology());  
  210.         }  
  211.          }         
  212. }  


最后执行
[java] view plain copy
 
  1. storm jar stormtest.jar socket.SocketProcess /home/hadoop/out_socket.txt true  
spout接受从socket服务器实时发送过来的数据，经过topology处理，最终将数据写入out_socket.txt文件
转：http://blog.csdn.net/u011750989/article/details/18547015

标签：#socket#try | 浏览：155 | 发布日期：

免责声明：文章转载自《storm实时计算实例（socket实时接入)》仅用于学习参考。如对内容有疑问，请及时联系本站处理。

上篇NET Framework 4.0无法安装!彻底理解Android Binder通信架构下篇

宿迁高防，2C2G15M，22元/月；香港BGP，2C5G5M，25元/月 雨云优惠码：MjYwNzM=

相关文章

使用wireshark抓包分析SOCKS5协议

目录 编写SOCKS5服务器运行代码（参考自Python编写socks5服务器） 使用SOCKS5服务器脚本和curl命令 分析抓取到的数据包理解SOCKS5协议的工作过程（感谢socks5代理服务器协议的说明让我预先知道SOCKS5协议数据消息传递的机理） 通信软件课选择了分析SOCKS5协议，想看一下这个协议在网络通信中是如何进行的，遂抓包...

自己封装的Socket组件，实现服务端多进程共享Socket对象，协同处理客户端请求

自己封装的Socket组件，实现服务端多进程共享Socket对象，协同处理客户端请求 DotNet.Net.MySocket是SLB.NET(Server Load Balance服务器负载均衡)项目中的核心组件。 在实际的项目中发现，单进程的服务端处理高并发的客户请求能力有限。 所以一直在考虑能不能象IIS Web园一样，多个进程提供一个服务，协同处理客...

asp.net 调用 excel 组件

Asp.net 如何调用 Excel ? 1.引用 Microsoft.Office.Interop.Excel.dll，自动包装成Interop.Microsoft.Office.Interop.Excel.dll 2.代码： ///<summary> ///生成 excel 报表 ///</summary> privatevoi...

.Net中的异常处理：高级异常

这种处理的有效性主要取决于所选择的语言和平台，因此，详细了解它们的正确用法和行为非常重要，这样我们的用户和其他开发人员在诊断代码中的问题时免受痛苦。在本文中，我们将了解C和.NET在错误处理方面的作用。 词汇表CLR：公共语言运行时的缩写，是.NET运行时，它负责执行用所有.NET语言编译的应用程序。除了虚拟机和实时编译器之外，它还具有额外的职责，如内存管...

[PDFBox]后台操作pdf的工具类

PDFBox是Apache下的一个操作pdf的类库。其也提供了一个命令行的工具，也提供了java调用的第三方类库。 下载地址:https://pdfbox.apache.org/ 下面的实验基于JDK8+pdfbox-2.0.13.jar+pdfbox-app-2.0.13.jar(命令行工具库) 1.命令行使用 文档参考:https://pdfbox.a...

python并发编程：阻塞IO

阻塞IO（blocking  IO） 在Linux中，默认情况下所有的socket都是blocking，一个典型的读操作流程大概是这样： 　　当用户进程调用了recvfrom这个系统调用，kernel就开始了IO的第一个阶段：准备数据。对于network io来说，很多时候数据在一开始还没有到达（比如，还没有收到一个完整的udp包），这个时候kernel...

最新文章

随机推荐

思享工具箱导航