如何防止系统发生异常时，别人传递过来的关键数据不丢失？（AOP + xxlJob）

需求

在开发中，客户每天需要定时调用我们的api去上传一些数据到数据库中，当数据库发生异常或者系统发生异常，上传的一些数据会丢失不做入库操作，现想防止数据库或系统发生异常，数据能不丢失，同时，等系统恢复时，能重新入库。

思路

一开始，我想到的是当系统异常时，备份那些sql语句到文件里，然后定时地执行这些语句，就是上一篇《MybatisPLus输出sql语句到指定文件（附带完整的参数）》，但是实现完去调测后，发现不对劲，异常时根本不走mybatis拦截器，于是，这种备份sql语句的方案行不通，我采取了另一种方案AOP获取调用方法时传入的请求体

（1）首先，创建一个注解，然后创建一个切面类将这个注解定义为一个切点，在controller层需要拦截的方法加上这个注解即可；
（2）在切面类中定义@AfterThrowing异常抛出后处理的方法，用于获取请求体并追加到文件中；
（3）使用xxl-job定时去执行文件中的内容

步骤

1、在采集程序模块中，利用AOP进行异常处理，以自定义 @LogPrint 注解为切点，当发生异常时捕获请求的相关内容封装成json字符串，然后调用备份的微服务（aomp-data-capture-backup）将其追加到文件中

注解

import java.lang.annotation.*;

@Retention(RetentionPolicy.RUNTIME)//注解不仅被保存到class文件中，jvm加载class文件之后，仍存在
@Target(ElementType.METHOD) //注解添加的位置
@Documented
public @interface LogPrint {
    String description() default "";

}

利用AOP进行异常处理

import com.alibaba.fastjson.JSONObject;
import com.cspg.dataworks.MainApplication;
import com.cspg.dataworks.api.SqlBackUpApi;
import com.fasterxml.jackson.core.JsonProcessingException;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.JoinPoint;
import org.aspectj.lang.annotation.AfterThrowing;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Pointcut;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import org.springframework.web.context.request.RequestContextHolder;
import org.springframework.web.context.request.ServletRequestAttributes;
import org.springframework.web.multipart.MultipartFile;

import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.BufferedReader;
import java.io.IOException;

@Aspect
@Component
@Slf4j
public class WebExceptionAspect {

    @Autowired
    SqlBackUpApi sqlBackUpApi;

    /** 以自定义 @LogPrint 注解为切点 */
    @Pointcut("@annotation(com.cspg.dataworks.exception.LogPrint)")
    public void logPrint() {}

    @AfterThrowing(pointcut = "logPrint()")
    //controller类抛出的异常在这边捕获
    public void handleThrowing(JoinPoint joinPoint) throws JsonProcessingException {
        // 开始打印请求日志
        ServletRequestAttributes attributes = (ServletRequestAttributes) RequestContextHolder.getRequestAttributes();
        HttpServletRequest request = attributes.getRequest();
        String requestArgs = getJsonRequest(request);
        log.info("捕获异常Start");

        String applicatonName = MainApplication.applicatonName;

        String url = "http://" + applicatonName + request.getRequestURI();

        JSONObject content = new JSONObject();
        content.put("url", url);
        content.put("method", request.getMethod());

        if (!StringUtils.isEmpty(request.getHeader("accessToken"))){
            content.put("accessToken", "aomp-data-backup");
        }

        if (!StringUtils.isEmpty(requestArgs)){
            //请求体不为空，为post请求时
            content.put("requestBody", requestArgs);
            content.put("requestParam","");
        }else{
            //请求体为空而请求参数不为空时，为get请求时
            content.put("requestBody","");
            content.put("requestParam", getParams(joinPoint));
        }

        String contentStr = JSONObject.toJSONString(content) + ";";

        JSONObject result = new JSONObject();
        result.put("content", contentStr);
        result.put("filePath", "/data/home/backup" + applicatonName + ".txt");
        sqlBackUpApi.pushSqlToFile(result);
    }

