zoukankan      html  css  js  c++  java
  • 采用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的libext目录下(本人的为:JRE1.8libext),或者拷贝至项目中,然后Add to Build Path 到项目中;

      <2>.将Jpcap.dll拷贝至自己的jdk的bin(本人的为:JRE1.8in)的目录下

          

    3.安装WinPcap_4_1_2.exe

      若未安装这个,则没有相应的环境,后面一直会报错: java.lang.UnsatisfiedLinkError: D:JRE1.8inJpcap.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

  • 相关阅读:
    RDay2-Problem 2 B
    杭电 1862 EXCEL排序(sort+结构体)
    杭电 2803 The MAX(sort)
    杭电 5053 the Sum of Cube(求区间内的立方和)打表法
    杭电 2089 不要62
    杭电 4548 美素数(素数打表)
    杭电2098 分拆素数和
    杭电1722 Cake (分蛋糕)
    素数判定 (素数打表)
    最小公倍数
  • 原文地址:https://www.cnblogs.com/sun-flower1314/p/10630424.html
Copyright © 2011-2022 走看看