使用aop+异步队列实现请求日志持久化

前言

有些时候我们在线上出现一些bug, 无法定位到真正的请求数据是什么, 其实单纯的将日志打到控制台并不利于我们排错, 有些时候可能需要完整的请求参数和异常信息

需求: 将请求数据, 响应数据以及异常信息持久化到数据库中, 要求最大限度提高性能, 不要让日志处理影响业务时间

可能有的小伙伴一想到日志处理就会下意识想到aop处理, 但是仅仅通过aop处理的话, 我们会面临两个问题

1. 我们的性能如何保证

2.是否需要每一次请求都去持久化一次

这里我的方案是使用aop+异步队列+批量持久化来实现, 其中使用异步队列的灵感来源于nacos和zookeeper(居多)的架构设计

在zookeeper里大量使用了异步队列来提高性能, 包括nacos的服务注册也使用了异步队列来提高性能, 这些队列的实现既提高了性能, 也避免了线程的资源竞争

类结构

编码阶段, 主要分为这几个部分

1. aop切面编写

2.队列存取统一管理者

3.队列处理worker

完成之后的类结构:

1. aop切面

AdviceProcessor和ExceptionAdviceProcessor: 切面处理

AbstractLogAspect: 切面处理的父类(可以选择日志打印, 具体的处理动作延迟到子类)

EnduranceLogAspect: 持久化日志aop, 包括请求和响应, 并不处理异常

RequestRecordExceptionAdvice: 请求记录异常处理

2.队列存取统一管理者

AbstractRefreshTaskMessageProcessor: 此类囊括了所有的QueueTaskManager(这里用到了策略模式)

SimpleTaskMessageProcessor: 提交任务和拿取任务

其实TaskMessageHandler并不是仅仅做请求日志处理的工作, 他可以做得事情有很多, 只要是需要通过异步队列做得工作他都可以做, 但是前提要写一个QueueTaskManager的实现类, 实现你的队列处理

3.队列处理worker(这个涉及的比较多)

我来介绍下这些类的工作职责

1. TaskMessageHandler 队列处理的顶级接口

2. AbstractQueueTaskWorkerLifecycleController 队列工作者生命周期控制器

3.AbstractBatchTaskWorker批量处理任务工作者(重点)

4.BatchTaskProcessorMetadata 批量任务处理元数据

5.RequestRecordLogTaskWorker请求日志任务工作者(选择性实现, 不固定, 可以不选择持久化, 而是发到mq之类的做日志分析)

6. AbstractSimpleTaskWorker: 任务处理的逻辑我提供了第二种实现, 第一种就是上面的批量实现, 第二种就是这个简单实现, 他在从队列拿到一个数据之后不会等待, 而是马上处理

上代码基础代码

public class RequestRecordContext {

    private static final ThreadLocal tl = new ThreadLocal<>();

    public static RequestRecord get(){
        return tl.get();
    }

    public static void set(RequestRecord requestRecord){
        tl.set(requestRecord);
    }

    public static void remove(){
        tl.remove();
    }
}

public interface Message {
}


public class RequestRecord implements Message {

    
    private Long id;

    private String requestId;

    private String requestParam;

    private String responseParam;

    private String exceptionMsg;

    private Date createTime;
    
    //get and set ....
}

@Component
public class TaskExecutorService implements BeanFactoryAware, InitializingBean {

    private static Executor executor;

    private static BeanFactory beanFactory;

    @Override
    public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
        TaskExecutorService.beanFactory = beanFactory;
    }

    public static void execute(Runnable runnable){
        executor.execute(runnable);
    }

    public static void setExecutor(Executor executor) {
        TaskExecutorService.executor = executor;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        setExecutor(beanFactory.getBean(TaskExecutor.class));
    }
}

基础代码里包括ThreadLocal, 请求记录实体以及线程池

aop切面代码

public interface AdviceProcessor {

    void beforeProcessor(HttpServletRequest request, JoinPoint joinPoint);

    void afterProcessor(Object o);
}


public interface ExceptionAdviceProcessor extends AdviceProcessor{

    void exceptionProcessor(Exception e);

    @Override
    default void afterProcessor(Object o){

    }

    @Override
    default void beforeProcessor(HttpServletRequest request, JoinPoint joinPoint) {

    }
}

@Aspect
public abstract class AbstractLogAspect implements AdviceProcessor {

    private static Logger log = LoggerFactory.getLogger(AdviceProcessor.class);

    @Pointcut("@annotation(com.requestrecord.autoconfiguration.annotation.EnduranceLog)")
    private void logAspect(){}

