采用Jpcap+redis+線程 設備網絡流量監控 應用實戰實例

轉載請注明出處：https://www.cnblogs.com/sun-flower1314/p/10630424.html

一、問題場景：

　　目前需要服務器端去監控部署在各個城市的設備（包括終端、大屏互動設備、廣告機等）的流量情況，包括每台設備對服務器端發出字節數、發出數據包數、最大數據包、IP地址等數據，而設備對其他的服務的訪問不予監控。

二、目前主要采用的技術組合是Jpcap+redis+線程的方式,

　　實現思路是：不斷的從網絡中抓取對應設備對服務器請求的包數據，對包進行分析，然后將數據放入redis緩存中（若需要數據需要可定時存入數據庫進行記錄），每抓到對應的設備的數據就會刷新在緩存中的對應設備的數據。從而實現監控。注：用hash結構存儲， 存入緩存中的是以設備ip為key，用map值作為value，取數據直接根據設備ip就能拿到對應的數據

三、功能實現

環境的搭建

 　1. redis服務的搭建

（我是直接在虛擬機中搭建了一個redis作為開發測試），關於搭建redis服務，請網上自行百度

　 2.底層抓包環境的配置

　　下載jpcap所使用的jar包以及jpcap需要的環境：

　　jpcap所需要的用到的文件為：jpcap.jar  Jpcap.dll 安裝環境：WinPcap_4_1_2.exe 

　　<1>.將jpcap.jar拷貝jdk的lib\ext目錄下（本人的為：JRE1.8\lib\ext），或者拷貝至項目中，然后Add to Build Path 到項目中；

　　<2>.將Jpcap.dll拷貝至自己的jdk的bin（本人的為：JRE1.8\bin）的目錄下

      

 

3.安裝WinPcap_4_1_2.exe

　　若未安裝這個，則沒有相應的環境，后面一直會報錯： java.lang.UnsatisfiedLinkError: D:\JRE1.8\bin\Jpcap.dll: Can't find dependent libraries  如下圖：

 

     

 

4.以下為實現代碼

　　代碼部分較多，會貼主要的，（部分工具類和初始化未貼出）代碼上也有相應注釋。需要完整的源碼，請留言郵箱。

完整的包結構：

       

 

配置文件部分：

      

然后是實現代碼：

　　主要思路是 動態傳入抓包的過濾條件，再啟動一個線程去抓包

 

package com.hxc.hwkj.jpcapcatch.impl;

import java.util.logging.Logger;

import com.hxc.hwkj.core.CatchDataStart;
import com.hxc.hwkj.entity.NetworkInterfaceEntity;
import com.hxc.hwkj.jpcapcatch.CatchDataMonitorStart;
import com.hxc.hwkj.util.LocalWinInfoUtils;

import jpcap.JpcapCaptor;
import jpcap.NetworkInterface;
import jpcap.NetworkInterfaceAddress;

/**
 * @Description 設備接口信息
 * 2019年3月26日  下午3:53:01
 * @Author Huangxiaocong
 */
public class CatchDataMonitorStartImpl implements CatchDataMonitorStart {
    private static Logger log = Logger.getLogger(CatchDataMonitorStartImpl.class.toString());
    /**
     * 通過過濾條件去抓取數據
     * @param cond 設置過濾的條件（包括設備的ip地址，但不僅限於此）
     * @Author Huangxiaocong 2019年3月26日 下午3:40:25
     */
    public void catchDataByConditon(String cond) {
        NetworkInterface[] devices = JpcapCaptor.getDeviceList();
        NetworkInterfaceEntity nICEntity = null;
        for(int i = 0; i < devices.length; i++) {
            NetworkInterface netInterface = devices[i];
            nICEntity = getServerDeciverInfo(netInterface);
            String ipv4 = nICEntity.getIpv4Addr();
            String mac = nICEntity.getMacAddr().replace(':', '-');
            //與當前網卡比較mac地址和ipv4地址
            if(LocalWinInfoUtils.getIpAddress().equals(ipv4) //
                    && LocalWinInfoUtils.getMacAddress().equalsIgnoreCase(mac)) {
                //生成情況  服務器端作為dst 目標  客戶端作為源src
                String filterCond = cond == null || cond.equals("") ? "( dst host " + ipv4 + ")" : cond + " and ( dst host " + ipv4 + ")";
                //log.info("過濾條件為：" + filterCond);
                startCapThread(netInterface, filterCond);
            } else {
                continue;
            }
            System.out.println("設備信息為：" + nICEntity);
        }
    }
    
