logstash占用cpu过高

之前用logstash读取文件然后入库，占用cpu平均在20%左右，最近加了一个监控目录，cpu平均就飙到80%多了

下面是logstash的配置文件，之前file都是监控单目录，加了一个目录后CPU就飙高了，麻烦各位看下什么问题？

input { 
    file {
                path => ["/data/rsync_files/assistants/web_100/*/users.log","/data/rsync_files/assistants/web_101/*/users.log"]
                type => "users"
                start_position => "beginning"
        codec => "json"
        }  
    file {
                path => ["/data/rsync_files/assistants/web_100/*/user_states.log","/data/rsync_files/assistants/web_101/*/user_states.log"]
                type => "user_states"
                start_position => "beginning"
        codec => "json"
        }
    file {
                path => ["/data/rsync_files/assistants/web_100/*/user_events.log","/data/rsync_files/assistants/web_101/*/user_events.log"]
                type => "user_events"
                start_position => "beginning"
        codec => "json"
        } 

    file {
                path => ["/data/rsync_files/assistants/web_100/*/user_locations.log","/data/rsync_files/assistants/web_101/*/user_locations.log"]
                type => "user_locations"
                start_position => "beginning"
        codec => "json"
        } 

    file {
                path => ["/data/rsync_files/assistants/web_100/*/user_snapshots.log","/data/rsync_files/assistants/web_101/*/user_snapshots.log"]
                type => "user_snapshots"
                start_position => "beginning"
        codec => "json"
        } 
}

filter {
    if [type] == "users" {
        if ![@fields][ctxt_type] {
        mutate {
            add_field => ["[@fields][ctxt_type]", "0"]
        }
        }
        mutate {
        convert => ["[@fields][ctxt_id]", "string"]
        convert => ["[@fields][ctxt_type]", "string"]
        }
    } else if [type] == "user_states" {
        mutate {
        convert => ["[@fields][ctxt_user_id]", "string"]
        }
    } else if [type] == "user_events"{
        mutate {
        convert => ["[@fields][ctxt_id]", "string"]
        convert => ["[@fields][ctxt_user_id]", "string"]
        convert => ["[@fields][ctxt_time]", "string"]
        }
    } else if [type] == "user_locations"{
        mutate {
        convert => ["[@fields][ctxt_user_id]", "string"]
        }
    } else if [type] == "user_snapshots"{
        mutate {
        convert => ["[@fields][ctxt_user_id]", "string"]
        }
    }
}

output {
    if [type] == "users" {
        if [@fields][ctxt_type] and [@fields][ctxt_imei] != '' {
            jdbc {
                driver_jar_path => "/usr/local/jar/mysql-connector-java-5.1.36-bin.jar"
                driver_class => "com.mysql.jdbc.Driver"
                connection_string => "jdbc:mysql://127.0.0.1/analysis?user=root&password=wgc@db123"
                statement => ["insert into user_lists(user_id,device,imei,v,lang,brand,ov,ctime,type) values(?,?,?,?,?,?,?,UNIX_TIMESTAMP(?),?) ON DUPLICATE KEY UPDATE device=VALUES(`device`),v=VALUES(`v`),lang=VALUES(`lang`),brand=VALUES(`brand`),ov=VALUES(`ov`),ctime=VALUES(`ctime`),user_id=VALUES(`user_id`),type=VALUES(type),imei=VALUES(`imei`)", "%{[@fields][ctxt_id]}","%{[@fields][ctxt_device]}","%{[@fields][ctxt_imei]}","%{[@fields][ctxt_v]}","%{[@fields][ctxt_lang]}","%{[@fields][ctxt_brand]}","%{[@fields][ctxt_ov]}","%{[@fields][ctxt_created_at]}","%{[@fields][ctxt_type]}"]
            }
        }
    } else if [type] == "user_states" {
        if [@fields][ctxt_user_id] != "0" {
            jdbc {
                driver_jar_path => "/usr/local/jar/mysql-connector-java-5.1.36-bin.jar"
                driver_class => "com.mysql.jdbc.Driver"
                connection_string => "jdbc:mysql://127.0.0.1/analysis?user=root&password=wgc@db123"
                statement => ["insert into user_lists(user_id,mtime) values(?, UNIX_TIMESTAMP(?) ) ON DUPLICATE KEY UPDATE mtime=VALUES(`mtime`)", "%{[@fields][ctxt_user_id]}", "%{[@fields][ctxt_updated_at]}"]
            }
        }
    } else if [type] == "user_events" {
        jdbc {
            driver_jar_path => "/usr/local/jar/mysql-connector-java-5.1.36-bin.jar"
            driver_class => "com.mysql.jdbc.Driver"
            connection_string => "jdbc:mysql://127.0.0.1/analysis?user=root&password=wgc@db123"        
            statement => [ "INSERT INTO user_events(id,user_id,event,time,name,package,version,created_at,updated_at) values(?,?,?,?,?,?,?,?,?)","%{[@fields][ctxt_id]}","%{[@fields][ctxt_user_id]}","%{[@fields][ctxt_event]}","%{[@fields][ctxt_time]}","%{[@fields][ctxt_name]}","%{[@fields][ctxt_package]}","%{[@fields][ctxt_version]}","%{[@fields][ctxt_created_at]}","%{[@fields][ctxt_updated_at]}" ]
        }
    } else if [type] == "user_locations" {
        if [@fields][ctxt_user_id] != "0" {
            jdbc {
                driver_jar_path => "/usr/local/jar/mysql-connector-java-5.1.36-bin.jar"
                driver_class => "com.mysql.jdbc.Driver"
                connection_string => "jdbc:mysql://127.0.0.1/analysis?user=root&password=wgc@db123"
                statement => [ "insert into user_lists(country_id,country,area_id,area,region_id,region,city_id,city,isp_id,isp,user_id) VALUES(?,?,?,?,?,?,?,?,?,?,?) ON DUPLICATE KEY UPDATE country_id=VALUES(`country_id`), country=VALUES(`country`), area_id=VALUES(`area_id`), area=VALUES(`area`), region_id=VALUES(`region_id`), region=VALUES(`region`), city_id=VALUES(`city_id`), city=VALUES(`city`), isp_id=VALUES(`isp_id`),isp=VALUES(`isp`)","%{[@fields][ctxt_country_id]}","%{[@fields][ctxt_country]}","%{[@fields][ctxt_area_id]}","%{[@fields][ctxt_area]}","%{[@fields][ctxt_region_id]}","%{[@fields][ctxt_region]}","%{[@fields][ctxt_city_id]}","%{[@fields][ctxt_city]}","%{[@fields][ctxt_isp_id]}","%{[@fields][ctxt_isp]}","%{[@fields][ctxt_user_id]}" ]
            }
        }
    } else if [type] == "user_snapshots" {
        if [@fields][ctxt_user_id] != "0" {
            jdbc {
                driver_jar_path => "/usr/local/jar/mysql-connector-java-5.1.36-bin.jar"
                driver_class => "com.mysql.jdbc.Driver"
                connection_string => "jdbc:mysql://127.0.0.1/analysis?user=root&password=wgc@db123"
                statement => [ "INSERT INTO apps_user_lists(`user_id`,`packages`,`decode`) VALUES(?,?,?) ON DUPLICATE KEY UPDATE packages=VALUES(`packages`),decode=0","%{[@fields][ctxt_user_id]}","%{[@fields][ctxt_snapshot]}","%[@fields][ctxt_snapshot]" ]
            }
        }
    }
    
}