     //可以不打印, 设置为false即可
    @Value("${endurance.log.request-record.print:false}")
    private boolean print;

    
    @Before("logAspect()")
    public void before(JoinPoint joinPoint){
        try {
            ServletRequestAttributes requestAttributes = (ServletRequestAttributes) RequestContextHolder
                    .getRequestAttributes();
            if (Objects.isNull(requestAttributes)){
                return;
            }
            HttpServletRequest request = requestAttributes.getRequest();
            if (print){
                // 打印请求内容
                log.info("===============请求内容===============");
                log.info("请求地址:" + request.getRequestURL().toString());
                log.info("请求方式:" + request.getMethod());
                log.info("请求类方法:" + joinPoint.getSignature());
                log.info("请求类方法参数:" + JSONObject.toJSONString(joinPoint.getArgs()));
                log.info("===============请求内容===============");
            }
            beforeProcessor(request,joinPoint);
        } catch (Exception e) {
            log.error("LogAspect#before() error >>> msg: "+e.getMessage());
        }
    }


    
    @AfterReturning(returning = "o",pointcut = "logAspect()")
    public void afterReturn(Object o){
        try {
            if (print){
                log.info("--------------返回内容----------------");
                log.info(JSONObject.toJSONString(o));
                log.info("--------------返回内容----------------");
            }
            afterProcessor(o);
        } catch (Exception e) {
            log.error("LogAspect#afterReturn() error >>> msg: "+e.getMessage());
        }
    }

    @AfterThrowing(throwing = "e",pointcut = "logAspect()")
    public void afterThrowing(Exception e){
        if (print){
            log.error("--------------发生异常----------------");
            log.error(e.getMessage());
            log.error("--------------发生异常----------------");
        }
        exceptionProcessor(e);
    }

    protected abstract void exceptionProcessor(Exception e);
}

public class EnduranceLogAspect extends AbstractLogAspect {

    @Autowired
    private TaskMessageProcessor taskMessageProcessor;

    @Override
    public void beforeProcessor(HttpServletRequest request, JoinPoint joinPoint) {
        RequestRecord requestRecord = getOrSetRequestRecord();
        requestRecord.setRequestId(UUID.randomUUID().toString().replace("-",""));
        requestRecord.setRequestParam(JSONObject.toJSONString(joinPoint.getArgs()));
    }

    @Override
    public void afterProcessor(Object o) {
        RequestRecord requestRecord = getOrSetRequestRecord();
        requestRecord.setResponseParam(JSONObject.toJSONString(o));
        //提交持久化任务
        taskMessageProcessor.submit(requestRecord);
    }

    protected RequestRecord getOrSetRequestRecord(){
        RequestRecord requestRecord = RequestRecordContext.get();
        if (requestRecord != null){
            return requestRecord;
        }
        requestRecord = new RequestRecord();
        RequestRecordContext.set(requestRecord);
        return requestRecord;
    }

    @Override
    protected void exceptionProcessor(Exception e) {}


    protected TaskMessageProcessor getTaskMessageProcessor() {
        return taskMessageProcessor;
    }
}

@Component
public class RequestRecordExceptionAdvice extends EnduranceLogAspect implements ExceptionAdviceProcessor {


    @Override
    public void exceptionProcessor(Exception e) {
        RequestRecord requestRecord = RequestRecordContext.get();
        submitRequestRecord(e, requestRecord);
    }

    private void submitRequestRecord(Exception e, RequestRecord requestRecord) {
        getTaskMessageProcessor().submit(requestRecord, m -> {
            StackTraceElement[] stackTrace = e.getStackTrace();
            StringBuilder sb = new StringBuilder(e.getClass().getName());
            sb.append(": ").append(e.getMessage());
            sb.append("n");
            sb.append("tat ");
            for (StackTraceElement s : stackTrace) {
                sb.append(s.getClassName()).append("(").append(s.getFileName()).append(":").append(s.getLineNumber()).append(")");
                sb.append("n");
                sb.append("tat ");
            }
            ((RequestRecord) m).setExceptionMsg(sb.toString());
        });
    }
}

在aop切面里, 进行捕捉请求, 然后提交任务

队列存取统一管理者

public interface TaskMessageProcessor {

    void submit(Message message);

    void submit(Message message, Consumer supplier);

     T take(long waitTime,Class clazz) throws InterruptedException;

     T take(Class clazz) throws InterruptedException;

}

public abstract class AbstractRefreshTaskMessageProcessor implements TaskMessageProcessor,InitializingBean {
    @Autowired
    private List> queueTaskManagerList = new ArrayList<>();

    private Map, QueueTaskManager> queueTaskManagerMap;

