Authored by zhengyouwei

任务配置管理

Showing 19 changed files with 869 additions and 2 deletions
  1 +package com.monitor.common.util;
  2 +
  3 +import org.springframework.beans.BeansException;
  4 +import org.springframework.context.ApplicationContext;
  5 +import org.springframework.context.ApplicationContextAware;
  6 +import org.springframework.stereotype.Component;
  7 +
  8 +/**
  9 + * Created by zhengyouwei on 2016/8/12.
  10 + */
  11 +@Component
  12 +public class ApplicationUtil implements ApplicationContextAware {
  13 + // Spring应用上下文环境
  14 + private static ApplicationContext applicationContext;
  15 +
  16 + /**
  17 + * 实现ApplicationContextAware接口的回调方法,设置上下文环境
  18 + */
  19 + public void setApplicationContext(ApplicationContext applicationContext)
  20 + throws BeansException {
  21 + ApplicationUtil.applicationContext = applicationContext;
  22 + }
  23 +
  24 + public static ApplicationContext getApplicationContext() {
  25 + return applicationContext;
  26 + }
  27 +
  28 + /**
  29 + * 获取对象 这里重写了bean方法,起主要作用
  30 + */
  31 + public static Object getBean(String beanId) throws BeansException {
  32 + return applicationContext.getBean(beanId);
  33 + }
  34 +
  35 + /**
  36 + * 获取对象 这里重写了bean方法,起主要作用
  37 + */
  38 + public static Object getBean(Class T) throws BeansException {
  39 + return applicationContext.getBean(T);
  40 + }
  41 +}
