网页主动探测工具-增加Netty模式

接前文
http://blog.itpub.net/29254281/viewspace-1344706/
http://blog.itpub.net/29254281/viewspace-1347985/
http://blog.itpub.net/29254281/viewspace-2134876/
http://blog.itpub.net/29254281/viewspace-2135131/

还是那个程序,在之前的基础上,改用Netty作为客户端.
也不知道用的到底对不对,先记录一下,以后慢慢学习.

1. import java.io.IOException;  
2. import java.nio.channels.Selector;  
3. import java.sql.Connection;  
4. import java.sql.DriverManager;  
5. import java.sql.PreparedStatement;  
6. import java.sql.SQLException;  
7. import java.sql.Timestamp;  
8. import java.util.ArrayList;  
9. import java.util.HashSet;  
10. import java.util.Iterator;  
11. import java.util.List;  
12. import java.util.Set;  
13. import java.util.concurrent.BlockingQueue;  
14. import java.util.concurrent.LinkedBlockingQueue;  
15. import java.util.concurrent.atomic.AtomicInteger;  
16. import java.util.regex.Matcher;  
17. import java.util.regex.Pattern;  
18.   
19. import io.netty.bootstrap.Bootstrap;  
20. import io.netty.buffer.ByteBuf;  
21. import io.netty.buffer.Unpooled;  
22. import io.netty.channel.Channel;  
23. import io.netty.channel.ChannelHandlerContext;  
24. import io.netty.channel.ChannelInboundHandlerAdapter;  
25. import io.netty.channel.ChannelInitializer;  
26. import io.netty.channel.EventLoopGroup;  
27. import io.netty.channel.nio.NioEventLoopGroup;  
28. import io.netty.channel.socket.nio.NioSocketChannel;  
29. import io.netty.handler.codec.LineBasedFrameDecoder;  
30. import io.netty.handler.codec.string.StringDecoder;  
31.   
32. class Reactor implements Runnable {  
33.     public static int GETCOUNT() {  
34.         return COUNT.get();  
35.   
36.     }  
37.   
38.     public static int getQueueSize() {  
39.         return QUEUE.size();  
40.     }  
41.   
42.     private static final AtomicInteger COUNT = new AtomicInteger();  
43.     private static final AtomicInteger TASKCOUNT = new AtomicInteger();  
44.   
45.     public int startTask() {  
46.         return TASKCOUNT.incrementAndGet();  
47.     }  
48.   
49.     public int finishTask() {  
50.         return TASKCOUNT.decrementAndGet();  
51.     }  
52.   
53.     public int incrementAndGet() {  
54.         return COUNT.incrementAndGet();  
55.     }  
56.   
57.     public final Selector selector;  
58.     private static BlockingQueue QUEUE = new LinkedBlockingQueue();  
59.   
60.     public void addTask(Task task) {  
61.         try {  
62.             QUEUE.put(task);  
63.         } catch (InterruptedException e) {  
64.             e.printStackTrace();  
65.         }  
66.     }  
67.   
68.     public Reactor() throws IOException {  
69.         selector = Selector.open();  
70.     }  
71.   
72.     @Override  
73.     public void run() {  
74.   
75.         EventLoopGroup group = new NioEventLoopGroup(3);  
76.   
77.         final Reactor reactor = this;  
78.         while (!Thread.interrupted()) {  
79.             int maxClient = 500;  
80.             Task task = null;  
81.             if (TASKCOUNT.get() < maxClient) {  
82.                 try {  
83.                     while ((task = (Task) QUEUE.take()) != null) {  
84.   
85.                         final Task t = task;  
86.                         reactor.startTask();  
87.                         Bootstrap boot = new Bootstrap();  
88.                         boot.group(group).channel(NioSocketChannel.class).handler(new ChannelInitializer() {  
89.   
90.                             @Override  
91.                             protected void initChannel(Channel ch) throws Exception {  
92.                                 ch.pipeline().addLast(new LineBasedFrameDecoder(409600));  
93.                                 ch.pipeline().addLast(new StringDecoder());  
94.                                 ch.pipeline().addLast(new HttpClientInboundHandler(reactor, t));  
95.                             }  
96.                         });  
97.   
98.                         boot.connect(task.getHost(), task.getPort());  
99.   
100.                         if (TASKCOUNT.get() > maxClient) {  
101.                             break;  
102.                         }  
103.                     }  
104.   
105.                 } catch (InterruptedException e) {  
106.                     e.printStackTrace();  
107.                 }  
108.             } else {  
109.                 //如果已经连接了500个网页,则主线程休眠一段时间.  
110.                 try {  
111.                     Thread.sleep(10);  
112.                 } catch (InterruptedException e) {  
113.                     e.printStackTrace();  
114.                 }  
115.             }  
116.   
117.         }  
118.         group.shutdownGracefully();  
119.   
120.     }  
121. }  
122.   
123. class HttpClientInboundHandler extends ChannelInboundHandlerAdapter {  
124.   
125.     private Task task;  
126.     private Reactor reactor;  
127.   
128.     @Override  
129.     public void channelInactive(ChannelHandlerContext ctx) throws Exception {  
130.         ctx.channel().closeFuture();  
131.         ctx.close();  
132.         this.reactor.finishTask();  
133.         task.setEndtime(System.currentTimeMillis());  
134.         this.reactor.incrementAndGet();  
135.         new ParseHandler(reactor, task).run();  
136.     }  
137.   
138.     public HttpClientInboundHandler(Reactor reactor, Task task) {  
139.         this.task = task;  
140.         this.reactor = reactor;  
141.     }  
142.   
143.     @Override  
144.     public void channelActive(ChannelHandlerContext ctx) throws Exception {  
145.   
146.         task.setStarttime(System.currentTimeMillis());  
147.         StringBuilder sb = new StringBuilder();  
148.         sb.append("GET " + task.getCurrentPath() + " HTTP/1.0\r\n");  
149.         sb.append("HOST:" + task.getHost() + "\r\n");  
150.         sb.append("Accept:*/*\r\n");  
151.         sb.append("\r\n");  
152.         ByteBuf bb = Unpooled.copiedBuffer(sb.toString().getBytes("utf8"));  
153.         ctx.writeAndFlush(bb);  
154.   
155.     }  
156.   
157.     @Override  
158.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {  
159.         String content = (String) msg;  
160.         task.getContent().append(content);  
161.         task.getContent().append("\n");  
162.   
163.     }  
164.   
165.     @Override  
166.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {  
167.         cause.printStackTrace();  
168.         ctx.close();  
169.     }  
170.   
171. }  
172.   
173. public class Probe {  
174.     public static void main(String[] args) throws IOException, InterruptedException {  
175.         for (int i = 0; i < 1; i++) {  
176.             Reactor reactor = new Reactor();  
177.             reactor.addTask(new Task("news.163.com", 80, "/index.html"));  
178.             new Thread(reactor, "ReactorThread_" + i).start();  
179.         }  
180.         long start = System.currentTimeMillis();  
181.         while (true) {  
182.             Thread.sleep(1000);  
183.             long end = System.currentTimeMillis();  
184.             float interval = ((end - start) / 1000);  
185.             int connectTotal = Reactor.GETCOUNT();  
186.   
187.             int persistenceTotal = PersistenceHandler.GETCOUNT();  
188.   
189.             int connectps = Math.round(connectTotal / interval);  
190.             int persistenceps = Math.round(persistenceTotal / interval);  
191.             System.out.print("\r连接总数:" + connectTotal + " \t每秒连接:" + connectps + "\t连接队列剩余:" + Reactor.getQueueSize()  
192.                     + " \t持久化总数:" + persistenceTotal + " \t每秒持久化:" + persistenceps + "\t持久化队列剩余:"  
193.                     + PersistenceHandler.getInstance().getSize());  
194.         }  
195.     }  
196.   
197. }  
198.   
199. class Task {  
200.     private String host;  
201.     private int port;  
202.     private String currentPath;  
203.     private long starttime;  
204.     private long endtime;  
205.   
206.     private String type;  
207.     private StringBuilder content = new StringBuilder(2400);  
208.     private int state;  
209.     private boolean isValid = true;  
210.   
211.     public Task() {  
212.     }  
213.   
214.     public Task(String host, int port, String path) {  
215.         init(host, port, path);  
216.     }  
217.   
218.     public void init(String host, int port, String path) {  
219.         this.setCurrentPath(path);  
220.         this.host = host;  
221.         this.port = port;  
222.     }  
223.   
224.     public long getStarttime() {  
225.         return starttime;  
226.     }  
227.   
228.     public void setStarttime(long starttime) {  
229.         this.starttime = starttime;  
230.     }  
231.   
232.     public long getEndtime() {  
233.         return endtime;  
234.     }  
235.   
236.     public void setEndtime(long endtime) {  
237.         this.endtime = endtime;  
238.     }  
239.   
240.     public boolean isValid() {  
241.         return isValid;  
242.     }  
243.   
244.     public void setValid(boolean isValid) {  
245.         this.isValid = isValid;  
246.     }  
247.   
248.     public int getState() {  
249.         return state;  
250.     }  
251.   
252.     public void setState(int state) {  
253.         this.state = state;  
254.     }  
255.   
256.     public String getCurrentPath() {  
257.         return currentPath;  
258.     }  
259.   
260.     public void setCurrentPath(String currentPath) {  
261.         this.currentPath = currentPath;  
262.         int i = 0;  
263.         if (currentPath.indexOf("?") != -1) {  
264.             i = currentPath.indexOf("?");  
265.         } else {  
266.             if (currentPath.indexOf("#") != -1) {  
267.                 i = currentPath.indexOf("#");  
268.             } else {  
269.                 i = currentPath.length();  
270.             }  
271.         }  
272.         this.type = currentPath.substring(currentPath.indexOf(".") + 1, i);  
273.     }  
274.   
275.     public long getTaskTime() {  
276.         return getEndtime() - getStarttime();  
277.     }  
278.   
279.     public String getType() {  
280.         return type;  
281.     }  
282.   
283.     public void setType(String type) {  
284.         this.type = type;  
285.     }  
286.   
287.     public String getHost() {  
288.         return host;  
289.     }  
290.   
291.     public int getPort() {  
292.         return port;  
293.     }  
294.   
295.     public StringBuilder getContent() {  
296.         return content;  
297.     }  
298.   
299.     public void setContent(StringBuilder content) {  
300.         this.content = content;  
301.     }  
302.   
303. }  
304.   
305. class ParseHandler implements Runnable {  
306.     private static final Set SET = new HashSet();  
307.   
308.     PersistenceHandler persistencehandler = PersistenceHandler.getInstance();  
309.   
310.     List domainlist = new ArrayList();  
311.   
312.     Task task;  
313.   
314.     private interface Filter {  
315.         void doFilter(Task fatherTask, Task newTask, String path, Filter chain);  
316.     }  
317.   
318.     private class FilterChain implements Filter {  
319.         private List list = new ArrayList();  
320.   
321.         {  
322.             addFilter(new TwoLevel());  
323.             addFilter(new OneLevel());  
324.             addFilter(new FullPath());  
325.             addFilter(new Root());  
326.             addFilter(new Default());  
327.         }  
328.   
329.         private void addFilter(Filter filter) {  
330.             list.add(filter);  
331.         }  
332.   
333.         private Iterator it = list.iterator();  
334.   
335.         @Override  
336.         public void doFilter(Task fatherTask, Task newTask, String path, Filter chain) {  
337.             if (it.hasNext()) {  
338.                 ((Filter) it.next()).doFilter(fatherTask, newTask, path, chain);  
339.             }  
340.         }  
341.   
342.     }  
343.   
344.     private class TwoLevel implements Filter {  
345.   
346.         @Override  
347.         public void doFilter(Task fatherTask, Task newTask, String path, Filter chain) {  
348.             if (path.startsWith("../../")) {  
349.                 String prefix = getPrefix(fatherTask.getCurrentPath(), 3);  
350.                 newTask.init(fatherTask.getHost(), fatherTask.getPort(), path.replace("../../", prefix));  
351.             } else {  
352.                 chain.doFilter(fatherTask, newTask, path, chain);  
353.             }  
354.   
355.         }  
356.     }  
357.   
358.     private class OneLevel implements Filter {  
359.   
360.         @Override  
361.         public void doFilter(Task fatherTask, Task newTask, String path, Filter chain) {  
362.             if (path.startsWith("../")) {  
363.                 String prefix = getPrefix(fatherTask.getCurrentPath(), 2);  
364.                 newTask.init(fatherTask.getHost(), fatherTask.getPort(), path.replace("../", prefix));  
365.             } else {  
366.                 chain.doFilter(fatherTask, newTask, path, chain);  
367.             }  
368.   
369.         }  
370.   
371.     }  
372.   
373.     private class FullPath implements Filter {  
374.   
375.         @Override  
376.         public void doFilter(Task fatherTask, Task newTask, String path, Filter chain) {  
377.             if (path.startsWith("http://")) {  
378.                 Iterator it = domainlist.iterator();  
379.                 boolean flag = false;  
380.                 while (it.hasNext()) {  
381.                     String domain = (String) it.next();  
382.                     if (path.startsWith("http://" + domain + "/")) {  
383.                         newTask.init(domain, fatherTask.getPort(), path.replace("http://" + domain + "/", "/"));  
384.                         flag = true;  
385.                         break;  
386.                     }  
387.                 }  
388.                 if (!flag) {  
389.                     newTask.setValid(false);  
390.                 }  
391.             } else {  
392.                 chain.doFilter(fatherTask, newTask, path, chain);  
393.             }  
394.         }  
395.   
396.     }  
397.   
398.     private class Root implements Filter {  
399.   
400.         @Override  
401.         public void doFilter(Task fatherTask, Task newTask, String path, Filter chain) {  
402.             if (path.startsWith("/")) {  
403.                 newTask.init(fatherTask.getHost(), fatherTask.getPort(), path);  
404.             } else {  
405.                 chain.doFilter(fatherTask, newTask, path, chain);  
406.             }  
407.         }  
408.   
409.     }  
410.   
411.     private class Default implements Filter {  
412.   
413.         @Override  
414.         public void doFilter(Task fatherTask, Task newTask, String path, Filter chain) {  
415.             if (path.contains(":")) {  
416.                 newTask.setValid(false);  
417.                 return;  
418.             }  
419.             String prefix = getPrefix(fatherTask.getCurrentPath(), 1);  
420.             newTask.init(fatherTask.getHost(), fatherTask.getPort(), prefix + "/" + path);  
421.         }  
422.     }  
423.   
424.     public ParseHandler(Reactor reactor, Task task) {  
425.         this.reactor = reactor;  
426.         this.task = task;  
427.   
428.         // 增加白名单  
429.         this.domainlist.add("news.163.com");  
430.     }  
431.   
432.     private Reactor reactor;  
433.     private Pattern pattern = Pattern.compile("\"[^\"]+\\.htm[^\"]*\"");  
434.   
435.     private void parseTaskState(Task task) {  
436.         if (task.getContent().toString().startsWith("HTTP/1.1")) {  
437.             task.setState(Integer.parseInt(task.getContent().substring(9, 12)));  
438.         } else {  
439.             try {  
440.                 task.setState(Integer.parseInt(task.getContent().substring(9, 12)));  
441.             } catch (Exception ex) {  
442.                 ex.printStackTrace();  
443.                 System.out.println(task.getContent());  
444.             }  
445.         }  
446.     }  
447.   
448.     /**  
449.      * @param fatherTask  
450.      * @param path  
451.      * @throws Exception  
452.      */  
453.     private void createNewTask(Task fatherTask, String path) throws Exception {  
454.         Task newTask = new Task();  
455.         FilterChain filterchain = new FilterChain();  
456.         filterchain.doFilter(fatherTask, newTask, path, filterchain);  
457.         if (newTask.isValid()) {  
458.             synchronized (SET) {  
459.                 if (SET.contains(newTask.getHost() + newTask.getCurrentPath())) {  
460.                     return;  
461.                 }  
462.                 SET.add(newTask.getHost() + newTask.getCurrentPath());  
463.             }  
464.             reactor.addTask(newTask);  
465.         }  
466.     }  
467.   
468.     private String getPrefix(String s, 
     分享标题：网页主动探测工具-增加Netty模式
     当前地址：http://6mz.cn/article/jhpshc.html