    @Override
    public void afterPropertiesSet() throws Exception {
        queueTaskManagerMap = new HashMap<>(
                queueTaskManagerList.size() + (queueTaskManagerList.size() >>> 1));

        for (QueueTaskManager queueTaskManager : queueTaskManagerList) {
            queueTaskManagerMap.put(queueTaskManager.supportType(),queueTaskManager);
        }
    }

    protected QueueTaskManager getQueueTaskManager(Class clazz){
        return queueTaskManagerMap.get(clazz);
    }
}

@Component
public class SimpleTaskMessageProcessor extends AbstractRefreshTaskMessageProcessor {

    @Override
    public void submit(Message message) {
        submit(message,null);
    }

    @SuppressWarnings("unchecked")
    @Override
    public void submit(Message message, Consumer consumer) {
        TaskExecutorService.execute(() -> {
            if (consumer != null){
                consumer.accept(message);
            }
            QueueTaskManager queueTaskManager = getQueueTaskManager(message.getClass());
            if (queueTaskManager != null) {
                queueTaskManager.offer(message);
            }
            // ....
        });
    }

    @SuppressWarnings("unchecked")
    @Override
    public  T take(long waitTime, Class clazz) throws InterruptedException {
        QueueTaskManager queueTaskManager = getQueueTaskManager(clazz);
        return queueTaskManager != null ? (T) queueTaskManager.take(waitTime) : null;
    }

    @SuppressWarnings("unchecked")
    @Override
    public  T take(Class clazz) throws InterruptedException {
        QueueTaskManager queueTaskManager = getQueueTaskManager(clazz);
        return queueTaskManager != null ? (T) queueTaskManager.take() : null;
    }
}

此类囊括了所有的QueueTaskManager实现类, 通过实现QueueTaskManager来做你的队列处理, 这些队列的访问通过TaskMessageProcessor来做

队列处理worker

接口

public interface TaskMessageHandler {

    Class handleType();

    void batchHandle(List t);

    void handle(T t);

}


public interface BatchTaskProcessorMetadata {

    int getMaxProcessCount();


    long getMaxWaitTime();
}

这些接口规定了一些对外方法

具体实现

public abstract class AbstractQueueTaskWorkerLifecycleController
        implements Runnable,BeanNameAware, DisposableBean, SmartLifecycle,TaskMessageHandler {

    protected static Logger log = LoggerFactory.getLogger(TaskMessageHandler.class);

    private AtomicBoolean running = new AtomicBoolean();

    private String beanName;
    @Override
    public void start() {
        if (running.compareAndSet(Boolean.FALSE, Boolean.TRUE)) {
            new Thread(this).start();
            log.info("{} worker thread started ....", beanName);
        }
    }

    @Override
    public boolean isRunning() {
        return running.get();
    }

    @Override
    public void setBeanName(String beanName) {
        this.beanName = beanName;
    }

    @Override
    public void stop() {
        running.compareAndSet(Boolean.TRUE, Boolean.FALSE);
    }

    @Override
    public void destroy() throws Exception {
        stop();
    }

    protected String getBeanName() {
        return beanName;
    }
}

此类主要就是对处理任务的线程做生命周期控制

