snifferparser.java

流量分析 可以对SNIFFER抓的包进行分析

package com.tianxun.NEI.sniffer;

import com.tianxun.NEI.sniffer.util.NetFlowRecord;
import com.tianxun.NEI.sniffer.util.SnifferData;
import com.tianxun.NEI.sniffer.util.GscDiag;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 解析Sniffer网络数据包的线程.
 *
 * @author 聂军
 * @version 1.0 2004-9-13
 */

public class SnifferParser extends Thread {

    private static final int THREAD_SLEEP_MILLIS_TIME =
        NSXSniffer.getConfigParam("Parse.thread.sleep.millis.time", 2);

    private static final int THREAD_SLEEP_NANOS_TIME =
        NSXSniffer.getConfigParam("Parse.thread.sleep.nanos.time", 10000);

    private static final int MAX_NUM_DELETED_ONCE =
        NSXSniffer.getConfigParam("Parse.thread.max.delete.num", 1000);

    private long millisTime = 0;
    private int nanosTime = 0;

    // 在一次线程运行中连续删除的最大记录数量
    private int numberDeleted = 1;

    // 选择不同存储有效记录的标志
    private boolean flag = false;

    // 原始数据缓冲区
    private List snifferDatas = null;

    // 缓冲有效NetFlow Record
    private Map validRecords = null;

    // 缓冲有效NetFlow Record
    private Map validRecords1 = null;

    // 当前使用的有效缓冲区
    private Map currentValidRecords = null;

    /**
     * 构造器.
     *
     * @param l - 原始Sniffer数据集合.
     * @param ht - 没有过期的NetFlow包.
     * @param v - 已经过期的NetFlow包.
     */
    public SnifferParser(List l) {
        snifferDatas = l;
        validRecords = new HashMap(
            NSXSniffer.getConfigParam("Default.flow.buffer.size", 5011));
        validRecords1 = new HashMap(
            NSXSniffer.getConfigParam("Default.flow.buffer.size", 5013));
        currentValidRecords = validRecords;

        if (THREAD_SLEEP_MILLIS_TIME < 0) {
            millisTime = 0;
        } else {
            millisTime = THREAD_SLEEP_MILLIS_TIME;
        }

        if (THREAD_SLEEP_NANOS_TIME <= 0) {
            nanosTime = 1;
        } else if (THREAD_SLEEP_NANOS_TIME > 999999) {
            nanosTime = 999999;
        } else {
            nanosTime = THREAD_SLEEP_NANOS_TIME;
        }

        if (MAX_NUM_DELETED_ONCE < 1) {
            numberDeleted = 1;
        } else if (MAX_NUM_DELETED_ONCE > 1000) {
            numberDeleted = 1000;
        } else {
            numberDeleted = MAX_NUM_DELETED_ONCE;
        }
    }

    /**
     * @see java.lang.Thread#run()
     */
    public void run() {
        Filter filter = NSXSniffer.getApp().getFilter();
        int size;
        while (true) {
            size = snifferDatas.size();
            if (size > numberDeleted) {
                size = numberDeleted;
            }

            try {
                for (int i = 0; i < size; i++) {    
                    parseSnifferData((SnifferData) snifferDatas.remove(0));
                }
            } catch (Exception ex) {
                ex.printStackTrace(System.out);
                GscDiag.printStackTrace(5, "SnifferParserException:",ex);
            }

            try {
                sleep(millisTime, nanosTime);
            } catch(Exception ex) {
                ex.printStackTrace(System.out);
            }
        }
    }

    private void parseSnifferData(SnifferData packet) {
        if (currentValidRecords.containsKey(packet)) {//把相同的包进行合并
            NetFlowRecord record = (NetFlowRecord)
                currentValidRecords.get(packet);
            record.addDPkts(1);
            record.addDOctets(packet.getLength());
        } else {
            // 创建一个新的NetFlow记录
            synchronized(currentValidRecords) {
                currentValidRecords.put(packet, createNewFlow(packet));
            }
        }
    }

    private NetFlowRecord createNewFlow(SnifferData packet) {
        NetFlowRecord record = new NetFlowRecord(
            packet.getSrcaddr(), packet.getDstaddr(),
            packet.getSrcport(), packet.getDstport(),
            packet.getProtocol(), packet.getTos());
        record.addDPkts(1);
        record.addDOctets(packet.getLength());
        return record;
    }

    /**
     * 得到当前正在使用的有效记录缓冲区.
     */
    public Map getCurrentValidBuffer() {
        flag = !flag;
        if (flag) {
            synchronized(currentValidRecords) {
                currentValidRecords = validRecords1;
            }
            return validRecords;
        } else {
            synchronized(currentValidRecords) {
                currentValidRecords = validRecords;
            }
            return validRecords1;
        }   //相当与得到ValidRecords之后，MAP清空。写的好
    }

}