    /**
     * 啟動一個線程獨立運行抓包
     * @param deviceName
     * @param condition
     * @Author Huangxiaocong 2019年3月26日 下午2:35:51
     */
    public void startCapThread(NetworkInterface deviceName, String filterCond) {
        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                log.info("啟動對 " + deviceName + "抓包線程");
                CatchDataStart catchDataStart = new CatchDataStart();
                catchDataStart.init(deviceName, filterCond);
            }
        };
        new Thread(runnable).start();
    }
    
    /**
     * 獲取網卡的信息
     * @param netInterface 
     * @return
     * @Author Huangxiaocong 2019年3月26日 下午2:22:36
     */
    public NetworkInterfaceEntity getServerDeciverInfo(NetworkInterface netInterface) {
        NetworkInterfaceEntity nICEntity = new NetworkInterfaceEntity();
        nICEntity.setNICName(netInterface.name);
        nICEntity.setNICDesc(netInterface.description);
        nICEntity.setDataLinkName(netInterface.datalink_name);
        nICEntity.setDataLinkDesc(netInterface.datalink_description);
        //計算mac地址
        byte[] bs = netInterface.mac_address;
        StringBuilder sBuilder = new StringBuilder();
        int count = 1;
        for(byte b : bs) {
            sBuilder.append(Integer.toHexString(b & 0xff));
            if(count++ != bs.length) sBuilder.append(":");
        }
        nICEntity.setMacAddr(sBuilder.toString());
        //查找ip地址
        NetworkInterfaceAddress[] netInterAddresses = netInterface.addresses;
        for(int i = 0; i < netInterAddresses.length; i++) {
            if(i == 0) {
                nICEntity.setIpv6Addr(netInterAddresses[i].address.getHostAddress());
            } else if(i == 1) {
                nICEntity.setIpv4Addr(netInterAddresses[i].address.getHostAddress());
                nICEntity.setBroadcase(netInterAddresses[i].broadcast.getHostAddress());
                nICEntity.setSubnet(netInterAddresses[i].subnet.getHostAddress());
            }
        }
        return nICEntity;
    }
}

上面代碼中的  getDeviceList表示可以獲取機器上網絡接口卡對象的數組，數組中的每個NetworkInterface代表一個網絡接口。 

 

package com.hxc.hwkj.core;

import java.io.IOException;
import java.util.logging.Logger;

import com.hxc.hwkj.init.InitJpcat;
import com.hxc.hwkj.jpcapcatch.impl.CatchDataPacketInfo;

import jpcap.JpcapCaptor;
import jpcap.NetworkInterface;

public class CatchDataStart {
    private static Logger log = Logger.getLogger(CatchDataStart.class.toString());
    /**
     * 初始化參數信息 取得在指定網卡上的Jpcapcator對象
     * @param deviceName 網卡名稱
     * @param filterCond 過濾條件
     * @Author Huangxiaocong 2019年3月26日 下午2:47:53
     */
    public void init(NetworkInterface deviceName, String filterCond) {
        JpcapCaptor jpcap = null;
        try {
            jpcap = JpcapCaptor.openDevice(deviceName, InitJpcat.snaplen, InitJpcat.promisc, InitJpcat.to_ms);
            //過濾代碼 可以是協議 端口 IP 組合
            if(filterCond != null && !"".equals(filterCond)) {
                jpcap.setFilter(filterCond, true);
            }
        } catch (IOException e) {
            log.info("打開一個網卡失敗" + e);
        }
        jpcap.loopPacket(InitJpcat.loopCount, new CatchDataPacketInfo(jpcap));
    }
}

上面代碼中關於openDevice方法和loopPacket方法的解釋：

①openDevice(NetworkInterface intrface, int snaplen, boolean promisc, int to_ms)
取得在指定網卡上的Jpcapcator對象，Interface:所返回的某個網卡對象snaplen；snaplen:一次性要抓取數據包的最大長度
promisc：設置是否混雜模式，處於混雜模式將接收所有數據包，如果設置為混雜模式后，調用了包過濾函數setFilter()將不起任何作用；
to_ms : 這個參數主要用於processPacket()方法，指定超時的時間。

 ②int loopPacket(int count, PacketReceiver handler) ：通過openDevice方法取得每個網絡接口上的JpcapCaptor對象，就可通過這個方法抓包了。

count：表示要抓的包的數目，如果設置為-1表示永遠抓下去；handler：第二個參數必須是實現了PacketReceiver接口的一個對象，抓到的包將調用這個對象的
receivePacket方法處理，這個方法調用會阻塞等待 與該方法相對應的是breakLoop()，在JacapCaptor對象上的阻塞等待的方法將強制終止

 