public abstract class AbstractBatchTaskWorker extends AbstractQueueTaskWorkerLifecycleController
        implements BatchTaskProcessorMetadata, InitializingBean {

    @Value("${endurance.log.request-record.maxProcessCount:500}")
    private int maxProcessCount;

    @Value("${endurance.log.request-record.maxWaitTime:5000}")
    private int maxWaitTime;

    @Autowired
    private TaskMessageProcessor taskMessageProcessor;

    @Override
    public void run() {
        List list = new ArrayList<>();
        long startTime = System.currentTimeMillis();
        while (isRunning()) {
            try {
                //剩下的沉睡时间
                long surplusWaitTime = getMaxWaitTime() - (System.currentTimeMillis() - startTime);
                T t = taskMessageProcessor.take(surplusWaitTime,handleType());
                if (t == null && getMaxWaitTime() - (System.currentTimeMillis() - startTime) > 0L) {
                    continue;
                }
                if (t != null) {
                    list.add(t);
                }
                if ((System.currentTimeMillis() - startTime) < getMaxWaitTime()
                        && list.size() < getMaxProcessCount()) {
                    continue;
                }
                if (!CollectionUtils.isEmpty(list)) {
                    batchHandle(list);
                }
                list.clear();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (Exception e) {
                log.error("{}#run() error, msg: {}", getBeanName(), e.getMessage());
            }
            startTime = System.currentTimeMillis();

        }
        log.warn("{} worker thread shutdown ....", getBeanName());
    }

    @Override
    public int getMaxProcessCount() {
        return maxProcessCount;
    }

    @Override
    public long getMaxWaitTime() {
        return maxWaitTime;
    }

    @Override
    public void handle(T t) {
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        init();
    }

    protected void init() {
        log.info("{} worker init", getBeanName());
    }
}

此类主要做批量处理任务的工作, 当任务数量达到maxProcessCount时或者等待时间达到maxWaitTime时, 会进行处理

//此类为队列处理的第二种实现, 简单处理, 从队列中take到数据后直接调用子类的处理方法, 不会等待
public abstract class AbstractSimpleTaskWorker extends AbstractQueueTaskWorkerLifecycleController {
    private static Logger log = LoggerFactory.getLogger(TaskMessageHandler.class);

    @Autowired
    private TaskMessageProcessor taskMessageProcessor;
    @Override
    public void run() {
        while (isRunning()) {
            try {
                T t = taskMessageProcessor.take(handleType());
                handle(t);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (Exception e) {
                log.error("{}#run() error, msg: {}",getBeanName(), e.getMessage());
            }
        }
        log.warn("{} worker thread shutdown ....", getBeanName());
    }

    @Override
    public void batchHandle(List t) {}
}

此类为队列处理的第二种实现, 简单处理, 从队列中take到数据后直接调用子类的处理方法, 不会等待

@Slf4j
@Component
public class RequestRecordLogTaskWorker extends AbstractBatchTaskWorker {

    @Override
    public Class handleType() {
        //这里返回的是期望处理的数据类型
        return RequestRecord.class;
    }

    @Override
    public void batchHandle(List t) {
        //持久化的代码或者其他处理逻辑......
    }

    //队列
    @Component
    class RequestRecordQueueTaskManager implements QueueTaskManager {

        private ArrayBlockingQueue arrayBlockingQueue = new ArrayBlockingQueue<>(1024 * 1024);

        @Override
        public void offer(Message requestRecord) {
            arrayBlockingQueue.offer((RequestRecord) requestRecord);
        }

        @Override
        public RequestRecord take(long waitTime) throws InterruptedException {
            return arrayBlockingQueue.poll(waitTime, TimeUnit.MILLISECONDS);
        }

        @Override
        public RequestRecord take() throws InterruptedException {
            return arrayBlockingQueue.take();
        }

        @Override
        public Class supportType() {
            return handleType();
        }
    }
}

你需要写一个AbstractBatchTaskWorker或者AbstractSimpleTaskWorker的子类, 然后实现队列访问(基于内存还是分布式), 就可以处理任务

以上就是全部的代码

总结

我们来做一个流程的总结

1. springboot run启动

2. spring ioc在填充完成AbstractRefreshTaskMessageProcessor bean时, 回调他的afterPropertiesSet()方法

在afterPropertiesSet()方法里把所有QueueTaskManager对象(交给spring管理的), 以 handleType()方法返回的类型为key, 对象本身为value, 全部放入到Map中

3. 在spring ioc启动完成后, 由于AbstractQueueTaskWorkerLifecycleController实现了SmartLifecycle接口(isAutoStartup()返回true), 会回调他的start方法, 启动这个任务线程, 启动之后执行run()方法, 在run方法里调用take方法, 此时没有数据, 会陷入沉睡

在ioc关闭时, 如果isRunning()返回true, 会回调stop方法

4. 至此启动完成, 此时有请求, 先进入到aop切面中进行解析, 封装为一个RequestRecord对象, 放入到ThreadLocal中, 在请求结束后或者出现异常, 从ThreadLocal中拿到RequestRecord记录信息, 然后提交到TaskMessageProcessor中

5. TaskMessageProcessor提交任务到线程池, 此任务会根据提交的Message的class为key从Map中拿到一个QueueTaskManager, 然后调用offer()方法, 将数据放入到queue中

6. queue中有数据, 唤醒沉睡的TaskMessageHandler线程, 此时拿到提交的数据, 如果不满, 或者沉睡时间不足, 再次尝试拿取, 第二次拿不到又会陷入沉睡, 如果数量达到了指定的maxProcessCount(子类可重写)则会调用batchHandle方法处理, 具体的逻辑是持久化还是其他逻辑, 就由你定制了.

其实第5步和第6步的逻辑, 比较通用, 并不一定是处理日志, 也可以做其他的逻辑, 只需要写一个AbstractBatchTaskWorker或者AbstractSimpleTaskWorker的子类, 然后实现QueueTaskManager的队列访问逻辑(基于内存还是分布式), 然后把任务提交到TaskMessageProcessor中就可以了

流程图(点击放大)

大家有什么建议欢迎在评论区指出, 谢谢

使用aop+异步队列实现请求日志持久化

.Net相关栏目本月热门文章