    private String getParams(JoinPoint joinPoint) {
        String params = "";
        if (joinPoint.getArgs() != null && joinPoint.getArgs().length > 0) {
            for (int i = 0; i < joinPoint.getArgs().length; i++) {
                Object arg = joinPoint.getArgs()[i];
                if ((arg instanceof HttpServletResponse) || (arg instanceof HttpServletRequest)
                        || (arg instanceof MultipartFile) || (arg instanceof MultipartFile[])) {
                    continue;
                }
                try {
                    params += JSONObject.toJSONString(joinPoint.getArgs()[i]);
                } catch (Exception e1) {
                    log.error(e1.getMessage());
                }
            }
        }
        return params;
    }

    private String getJsonRequest(HttpServletRequest request) {
        org.json.JSONObject result = null;
        StringBuilder sb = new StringBuilder();
        try (BufferedReader reader = request.getReader();) {
            char[] buff = new char[1024];
            int len;
            while ((len = reader.read(buff)) != -1) {
                sb.append(buff, 0, len);
            }
            log.info("request中参数为：{}", sb.toString());
            String s = sb.toString();
            return s;
        } catch (IOException e) {
            log.error("", e);
        }
        return "";
    }

}

文件里主要存这些参数
在这里插入图片描述

SqlBackUpApi

import com.alibaba.fastjson.JSONObject;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;

@FeignClient(value = "aomp-data-backup",path = "/api/dataBackUp")
public interface SqlBackUpApi {

    @PostMapping("/pushSqlToFile")
    void pushSqlToFile(@RequestBody JSONObject jsonObject);
}

在需要捕获异常的方法上添加@LogPrint 注解

    @PostMapping("/battery/upload")
    @ApiOperation("2.2.上报充电过程电池信息数据")
    @LogPrint
    public Map<Object,Object> saveBattery(@RequestBody BatteryStatus batteryStatus) {
        Map<Object, Object> map = checkToken();
        if (!CollectionUtils.isEmpty(map)) {
            return map;
        }
        batteryStatusService.saveBatteryStatus(batteryStatus);
        String remoteHost = ReqUtils.getRequestIP(request);
        log.info("{} 上传充电过程电池信息数据成功",remoteHost);
        return Result.msg(Result.SUCCESS_CODE);
    }

2、创建一个名为aomp-data-capture-backup的微服务，其向外提供一个api用于异常时将未执行的请求内容封装成json字符串插入到文件中，另外，结合了xxl-job分布式任务调度框架，可定时读取文件中的请求内容，然后使用restTemplate重新请求远程地址去执行入库，执行成不成功都要删除对应的的json字符串，因为不成功aop会继续追加json字符串到文件中；

（1）向外提供用于将未执行的请求内容插入到文件的api

    @SneakyThrows
    @PostMapping("/pushSqlToFile")
    @ApiOperation("将未执行的请求内容插入到文件中")
    public void pushSqlToFile(@RequestBody FileRequest fileRequest){
        FileUtils.insertSqlToFile(fileRequest);
    }

（2）定时读取文件中的请求内容

@Component
@Slf4j
public class DataBackUpXxlJob {

    @Autowired
    private RestTemplate restTemplate;

    @XxlJob("demoJobHandler")
    public void demoJobHandler() throws Exception {
        System.out.println("执行定时任务,执行时间:" + new Date());
    }

    @XxlJob("dataworksHandler")
    public void dataworksHandler() throws Exception {
        String fileName = "aomp-data-capture-dataworks.txt";
        // 读取文件内容拆分每条JSON字符串
        JSONArray jsonStrs = FileUtils.readJsonStr("/data/home/backup" + fileName);
        dataHandler(jsonStrs, fileName);
    }

    @XxlJob("photovoltHandler")
    public void photovoltHandler() throws Exception {
        String fileName = "aomp-data-capture-photovolt.txt";
        // 读取文件内容拆分每条JSON字符串
        JSONArray jsonStrs = FileUtils.readJsonStr("/data/home/backup" + fileName);
        dataHandler(jsonStrs, fileName);
    }

    @XxlJob("parrotHandler")
    public void parrotHandler() throws Exception {
        String fileName = "aomp-data-capture-parrot.txt";
        // 读取文件内容拆分每条JSON字符串
        JSONArray jsonStrs = FileUtils.readJsonStr("/data/home/backup" + fileName);
        dataHandler(jsonStrs, fileName);
    }