下面這部分是對包的解析，包括各種協議： 

package com.hxc.hwkj.jpcapcatch.impl;

import java.util.HashMap;
import java.util.Map;
import java.util.logging.Logger;

import com.hxc.hwkj.jpcapcatch.CatchDataToCache;

import jpcap.JpcapCaptor;
import jpcap.PacketReceiver;
import jpcap.packet.ARPPacket;
import jpcap.packet.EthernetPacket;
import jpcap.packet.ICMPPacket;
import jpcap.packet.Packet;
import jpcap.packet.TCPPacket;
import jpcap.packet.UDPPacket;
/**
 * @Description 抓包獲取數據並分析信息  
 * 2019年3月26日  下午2:42:12
 * @Author Huangxiaocong
 */
public class CatchDataPacketInfo implements PacketReceiver {
    private static Logger log = Logger.getLogger(CatchDataPacketInfo.class.toString());
    
    private JpcapCaptor jpcap = null;
    public CatchDataPacketInfo(JpcapCaptor jpcap) {
        this.jpcap = jpcap;
    }
    
    /**
     * 解析包信息
     */
    @Override
    public void receivePacket(Packet packet) {
        //封裝抓包獲取數據
        Map<String, String> infoMap = new HashMap<>();
        //分析協議類型
        if(packet instanceof ARPPacket) { //該協議無端口號
            ARPPacket arpPacket = (ARPPacket) packet;
            infoMap.put("ContractType", "ARP協議");
            infoMap.put("Caplen", String.valueOf(arpPacket.caplen));
            infoMap.put("SecTime", String.valueOf(arpPacket.sec));
            infoMap.put("SourceIp", arpPacket.getSenderProtocolAddress().toString().replace("/", ""));
            infoMap.put("SourceMacAddr", arpPacket.getSenderHardwareAddress().toString());
            infoMap.put("TargetIp", arpPacket.getTargetProtocolAddress().toString().replace("/", ""));
            infoMap.put("TargetMacAddr", arpPacket.getTargetHardwareAddress().toString());
        } else if(packet instanceof UDPPacket) {
            UDPPacket udpPacket = (UDPPacket) packet;
            EthernetPacket datalink = (EthernetPacket) udpPacket.datalink;
            infoMap.put("ContractType", "UDP協議");
            infoMap.put("Caplen", String.valueOf(udpPacket.caplen));
            infoMap.put("SecTime", String.valueOf(udpPacket.sec));
            infoMap.put("SourceIp", udpPacket.src_ip.getHostAddress());
            infoMap.put("SourcePort", String.valueOf(udpPacket.src_port));
            
            infoMap.put("SourceMacAddr", getMacInfo(datalink.src_mac));
            infoMap.put("TargetIp", udpPacket.dst_ip.getHostAddress());
            infoMap.put("TargetPort", String.valueOf(udpPacket.dst_port));
            infoMap.put("TargetMacAddr", getMacInfo(datalink.dst_mac));
        } else if(packet instanceof TCPPacket) {
            TCPPacket tcpPacket = (TCPPacket) packet;
            EthernetPacket datalink = (EthernetPacket) tcpPacket.datalink;
            infoMap.put("ContractType", "TCP協議");
            infoMap.put("Caplen", String.valueOf(tcpPacket.caplen));
            infoMap.put("SecTime", String.valueOf(tcpPacket.sec));
            infoMap.put("SourceIp", tcpPacket.src_ip.getHostAddress());
            infoMap.put("SourcePort", String.valueOf(tcpPacket.src_port));
            
            infoMap.put("SourceMacAddr", getMacInfo(datalink.src_mac));
            infoMap.put("TargetIp", tcpPacket.dst_ip.getHostAddress());
            infoMap.put("TargetPort", String.valueOf(tcpPacket.dst_port));
            infoMap.put("TargetMacAddr", getMacInfo(datalink.dst_mac));
        } else if(packet instanceof ICMPPacket) { //該協議無端口號
            ICMPPacket icmpPacket = (ICMPPacket) packet;
            EthernetPacket datalink = (EthernetPacket) icmpPacket.datalink;
            infoMap.put("ContractType", "ICMP協議");
            infoMap.put("Caplen", String.valueOf(icmpPacket.caplen));
            infoMap.put("SecTime", String.valueOf(icmpPacket.sec));
            infoMap.put("SourceIp", icmpPacket.src_ip.getHostAddress());
            
            infoMap.put("SourceMacAddr", getMacInfo(datalink.src_mac));
            infoMap.put("TargetIp", icmpPacket.dst_ip.getHostAddress());
            infoMap.put("TargetMacAddr", getMacInfo(datalink.dst_mac));
        }
        try {
            CatchDataToCache catchDataToCache = new CatchDataToCacheImpl();
            catchDataToCache.setInfoToCache(infoMap);
        } catch (Exception e) {
            log.info("抓取數據裝入緩存時 出現異常，請檢查：" + e);
            jpcap.breakLoop();
            if(jpcap != null) {
                jpcap.close();
            }
        }
        
    }
    /**
     * 獲取Mac信息
     * @param macByte
     * @return
     * @Author Huangxiaocong 2019年3月24日 下午3:19:30
     */
    protected String getMacInfo(byte[] macByte) {
        StringBuffer srcMacStr = new StringBuffer(); 
        int count = 1;
        for (byte b : macByte) {  
            srcMacStr.append(Integer.toHexString(b & 0xff));
            if(count++ != macByte.length) 
                srcMacStr.append(":");
        }
        return srcMacStr.toString();
    }
}

 

 下面部分代碼是將數據存入redis中，有些數據是根據需要進行覆蓋或者疊加：