@@ -13,7 +13,7 @@ public class InfluxDBQuery { @@ -13,7 +13,7 @@ public class InfluxDBQuery {
13 13
14 protected final Logger logger = LoggerFactory.getLogger(getClass()); 14 protected final Logger logger = LoggerFactory.getLogger(getClass());
15 15
16 - protected QueryResult query(String influxDBName, String command, String database){ 16 + public QueryResult query(String influxDBName, String command, String database){
17 Query query = new Query(command, database); 17 Query query = new Query(command, database);
18 try { 18 try {
19 QueryResult queryResult = inluxDBSingle.getInfluxDBByName(influxDBName) 19 QueryResult queryResult = inluxDBSingle.getInfluxDBByName(influxDBName)
  1 +package com.monitor.influxdb.mapper.impl;
  2 +
  3 +import com.monitor.influxdb.InfluxDBQuery;
  4 +import org.springframework.stereotype.Service;
  5 +
  6 +/**
  7 + * Created by zhengyouwei on 2016/8/12.
  8 + */
  9 +@Service("CommonQuery")
  10 +public class CommonQuery extends InfluxDBQuery{
  11 +}
  1 +package com.model;
  2 +
  3 +import lombok.Data;
  4 +
  5 +/**
  6 + * Created by zhengyouwei on 2016/8/10.
  7 + */
  8 +@Data
  9 +public class TaskModel {
  10 +
  11 + private int id;
  12 +
  13 + /**
  14 + * 查询的列
  15 + */
  16 + private String fields;
  17 +
  18 +
  19 + /**
  20 + * 查询使用的tag
  21 + */
  22 + private String tags;
  23 +
  24 + /**
  25 + * 查询的influxdbs
  26 + */
  27 + private String influxdbs;
  28 +
  29 + /**
  30 + * 查询的influxdb 数据库
  31 + */
  32 + private String database;
  33 +
  34 + /**
  35 + * sql语句
  36 + */
  37 + private String influxSql;
  38 +
  39 + /**
  40 + * 告警号码
  41 + */
  42 + private String mobile;
  43 +
  44 + /**
  45 + * 告警短信${}表示替换词
  46 + */
  47 + private String sms;
  48 +
  49 + /**
  50 + * alarm type
  51 + */
  52 + private String type;
  53 +
  54 + /**
  55 + *执行间隔
  56 + */
  57 + private int interval;
  58 +
  59 + /**
  60 + * 休息开始时间 H
  61 + */
  62 + private int relaxStartTime;
  63 +
  64 + /**
  65 + * 休息结束时间 H
  66 + */
  67 + private int relaxEndTime;
  68 +
  69 + /**
  70 + * 1 一直执行任务 2 有休息
  71 + */
  72 + private int timeFlag;
  73 +
  74 + /**
  75 + * 告警方式
  76 + * 1:单个influxdb 未出出现想要的数据
  77 + * 2:多个influxdb 合起来未出现想要的数据
  78 + * 3:单个influxdb 出现次数超过一定值
  79 + * 4:出现一个就告警,单条记录单独告警
  80 + */
  81 + private int alarmType;
  82 +
  83 + /**
  84 + * 数量告警
  85 + */
  86 + private int alarmNumFlag;
  87 +
  88 + /**
  89 + * 是否在运行中 1 : 运行中
  90 + */
  91 + private int isRun;
  92 +
  93 + /**
  94 + * 是否需要监控 1: 需要监控
  95 + */
  96 + private int isOn;
  97 +
  98 + /**
  99 + * 持续需要告警的次数
  100 + */
  101 + private int alarmTimes;
  102 +
  103 + /**
  104 + *上次执行时间
  105 + */
  106 + private String lastTime;
  107 +
  108 + /**
  109 + * 描述
  110 + */
  111 + private String description;
  112 +
  113 +}
  1 +package com.monitor.mysql.mapper;
  2 +
  3 +import com.model.TaskModel;
  4 +import com.monitor.model.domain.PageBean;
  5 +
  6 +import java.util.List;
  7 +
  8 +/**
  9 + * Created by zhengyouwei on 2016/7/21.
  10 + */
  11 +public interface TaskSheduleMapper {
  12 +
  13 + List<TaskModel> selectAll();
  14 +
  15 + TaskModel selectById(int id);
  16 +
  17 + List<TaskModel> selectByPage(PageBean page);
  18 +
  19 + int deleteById(int id);
  20 +
  21 + int updateByid(TaskModel taskModel);
  22 +
  23 + int insert(TaskModel taskModel);
  24 +
  25 + int selectCount();
  26 +
  27 + int updateLastTime(int id);
  28 +}
  1 +<?xml version="1.0" encoding="UTF-8" ?>
  2 +<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
  3 +<mapper namespace="com.monitor.mysql.mapper.TaskSheduleMapper">
  4 + <resultMap id="BaseResultMap" type="com.model.TaskModel">
  5 + <id column="id" property="id" jdbcType="INTEGER"/>
  6 + <result column="alarm_interval" property="interval" jdbcType="INTEGER"/>
  7 + <result column="relax_start_time" property="relaxStartTime" jdbcType="INTEGER"/>
  8 + <result column="relax_end_time" property="relaxEndTime" jdbcType="INTEGER"/>
  9 + <result column="time_flag" property="timeFlag" jdbcType="INTEGER"/>
  10 + <result column="alarm_type" property="alarmType" jdbcType="INTEGER"/>
  11 + <result column="alarm_num_flag" property="alarmNumFlag" jdbcType="INTEGER"/>
  12 + <result column="is_run" property="isRun" jdbcType="INTEGER"/>
  13 + <result column="is_on" property="isOn" jdbcType="INTEGER"/>
  14 + <result column="continuous_alarm_times" property="alarmTimes" jdbcType="INTEGER"/>
  15 + <result column="fields" property="fields" jdbcType="VARCHAR"/>
  16 + <result column="tags" property="tags" jdbcType="VARCHAR"/>
  17 + <result column="influxdbs" property="influxdbs" jdbcType="VARCHAR"/>
  18 + <result column="influx_sql" property="influxSql" jdbcType="VARCHAR"/>
  19 + <result column="mobile" property="mobile" jdbcType="VARCHAR"/>
  20 + <result column="sms" property="sms" jdbcType="VARCHAR"/>
  21 + <result column="type" property="type" jdbcType="VARCHAR"/>
  22 + <result column="inf_database" property="database" jdbcType="VARCHAR"/>
  23 + <result column="description" property="description" jdbcType="VARCHAR"/>
  24 + <result column="last_time" property="lastTime"/>
  25 + </resultMap>
  26 +
  27 + <select id="selectById" resultMap="BaseResultMap" parameterType="java.lang.Integer">
  28 + select
  29 + *
  30 + from task_shedule
  31 + where id = #{id,jdbcType=INTEGER}
  32 + </select>
  33 +
  34 + <select id="selectAll" resultMap="BaseResultMap">
  35 + select
  36 + *
  37 + from task_shedule
  38 + </select>
  39 +
  40 + <select id="selectByPage" resultMap="BaseResultMap">
  41 + select
  42 + *
  43 + from task_shedule
  44 + order by id
  45 + limit #{startIndex},#{pageSize}
  46 + </select>
  47 +
  48 + <select id="selectCount" resultType="java.lang.Integer">
  49 + select
  50 + count(1)
  51 + from task_shedule
  52 + </select>
  53 +
  54 + <delete id="deleteById" parameterType="java.lang.Integer">
  55 + delete from task_shedule
  56 + where id = #{id,jdbcType=INTEGER}
  57 + </delete>
  58 +
  59 + <update id="updateByid" parameterType="com.model.TaskModel">
  60 + update task_shedule
  61 + set alarm_interval = #{interval,jdbcType=INTEGER},
  62 + relax_start_time = #{relaxStartTime,jdbcType=INTEGER},
  63 + time_flag = #{timeFlag,jdbcType=INTEGER},
  64 + relax_end_time = #{relaxEndTime,jdbcType=INTEGER},
  65 + alarm_type = #{alarmType,jdbcType=INTEGER},
  66 + alarm_num_flag = #{alarmNumFlag,jdbcType=INTEGER},
  67 + is_on = #{isOn,jdbcType=INTEGER},
  68 + fields = #{fields,jdbcType=VARCHAR},
  69 + tags = #{tags,jdbcType=VARCHAR},
  70 + influxdbs = #{influxdbs,jdbcType=VARCHAR},
  71 + influx_sql = #{influxSql,jdbcType=VARCHAR},
  72 + mobile = #{mobile,jdbcType=VARCHAR},
  73 + sms = #{sms,jdbcType=VARCHAR},
  74 + type = #{type,jdbcType=VARCHAR},
  75 + description = #{description,jdbcType=VARCHAR},
  76 + inf_database = #{database,jdbcType=VARCHAR}
  77 + where id = #{id,jdbcType=INTEGER}
  78 + </update>
  79 +
  80 + <insert id="insert" parameterType="com.model.TaskModel">
  81 + insert into task_shedule
  82 + (alarm_interval,
  83 + relax_start_time,
  84 + time_flag,
  85 + relax_end_time,
  86 + alarm_num_flag,
  87 + alarm_type,
  88 + is_on,
  89 + fields,
  90 + tags,
  91 + influxdbs,
  92 + influx_sql,
  93 + mobile,
  94 + sms,
  95 + type ,
  96 + inf_database,
  97 + description)
  98 + values
  99 + (#{interval,jdbcType=INTEGER},
  100 + #{relaxStartTime,jdbcType=INTEGER},
  101 + #{timeFlag,jdbcType=INTEGER},
  102 + #{relaxEndTime,jdbcType=INTEGER},
  103 + #{alarmNumFlag,jdbcType=INTEGER},
  104 + #{alarmType,jdbcType=INTEGER},
  105 + #{isOn,jdbcType=INTEGER},
  106 + #{fields,jdbcType=VARCHAR},
  107 + #{tags,jdbcType=VARCHAR},
  108 + #{influxdbs,jdbcType=VARCHAR},
  109 + #{influxSql,jdbcType=VARCHAR},
  110 + #{mobile,jdbcType=VARCHAR},
  111 + #{sms,jdbcType=VARCHAR},
  112 + #{type,jdbcType=VARCHAR},
  113 + #{database,jdbcType=VARCHAR},
  114 + #{description,jdbcType=VARCHAR}
  115 + )
  116 + </insert>
  117 +
  118 + <update id="updateLastTime" parameterType="java.lang.Integer">
  119 + update task_shedule
  120 + set last_time = now()
  121 + where id = #{id,jdbcType=INTEGER}
  122 + </update>
  123 +</mapper>
@@ -20,6 +20,11 @@ @@ -20,6 +20,11 @@
20 </dependency> 20 </dependency>
21 21
22 <dependency> 22 <dependency>
  23 + <groupId>monitor-service</groupId>
  24 + <artifactId>monitor-service-mysql</artifactId>
  25 + </dependency>
  26 +
  27 + <dependency>
23 <groupId>commons-codec</groupId> 28 <groupId>commons-codec</groupId>
24 <artifactId>commons-codec</artifactId> 29 <artifactId>commons-codec</artifactId>
25 </dependency> 30 </dependency>
  1 +package com.monitor.other.task;
  2 +
  3 +import com.model.TaskModel;
  4 +
  5 +import java.util.List;
  6 +import java.util.Map;
  7 +
  8 +/**
  9 + * Created by zhengyouwei on 2016/8/11.
  10 + */
  11 +public interface AlarmTypeHandler {
  12 +
  13 + void handle(TaskModel taskModel, Map<String, List<List<Map<String, Object>>>> result);
  14 +
  15 +}
  1 +package com.monitor.other.task;
  2 +
  3 +import com.model.TaskModel;
  4 +import com.monitor.common.util.ApplicationUtil;
  5 +import com.monitor.mysql.mapper.TaskSheduleMapper;
  6 +import org.slf4j.Logger;
  7 +import org.slf4j.LoggerFactory;
  8 +import org.springframework.scheduling.TaskScheduler;
  9 +
  10 +import java.util.List;
  11 +import java.util.concurrent.ScheduledFuture;
  12 +
  13 +/**
  14 + * Created by zhengyouwei on 2016/8/17.
  15 + */
  16 +public class NewTask implements Runnable{
  17 + Logger logger = LoggerFactory.getLogger(NewTask.class);
  18 +
  19 + @Override
  20 + public void run() {
  21 + try {
  22 + TaskSheduleMapper taskSheduleMapper = (TaskSheduleMapper) ApplicationUtil.getBean(TaskSheduleMapper.class);
  23 + TaskScheduler taskScheduler = (TaskScheduler) ApplicationUtil.getBean(TaskScheduler.class);
  24 + List<TaskModel> list = taskSheduleMapper.selectAll();
  25 + for (TaskModel taskModel : list){
  26 + if(taskModel.getIsOn() != 1){
  27 + continue;
  28 + }
  29 + if (!TaskStroe.isOn(taskModel.getId())){
  30 + SingleTask singleTask = new SingleTask(taskModel.getId());
  31 + ScheduledFuture scheduledFuture = taskScheduler.scheduleAtFixedRate(singleTask, taskModel.getInterval() * 60 * 1000);
  32 + TaskStroe.addTask(taskModel.getId(),scheduledFuture);
  33 + }
  34 + }
  35 + }catch (Exception e){
  36 + logger.error("new Task ",e);
  37 + }
  38 +
  39 + }
  40 +
  41 +}
  1 +package com.monitor.other.task;
  2 +
  3 +import org.springframework.scheduling.TaskScheduler;
  4 +
  5 +import javax.annotation.PostConstruct;
  6 +
  7 +
  8 +/**
  9 + * Created by zhengyouwei on 2016/8/10.
  10 + */
  11 +public class SheduleTask {
  12 +
  13 + private TaskScheduler taskScheduler;
  14 +
  15 + @PostConstruct
  16 + public void startTasks() {
  17 + taskScheduler.scheduleAtFixedRate(new NewTask(),1* 60 * 1000);//启动发现新任务线程
  18 + }
  19 +
  20 + public void setTaskScheduler(TaskScheduler taskScheduler) {
  21 + this.taskScheduler = taskScheduler;
  22 + }
  23 +}
  1 +package com.monitor.other.task;
  2 +
  3 +
  4 +import com.model.TaskModel;
  5 +import com.monitor.common.util.ApplicationUtil;
  6 +import com.monitor.influxdb.InfluxDBQuery;
  7 +import com.monitor.influxdb.mapper.impl.CommonQuery;
  8 +import com.monitor.influxdb.util.QueryResultUtil;
  9 +import com.monitor.mysql.mapper.TaskSheduleMapper;
  10 +import org.apache.commons.lang.StringUtils;
  11 +import org.influxdb.dto.QueryResult;
  12 +import org.slf4j.Logger;
  13 +import org.slf4j.LoggerFactory;
  14 +import org.springframework.beans.BeansException;
  15 +import org.springframework.context.ApplicationContext;
  16 +import org.springframework.context.ApplicationContextAware;
  17 +
  18 +import java.util.*;
  19 +
  20 +/**
  21 + * Created by zhengyouwei on 2016/8/11.
  22 + */
  23 +public class SingleTask implements Runnable {
  24 +
  25 + Logger logger = LoggerFactory.getLogger(SingleTask.class);
  26 +
  27 + private int id;
  28 +
  29 + public SingleTask(int id) {
  30 + this.id = id;
  31 +
  32 + }
  33 +
  34 + @Override
  35 + public void run() {
  36 + try {
  37 + TaskSheduleMapper taskSheduleMapper = (TaskSheduleMapper)ApplicationUtil.getBean(TaskSheduleMapper.class);
  38 + TaskModel taskModel = taskSheduleMapper.selectById(id);
  39 + if (taskModel == null) {
  40 + return;
  41 + }
  42 + if (taskModel.getIsOn() != 1) {//监控关闭
  43 + TaskStroe.removeTask(id);//关闭任务
  44 + return;
  45 + }
  46 + if (taskModel.getTimeFlag() == 2) { //判断是否在执行时间
  47 + Calendar calendar = Calendar.getInstance();
  48 + int hour = calendar.get(Calendar.HOUR_OF_DAY);
  49 + if (hour >= taskModel.getRelaxStartTime() && hour < taskModel.getRelaxEndTime()) {//在休息时间
  50 + return;
  51 + }
  52 + }
  53 + //获取数据
  54 + Map<String, List<List<Map<String, Object>>>> result = getResult(taskModel);
  55 + taskSheduleMapper.updateLastTime(id);//更新时间
  56 + //按照告警类型进行处理
  57 + AlarmTypeHandler alarmTypeHandler = getHandler(taskModel.getAlarmType());
  58 + alarmTypeHandler.handle(taskModel,result);
  59 +
  60 + } catch (Exception e) {
  61 + logger.error("excute task error,id = " + id, e);
  62 + }
  63 +
  64 + }
  65 +
  66 + public AlarmTypeHandler getHandler(int alarmType){
  67 + return (AlarmTypeHandler) ApplicationUtil.getApplicationContext().getBean("handlerType_" + alarmType);
  68 + }
  69 +
  70 + /**
  71 + * 获取数据
  72 + * @param taskModel
  73 + * @return
  74 + */
  75 + public Map<String, List<List<Map<String, Object>>>> getResult(TaskModel taskModel) {
  76 + String[] influxdbs = taskModel.getInfluxdbs().split(",");
  77 + String[] fields = taskModel.getFields().split(",");
  78 + String[] tags = null;
  79 + if(StringUtils.isNotBlank(taskModel.getTags())){
  80 + tags = taskModel.getTags().split(",");
  81 + }
  82 + Map<String, List<List<Map<String, Object>>>> resultMap = new HashMap<>();
  83 +
  84 + for (String influxdb : influxdbs) {
  85 + QueryResult queryResult = ((CommonQuery)ApplicationUtil.getBean(CommonQuery.class)).query(influxdb, taskModel.getInfluxSql(), taskModel.getDatabase());
  86 + List<QueryResult.Series> seriesList = QueryResultUtil.getSeries(queryResult);
  87 + if (seriesList == null) {
  88 + continue;
  89 + }
  90 + List<List<Map<String, Object>>> seriesResultList = new ArrayList<>();
  91 + for (QueryResult.Series series : seriesList) {
  92 + List<Map<String, Object>> valueResultList = new ArrayList<>();
  93 + List<List<Object>> values = series.getValues();
  94 + for (List<Object> objects : values) {
  95 + Map<String, Object> map = new HashMap<>();
  96 + for (int i = 0; i < fields.length; i++) {
  97 + map.put(fields[i], objects.get(i + 1));
  98 + }
  99 + if(tags != null){
  100 + for (int i = 0; i < tags.length; i++) {
  101 + map.put(tags[i], series.getTags().get(tags[i]));
  102 + }
  103 + }
  104 + valueResultList.add(map);
  105 + }
  106 + seriesResultList.add(valueResultList);
  107 + }
  108 + resultMap.put(influxdb, seriesResultList);
  109 + }
  110 + return resultMap;
  111 + }
  112 +
  113 +
  114 +}
  1 +package com.monitor.other.task;
  2 +
  3 +import com.model.TaskModel;
  4 +import com.monitor.model.domain.PageBean;
  5 +import com.monitor.model.page.PageRequest;
  6 +import com.monitor.model.page.PageResponse;
  7 +import com.monitor.model.response.BaseResponse;
  8 +import com.monitor.mysql.mapper.TaskSheduleMapper;
  9 +import org.slf4j.Logger;
  10 +import org.slf4j.LoggerFactory;
  11 +import org.springframework.beans.factory.annotation.Autowired;
  12 +import org.springframework.stereotype.Controller;
  13 +import org.springframework.util.CollectionUtils;
  14 +import org.springframework.web.bind.annotation.RequestBody;
  15 +import org.springframework.web.bind.annotation.RequestMapping;
  16 +import org.springframework.web.bind.annotation.ResponseBody;
  17 +
  18 +import java.util.List;
  19 +
  20 +/**
  21 + * Created by zhengyouwei on 2016/8/15.
  22 + */
  23 +@Controller
  24 +@RequestMapping("taskConfigure")
  25 +public class TaskConfigureCtrl {
  26 +
  27 + Logger log = LoggerFactory.getLogger(TaskConfigureCtrl.class);
  28 +
  29 + @Autowired
  30 + private TaskSheduleMapper taskSheduleMapper;
  31 +
  32 + @RequestMapping("/getTaskConfigure")
  33 + @ResponseBody
  34 + public BaseResponse<PageResponse<TaskModel>> getTaskConfigure(@RequestBody PageRequest request) {
  35 + try {
  36 + // 组装分页对象
  37 + PageBean page = PageBean.initPageInfo(request.getCurrentPage(),
  38 + request.getPageSize(), request);
  39 + // 先查询符合条件的总数量
  40 + int total = taskSheduleMapper.selectCount();
  41 + // 数量为0 直接返回
  42 + if (total == 0) {
  43 + // 返回初始page对象
  44 + return null;
  45 + }
  46 + // 获取列表
  47 + List<TaskModel> taskModels = taskSheduleMapper.selectByPage(page);
  48 + if (CollectionUtils.isEmpty(taskModels)) {
  49 + return null;
  50 + }
  51 + PageResponse<TaskModel> response = new PageResponse<TaskModel>();
  52 + response.setCurrentPage(request.getCurrentPage());
  53 + response.setPageSize(request.getPageSize());
  54 + response.setTotal(total);
  55 + response.setRows(taskModels);
  56 + return new BaseResponse<PageResponse<TaskModel>>(response);
  57 + } catch (Exception e) {
  58 + log.error("getTaskConfigure error", e);
  59 + return new BaseResponse<>(e.getMessage());
  60 + }
  61 +
  62 + }
  63 +
  64 + @RequestMapping("/saveTaskModel")
  65 + @ResponseBody
  66 + public BaseResponse<Integer> saveTaskModel(@RequestBody TaskModel taskModel) {
  67 + try {
  68 + int result = 0;
  69 + if (taskModel.getId() > 0) {
  70 + result = taskSheduleMapper.updateByid(taskModel);
  71 + } else {
  72 + result = taskSheduleMapper.insert(taskModel);
  73 + }
  74 + return new BaseResponse<Integer>(result);
  75 + } catch (Exception e) {
  76 + log.error("saveTaskModel error", e);
  77 + return new BaseResponse<>(e.getMessage());
  78 + }
  79 + }
  80 +
  81 + @RequestMapping("/delTaskModel")
  82 + @ResponseBody
  83 + public BaseResponse<Integer> delTaskModel(int id) {
  84 + try {
  85 + int result = taskSheduleMapper.deleteById(id);
  86 + return new BaseResponse<Integer>(result);
  87 + } catch (Exception e) {
  88 + log.error("delTaskModel error", e);
  89 + return new BaseResponse<>(e.getMessage());
  90 + }
  91 +
  92 + }
  93 +
  94 + @RequestMapping("/getTaskModelById")
  95 + @ResponseBody
  96 + public BaseResponse<TaskModel> getTaskModelById(int id){
  97 + try {
  98 + TaskModel taskModel = taskSheduleMapper.selectById(id);
  99 + return new BaseResponse<TaskModel>(taskModel);
  100 + } catch (Exception e) {
  101 + log.error("getTaskModelById error", e);
  102 + return new BaseResponse<>(e.getMessage());
  103 + }
  104 + }
  105 +
  106 +}
  1 +package com.monitor.other.task;
  2 +
  3 +import org.slf4j.Logger;
  4 +import org.slf4j.LoggerFactory;
  5 +
  6 +import java.util.HashMap;
  7 +import java.util.Map;
  8 +import java.util.concurrent.ConcurrentHashMap;
  9 +import java.util.concurrent.ScheduledFuture;
  10 +
  11 +/**
  12 + * Created by zhengyouwei on 2016/8/17.
  13 + */
  14 +public class TaskStroe {
  15 +
  16 + private static Logger logger = LoggerFactory.getLogger(TaskStroe.class);
  17 +
  18 + private static Map<Integer, ScheduledFuture> map = new ConcurrentHashMap<>();
  19 +
  20 + public static void addTask(Integer id, ScheduledFuture scheduledFuture) {
  21 + try {
  22 + if (map.containsKey(id)) {
  23 + map.get(id).cancel(false);
  24 + }
  25 + map.put(id, scheduledFuture);
  26 + } catch (Exception e) {
  27 + logger.error("addTask failed,id=" + id, e);
  28 + }
  29 + }
  30 +
  31 + public static void removeTask(Integer id) {
  32 + try {
  33 + if (map.containsKey(id)) {
  34 + map.get(id).cancel(false);
  35 + map.remove(id);
  36 + }
  37 + } catch (Exception e) {
  38 + logger.error("removeTask failed,id=" + id, e);
  39 + }
  40 + }
  41 +
  42 + public static boolean isOn(Integer integer){
  43 + return map.containsKey(integer);
  44 + }
  45 +
  46 +}
  1 +package com.monitor.other.task.handler;
  2 +
  3 +import com.model.TaskModel;
  4 +import com.monitor.common.service.AlarmMsgService;
  5 +import com.monitor.other.task.AlarmTypeHandler;
  6 +import org.springframework.beans.factory.annotation.Autowired;
  7 +
  8 +import java.util.Map;
  9 +import java.util.regex.Matcher;
  10 +import java.util.regex.Pattern;
  11 +
  12 +/**
  13 + * Created by zhengyouwei on 2016/8/12.
  14 + */
  15 +public abstract class BaseHandler implements AlarmTypeHandler {
  16 +
  17 + @Autowired
  18 + private AlarmMsgService alarmMsgService;
  19 +
  20 + protected void sendSms(TaskModel taskModel, Map<String, Object> map){
  21 + String newSms = replcaeSms(taskModel.getSms(), map);
  22 + alarmMsgService.sendSms(taskModel.getType(),newSms,taskModel.getMobile());
  23 + }
  24 +
  25 + private String replcaeSms(String sms, Map<String, Object> map) {
  26 + if (map == null){
  27 + return sms;
  28 + }
  29 + Pattern p = Pattern.compile("\\$\\{[a-zA-Z_]+?\\}"); // 正则表达式
  30 + while (true) {
  31 + Matcher m = p.matcher(sms); // 操作的字符串
  32 + if (m.find()) {
  33 + String a = m.group(0);
  34 + a = a.substring(2, a.length() - 1);
  35 + sms = m.replaceFirst(String.valueOf(map.get(a)));
  36 + } else {
  37 + break;
  38 + }
  39 + }
  40 + return sms;
  41 + }
  42 +
  43 +}
  1 +package com.monitor.other.task.handler;
  2 +
  3 +import com.model.TaskModel;
  4 +import com.monitor.influxdb.InfluxDBQuery;
  5 +import com.monitor.other.task.AlarmTypeHandler;
  6 +import org.influxdb.dto.QueryResult;
  7 +import org.springframework.beans.factory.annotation.Autowired;
  8 +import org.springframework.stereotype.Service;
  9 +
  10 +import java.util.HashMap;
  11 +import java.util.List;
  12 +import java.util.Map;
  13 +import java.util.regex.Matcher;
  14 +import java.util.regex.Pattern;
  15 +
  16 +/**
  17 + * Created by zhengyouwei on 2016/8/11.
  18 + * <p>
  19 + * 单个influxdb,只要未出现想要的数据,就告警
  20 + */
  21 +@Service("handlerType_1")
  22 +public class HandlerType_1 extends BaseHandler {
  23 +
  24 + @Override
  25 + public void handle(TaskModel taskModel, Map<String, List<List<Map<String, Object>>>> resultMap) {
  26 + String[] influxdbs = taskModel.getInfluxdbs().split(",");
  27 +
  28 + for (String influx : influxdbs) {
  29 + if (!resultMap.containsKey(influx)) {//没有取到数据,告警
  30 + Map<String, Object> map = new HashMap<>();
  31 + map.put("influxdb", influx);
  32 + sendSms(taskModel, map);
  33 + }
  34 +
  35 + }
  36 +
  37 + }
  38 +
  39 +
  40 +}
  1 +package com.monitor.other.task.handler;
  2 +
  3 +import com.model.TaskModel;
  4 +import org.springframework.stereotype.Service;
  5 +
  6 +import java.util.HashMap;
  7 +import java.util.List;
  8 +import java.util.Map;
  9 +
  10 +/**
  11 + * Created by zhengyouwei on 2016/8/11.
  12 + * <p>
  13 + * 2:多个influxdb 合起来未出现想要的数据
  14 + */
  15 +@Service("handlerType_2")
  16 +public class HandlerType_2 extends BaseHandler {
  17 +
  18 + @Override
  19 + public void handle(TaskModel taskModel, Map<String, List<List<Map<String, Object>>>> resultMap) {
  20 + String[] influxdbs = taskModel.getInfluxdbs().split(",");
  21 +
  22 + int count = 0;
  23 + for (String influx : influxdbs) {
  24 + if (!resultMap.containsKey(influx)) {//没有取到数据,告警
  25 + continue;
  26 + }else {
  27 + count ++;
  28 + }
  29 + }
  30 + if (count == 0){
  31 + sendSms(taskModel, null);
  32 + }
  33 + }
  34 +
  35 +
  36 +}
  1 +package com.monitor.other.task.handler;
  2 +
  3 +import com.model.TaskModel;
  4 +import org.springframework.stereotype.Service;
  5 +
  6 +import java.util.List;
  7 +import java.util.Map;
  8 +
  9 +/**
  10 + * Created by zhengyouwei on 2016/8/11.
  11 + * <p>
  12 + * 3:单个influxdb 出现次数超过一定值
  13 + */
  14 +@Service("handlerType_3")
  15 +public class HandlerType_3 extends BaseHandler {
  16 +
  17 + @Override
  18 + public void handle(TaskModel taskModel, Map<String, List<List<Map<String, Object>>>> resultMap) {
  19 + String[] influxdbs = taskModel.getInfluxdbs().split(",");
  20 +
  21 + for (String influx : influxdbs) {
  22 + if (!resultMap.containsKey(influx)) {//没有取到数据
  23 + continue;
  24 + }else {
  25 + List<List<Map<String, Object>>> lists = resultMap.get(influx);
  26 + for(List<Map<String, Object>> mapList : lists){
  27 + for (Map<String, Object> map : mapList){
  28 + double count = (double) map.get("count");
  29 + if(count > taskModel.getAlarmNumFlag()){
  30 + map.put("influxdb",influx);
  31 + sendSms(taskModel, map);
  32 + }
  33 + }
  34 + }
  35 + }
  36 + }
  37 +
  38 + }
  39 +
  40 +
  41 +}
  1 +package com.monitor.other.task.handler;
  2 +
  3 +import com.model.TaskModel;
  4 +import org.springframework.stereotype.Service;
  5 +
  6 +import java.util.List;
  7 +import java.util.Map;
  8 +
  9 +/**
  10 + * Created by zhengyouwei on 2016/8/11.
  11 + * <p>
  12 + * 出现一个就告警,单条记录单独告警
  13 + */
  14 +@Service("handlerType_4")
  15 +public class HandlerType_4 extends BaseHandler {
  16 +
  17 + @Override
  18 + public void handle(TaskModel taskModel, Map<String, List<List<Map<String, Object>>>> resultMap) {
  19 + String[] influxdbs = taskModel.getInfluxdbs().split(",");
  20 +
  21 + for (String influx : influxdbs) {
  22 + if (!resultMap.containsKey(influx)) {//没有取到数据
  23 + continue;
  24 + }else {
  25 + List<List<Map<String, Object>>> lists = resultMap.get(influx);
  26 + for(List<Map<String, Object>> mapList : lists){
  27 + for (Map<String, Object> map : mapList){
  28 + map.put("influxdb",influx);
  29 + sendSms(taskModel,map);
  30 + }
  31 + }
  32 + }
  33 + }
  34 +
  35 + }
  36 +
  37 +
  38 +}
@@ -8,4 +8,7 @@ @@ -8,4 +8,7 @@
8 <constructor-arg name = "url" value="${dns.config.url}"/> 8 <constructor-arg name = "url" value="${dns.config.url}"/>
9 <constructor-arg name = "dnsApiUrl" value="${dns.config.api.url}"/> 9 <constructor-arg name = "dnsApiUrl" value="${dns.config.api.url}"/>
10 </bean> 10 </bean>
11 -</beans> 11 +
  12 + <bean id="sheduletask" class="com.monitor.other.task.SheduleTask">
  13 + <property name="taskScheduler" ref="scheduler"/>
  14 + </bean></beans>