当前位置：首页 > news >正文

springboot(40) : 数据断流告警

news 2025/7/16 3:08:12

检测逻辑

package com.alibaba.gts.flm.push.data.client.service;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.gts.flm.push.data.client.common.util.DateUtil;
import com.alibaba.gts.flm.push.data.client.service.model.FcoWarningKeyDTO;
import com.google.common.util.concurrent.RateLimiter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

@Service
@Slf4j
public class FlowCutOffWarningService {


    @Autowired
    private RestTemplateService restTemplateService;

    @Value("${sendDingDingUrl:-}")
    private String sendDingDingApiUrl;

    @Value("${fcowTaskLimit:5}")
    private Integer fcowTaskLimit;

    private ScheduledExecutorService executorService;

    private Map<String, Long> keyTime = new ConcurrentHashMap<>();

    private Map<String, JSONObject> taskStatus = new ConcurrentHashMap<>();

    // uuid
    // 每秒许可数 = 12次 / 60秒
    private Map<String, RateLimiter> taskRateLimiters = new ConcurrentHashMap<>();

    private AtomicInteger threadCount;

    // 启动定时任务
    // limitVal : 每秒许可数 = 12次 / 60秒
    public void startTask(List<FcoWarningKeyDTO> fcoWarningKeyDTOS, Long second, Double limitVal) {
        if (fcoWarningKeyDTOS == null || fcoWarningKeyDTOS.size() == 0 || second == null || limitVal == 0) {
            throw new RuntimeException("参数为空");
        }
        if (executorService == null) {
            executorService = new ScheduledThreadPoolExecutor(fcowTaskLimit);
            threadCount = new AtomicInteger(0);
        }

        if (threadCount.get() >= fcowTaskLimit) {
            throw new RuntimeException("断流检测任务上限5个,需要启动更多任务请修改配置[fcowTaskLimit]的值");
        }
        String uuid = UUID.randomUUID().toString();
        taskRateLimiters.put(uuid, RateLimiter.create(limitVal));
        log.info("启动[{}秒]断流监测任务...uuid:{}", second, uuid);
        executorService.scheduleWithFixedDelay(() -> {
            try {
                //log.info("执行[{}秒]断流监测任务...", second);
                JSONObject status = new JSONObject();
                status.put("fcoWarningKeyDTOS", fcoWarningKeyDTOS);
                status.put("second", second);
                status.put("lastRunTime", DateUtil.now());
                taskStatus.put(uuid, status);
                StringBuilder sb = compare(fcoWarningKeyDTOS, second, keyTime);
                if (sb == null || sb.length() == 0) {
                    return;
                }
                if (!taskRateLimiters.get(uuid).tryAcquire()) {
                    log.warn("发送钉钉过于频繁,本次忽略,uuid:{},limitVal:{}", uuid, limitVal);
                    return;
                }
                sendDingDing(sb);
            } catch (Exception e) {
                log.error("断流监测任务执行出错,fcoWarningKeyDTOS:{},second:{}秒,error:{}", JSONObject.toJSONString(fcoWarningKeyDTOS), second, ExceptionUtils.getStackTrace(e));
            }
        }, second, second, TimeUnit.SECONDS);
        log.info("[{}秒]断流监测任务启动成功,uuid:{}", second, uuid);
        threadCount.set(threadCount.get() + 1);
    }

    public Map<String, JSONObject> getTaskStatus() {
        return taskStatus;
    }

    public Map<String, Long> getKeyTime() {
        return keyTime;
    }

    public Map<String, String> getKeyTimeFormat() {
        Map<String, String> map = new HashMap<>();
        keyTime.forEach((k, v) -> {
            map.put(k, DateUtil.format(v));
        });
        return map;
    }

    public void flow(String k) {
        keyTime.put(k, System.currentTimeMillis());
    }

    public StringBuilder compare(List<FcoWarningKeyDTO> fcoWarningKeyDTOS, Long second, Map<String, Long> keyTime) {
        StringBuilder sb = null;
        for (FcoWarningKeyDTO fcoWarningKeyDTO : fcoWarningKeyDTOS) {
            if (keyTime.containsKey(fcoWarningKeyDTO.getCode()) && (System.currentTimeMillis() - keyTime.get(fcoWarningKeyDTO.getCode())) / 1000 > second) {
                String lastTime = keyTime.containsKey(fcoWarningKeyDTO.getCode()) ? DateUtil.format(keyTime.get(fcoWarningKeyDTO.getCode())) : "-";
                sb = append(sb, fcoWarningKeyDTO, lastTime, second);
            }
        }
        return sb;
    }

    private StringBuilder append(StringBuilder sb, FcoWarningKeyDTO fcoWarningKeyDTO, String lastTime, Long second) {
        if (sb == null) {
            sb = new StringBuilder();
        }
        sb.append("#### " + "[" + fcoWarningKeyDTO.getName() + "]数据断流").append("\n\n");
        sb.append(" > 时间: ").append(DateUtil.now()).append("\n\n");
        sb.append(" > 描述: ").append(second).append("秒内无数据").append("\n\n");
        sb.append(" > 上次数据时间: " + lastTime).append("\n\n");
        return sb;
    }

    private void sendDingDing(StringBuilder sb) {
        JSONObject req = new JSONObject();
        JSONObject at = new JSONObject();
        at.put("isAtAll", "false");
        req.put("title", "数据断流警告");
        req.put("text", sb);
        req.put("at", at);
        restTemplateService.post(sendDingDingApiUrl, req);
    }
}

使用



    @Autowired
    private FlowCutOffWarningService flowCutOffWarningService;

List<FcoWarningKeyDTO> fcoWarningKeyDTOS3Second = new LinkedList<FcoWarningKeyDTO>() {{
            add(new FcoWarningKeyDTO("test", "测试"));
        }};
        flowCutOffWarningService.startTask(fcoWarningKeyDTOS3Second, 3L, 0.003);