package com.hxc.hwkj.jpcapcatch.impl;

import java.util.Map;

import com.hxc.hwkj.jpcapcatch.CatchDataToCache;
import com.hxc.hwkj.util.JedisPoolUtils;
import redis.clients.jedis.Jedis;

/**
 * @Description 將抓取到的數據裝入緩存中 
 * 2019年3月27日  下午4:52:50
 * @Author Huangxiaocong
 */
public class CatchDataToCacheImpl implements CatchDataToCache {
    /**
     * 將數據存入redis緩存中
     * @param infoMap
     * @Author Huangxiaocong 2019年3月26日 下午4:24:07
     */
    public void setInfoToCache(Map<String, String> infoMap) throws Exception {
        if(infoMap.isEmpty()) {
            return ;
        }
        String deviceIp = infoMap.get("SourceIp");
        if(deviceIp == null || deviceIp.equals("")) {
            return ;
        }
        Jedis jedis = null;
        jedis = JedisPoolUtils.getJedis();
        //處理數據，找出最大的包
        String caplen = jedis.hget(deviceIp, "MaxCaplen");
        if(caplen == null || caplen.equals("")) {
            caplen = "0";
        }
        int nowCaplen = Integer.parseInt(infoMap.get("Caplen"));
        if(Integer.parseInt(caplen) < nowCaplen) {
            infoMap.put("MaxCaplen", String.valueOf(nowCaplen));
        } else {
            infoMap.put("MaxCaplen", caplen);
        }
        jedis.hmset(deviceIp, infoMap);
        //發包次數
        jedis.hincrBy(deviceIp, "counttimes", 1);
        //生產現場需要刪除
        /*Map<String, String> tempMap = jedis.hgetAll(deviceIp);
        Set<Entry<String, String>> entry = tempMap.entrySet();
        for(Entry<String, String> en : entry ) {
            System.out.println(en.getKey() + " --- > " + en.getValue());
        }*/
        JedisPoolUtils.closeJedisResource(jedis);
    }
}

 測試部分，只需傳入過濾條件就行：

package com.hxc.hwkj.test;

import com.hxc.hwkj.jpcapcatch.CatchDataMonitorStart;
import com.hxc.hwkj.jpcapcatch.impl.CatchDataMonitorStartImpl;

public class MainTest {
    public static void main(String[] args) {
        CatchDataMonitorStart deciveMonitorStart = new CatchDataMonitorStartImpl();
        String cond = "src host 192.168.1.111";
        deciveMonitorStart.catchDataByConditon(cond);
    }
}

 

 注：jpcap的過濾條件規則：

 如：src host 192.168.1.111 表示過濾源主機地址為192.168.1.111的數據

1.dst host xx.xx.xx.xx  過濾目標主機ip為   xx.xx.xx.xx的數據

2.host xx.xx.xx.xx 過濾目標或源主機為xx.xx.xx.xx的數據

3.src port xx 過濾源端口號為xx的數據

4.http or telnet 過濾http或telnet協議的數據

等等...

其條件可以組合使用：

取反操作 (`!' 或 `not').
連接操作 (`&&' 或 `and').
選擇操作 (`||' 或 `or').

 這里列出的為常用的過濾表達式，關於更多的過濾條件請百度

 

以上就是實現，若有任何疑問或需要，可留言，歡迎點贊評論，謝謝你的鼓勵！！

 

轉載請注明出處：https://www.cnblogs.com/sun-flower1314/p/10630424.html