Spark Streaming 自定义接收器

1. public class JavaCustomReceiver extends Receiver<String> {  
2.   
3.   String host = null;  
4.   int port = -1;  
5.   
6.   public JavaCustomReceiver(String host_ , int port_) {  
7.     super(StorageLevel.MEMORY_AND_DISK_2());  
8.     host = host_;  
9.     port = port_;  
10.   }  
11.   
12.   public void onStart() {  
13.     // Start the thread that receives data over a connection  
14.     new Thread()  {  
15.       @Override public void run() {  
16.         receive();  
17.       }  
18.     }.start();  
19.   }  
20.   
21.   public void onStop() {  
22.     // There is nothing much to do as the thread calling receive()  
23.     // is designed to stop by itself if isStopped() returns false  
24.   }  
25.   
26.   /** Create a socket connection and receive data until receiver is stopped */  
27.   private void receive() {  
28.     Socket socket = null;  
29.     String userInput = null;  
30.   
31.     try {  
32.       // connect to the server  
33.       socket = new Socket(host, port);  
34.   
35.       BufferedReader reader = new BufferedReader(new InputStreamReader(socket.getInputStream()));  
36.   
37.       // Until stopped or connection broken continue reading  
38.       while (!isStopped() && (userInput = reader.readLine()) != null) {  
39.         System.out.println("Received data '" + userInput + "'");  
40.         store(userInput);  
41.       }  
42.       reader.close();  
43.       socket.close();  
44.   
45.       // Restart in an attempt to connect again when server is active again  
46.       restart("Trying to connect again");  
47.     } catch(ConnectException ce) {  
48.       // restart if could not connect to server  
49.       restart("Could not connect", ce);  
50.     } catch(Throwable t) {  
51.       // restart if there is any other error  
52.       restart("Error receiving data", t);  
53.     }  
54.   }  
55. }  

• 系统考虑到将数据转化为适当大小的块；
• 系统考虑到控制接收速率，如果速率限制已指定；
• 不可靠接收器比可靠接收器更加容易实现；

http://blog.csdn.net/ouyang111222/article/details/50414621