    @XxlJob("dataHandler")
    public void dataHandler(JSONArray jsonStrs, String fileName){
        if (jsonStrs != null && jsonStrs.size() > 0) {
            for (int i = 0; i < jsonStrs.size(); i++) {
                JSONObject jsonStr = jsonStrs.getJSONObject(i);
                String contentStr = jsonStr.getStr("content");
                JSONObject content = JSONUtil.parseObj(contentStr);

                // 获取要调用远程地址时要传递的参数
                String requestBody = content.getStr("requestBody");
                String requestParam = content.getStr("requestParam");
                String accessToken = content.getStr("accessToken");
                String requestBodyUrl = content.getStr("url");
                String method = content.getStr("method");

                try {
                    log.info("开始请求第三方传递参数重新执行方法");
                    ResponseEntity<String> response = null;
                    if (!StringUtils.isEmpty(requestBody)) {
                        // post请求，只有请求体有数据时
                        HttpHeaders headers = new HttpHeaders();
                        headers.setContentType(MediaType.APPLICATION_JSON);
                        if (!StringUtils.isEmpty(accessToken)){
                            headers.add("accessToken", accessToken);
                        }
                        HttpEntity<String> httpEntity = new HttpEntity<>(requestBody, headers);
                        response = restTemplate.exchange(requestBodyUrl, HttpMethod.resolve(method), httpEntity, String.class);

                    } else if (!StringUtils.isEmpty(requestParam)) {
                        // get请求，只有请求参数有数据时
                        JSONObject params = JSONUtil.parseObj(requestParam);
                        StringBuilder requestParamUrl = new StringBuilder();
                        requestParamUrl.append(requestBodyUrl).append("?");
                        int index = 0;
                        for (JSONObject.Entry<String, Object> entry : params) {
                            if (index == 0) {
                                requestParamUrl.append(entry.getKey()).append("=").append(entry.getValue());
                            } else {
                                requestParamUrl.append("&").append(entry.getKey()).append("=").append(entry.getValue());
                            }
                            index++;
                        }

                        HttpHeaders headers = new HttpHeaders();
                        if (!StringUtils.isEmpty(accessToken)){
                            headers.add("accessToken", accessToken);
                        }
                        HttpEntity<String> httpEntity  = new HttpEntity<>(headers);
                        response = restTemplate.exchange(requestParamUrl.toString() , HttpMethod.resolve(method), httpEntity, String.class);
                    }

                    String body = response.getBody();
                    log.info("请求第三方后响应的结果： " + body);
                    Integer endIndex = jsonStr.getInt("endIndex");
                    // 删除对应的json字符串
                    FileUtils.removeJsonStrFromFile("/data/home/backup" + fileName, endIndex);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

文件工具类

public class FileUtils {

    // 将请求内容保存到文件中
    @SneakyThrows
    public static void insertSqlToFile(FileRequest fileRequest) {
        // 创建文件
        File file = new File(fileRequest.getFilePath());
        if (!file.getParentFile().exists()) {
            file.getParentFile().mkdirs();
        }
        if (!file.exists()) {
            file.createNewFile();
        }

        // 写入文件
        FileWriter fw = new FileWriter(fileRequest.getFilePath(), true);
        BufferedWriter bw = new BufferedWriter(fw);
        bw.write(fileRequest.getContent());
        bw.newLine();
        bw.close();
        fw.close();
    }

    //读取文件内容拆分每条JSON字符串
    public static JSONArray readJsonStr(String fileName) throws Exception {
        JSONArray statements = JSONUtil.createArray();
        File file = new File(fileName);
        if (file.exists()) {
            BufferedReader reader = new BufferedReader(new FileReader(fileName));
            try {
                String line = reader.readLine();
                StringBuilder sb = new StringBuilder();
                while (line != null) {
                    sb.append(line);
                    //完整的一条JSON字符串
                    String singleSql = sb.toString();
                    // 语句以分号结尾
                    if (line.endsWith(";")) {
                        int endIndex = singleSql.lastIndexOf(singleSql.charAt(singleSql.length() -1));
                        JSONObject statement = JSONUtil.createObj();
                        statement.put("content", singleSql);
                        //最后一个字符的索引
                        statement.put("endIndex", endIndex);
                        statements.add(statement);
                        sb.setLength(0);
                    }
                    line = reader.readLine();
                }
            } catch (Exception e) {
                // 处理异常
                e.printStackTrace();
            } finally {
                if (reader != null) {
                    reader.close();
                }
            }
        }
        return statements;
    }

    public static void removeJsonStrFromFile(String fileName, int endIndex) throws Exception {
        // 创建输入流
        BufferedReader reader = new BufferedReader(new FileReader(fileName));

        // 读取文件内容
        StringBuilder content = new StringBuilder();
        String line;
        while ((line = reader.readLine()) != null) {
            if (!line.trim().isEmpty()){
                content.append(line).append(System.lineSeparator());
            }
        }

        // 删除指定索引的字符串
        String modifiedContent = content.substring(endIndex + 1);

        // 创建输出流
        BufferedWriter writer = new BufferedWriter(new FileWriter(fileName));

        // 关闭输入流
        reader.close();

        // 将修改后的内容写回文件中
        writer.write(modifiedContent);

        // 关闭输出流
        writer.close();
    }

}

aomp-data-capture-backup的xxl-job配置

### 调度中心部署根地址 [选填]：如调度中心集群部署存在多个地址则用逗号分隔。执行器将会使用该地址进行"执行器心跳注册"和"任务结果回调"；为空则关闭自动注册；
xxl:
  job:
    admin:
      addresses: http://127.0.0.1:30316/xxl-job-admin
### 执行器通讯TOKEN [选填]：非空时启用；
    accessToken: default_token
### 执行器AppName [选填]：执行器心跳注册分组依据；为空则关闭自动注册
    executor:
      appname: xxl-job-sqlBackUp-executor
### 执行器IP [选填]：默认为空表示自动获取IP，多网卡时可手动设置指定IP，该IP不会绑定Host仅作为通讯实用；地址信息用于 "执行器注册" 和 "调度中心请求并触发任务"；
      ip: 127.0.0.1
### 执行器端口号 [选填]：小于等于0则自动获取；默认端口为9999，单机部署多个执行器时，注意要配置不同执行器端口；
      port: 9999
### 执行器运行日志文件存储磁盘路径 [选填] ：需要对该路径拥有读写权限；为空则使用默认路径；
      logpath: /data/applogs/xxl-job/jobhandler
### 执行器日志文件保存天数 [选填] ： 过期日志自动清理, 限制值大于等于3时生效; 否则, 如-1, 关闭自动清理功能；
      logretentiondays: 30

3、xxl-job项目就不多说了，用开源的代码后配置信息改一改

### actuator
management:
  server:
    servlet:
      context-path: /actuator
  health:
    mail:
      enabled: false
### mybatis
mybatis:
  mapper-locations: classpath:/mybatis-mapper/*Mapper.xml
spring:
  datasource:
    ### datasource-pool
    type: com.zaxxer.hikari.HikariDataSource
    hikari:
      auto-commit: true
      connection-test-query: SELECT 1
      connection-timeout: 10000
      idle-timeout: 30000
      max-lifetime: 900000
      maximum-pool-size: 30
      minimum-idle: 10
      pool-name: HikariCP
      validation-timeout: 1000
    ### xxl-job, datasource
    driver-class-name: com.mysql.cj.jdbc.Driver
    url: jdbc:mysql://127.0.0.1:3306/cspg_aomp_db_ms_new?useUnicode=true&characterEncoding=UTF-8&autoReconnect=true&serverTimezone=Asia/Shanghai
    username: root
    password: ning
  ### freemarker
  freemarker:
    charset: UTF-8
    request-context-attribute: request
    settings:
      number_format: 0.##########
    suffix: .ftl
    templateLoaderPath: classpath:/templates/
  ### xxl-job, email
  mail:
    from: xxx@qq.com
    host: smtp.qq.com
    password: xxx
    port: 25
    properties:
      mail:
        smtp:
          auth: true
          socketFactory:
            class: javax.net.ssl.SSLSocketFactory
          starttls:
            enable: true
            required: true
    username: xxx@qq.com
  ### resources
  mvc:
    servlet:
      load-on-startup: 0
    static-path-pattern: /static/**
  resources:
    static-locations: classpath:/static/

xxl:
  job:
    ### xxl-job, access token
    accessToken: default_token
    ### xxl-job, i18n (default is zh_CN, and you can choose "zh_CN", "zh_TC" and "en")
    i18n: zh_CN
    ### xxl-job, log retention days
    logretentiondays: 30
    ## xxl-job, triggerpool max size
    triggerpool:
      fast:
        max: 200
      slow:
        max: 100

最后注意一下：这些项目都需注册到nacos，日后方便远程调用