网页主动探测工具-增加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://pwwzsj.com/article/jhpshc.html