refactor(logging): use reactive Sinks for asynchronous log processin.

Author: ReLive27

audit/src/main/java/io/github/simpleauth0/audit/AuditLogMethodInterceptor.java

@@ -1,7 +1,7 @@
 package io.github.simpleauth0.audit;
 
 import io.github.simpleauth0.audit.annotation.AuditLog;
-import io.github.simpleauth0.audit.expression.ExpressionAttribute;
+import io.github.simpleauth0.audit.expression.AuditLogExpressionAttribute;
 import io.github.simpleauth0.audit.handler.AuditLogStorageHandler;
 import io.github.simpleauth0.core.utils.AnnotationUtils;
 import org.aopalliance.aop.Advice;
@@ -36,7 +36,7 @@ public class AuditLogMethodInterceptor implements Ordered, MethodInterceptor, Po
 
     private AuditLogStorageHandler auditLogStorageHandler;
 
-    public AuditLogMethodInterceptor(AuditLogStorageHandler<? extends ExpressionAttribute> auditLogStorageHandler) {
+    public AuditLogMethodInterceptor(AuditLogStorageHandler<? extends AuditLogExpressionAttribute> auditLogStorageHandler) {
         Assert.notNull(auditLogStorageHandler, "auditLogStorageHandler cannot be null");
         this.auditLogStorageHandler = auditLogStorageHandler;
         this.pointcut = AuditLogMethodPointcuts.forAnnotations(AuditLog.class);
@@ -58,7 +58,7 @@ public Object invoke(MethodInvocation mi) throws Throwable {
             throw e;
         } finally {
             AuditLogExpressionAttribute attribute = this.registry.getAttribute(mi);
-            if (attribute != ExpressionAttribute.NULL_ATTRIBUTE) {
+            if (attribute != AuditLogExpressionAttribute.NULL_ATTRIBUTE) {
                 String[] parameterNames = discoverer.getParameterNames(mi.getMethod());
                 if (parameterNames != null && parameterNames.length > 0) {
                     attribute.setParams(parameterNames);
@@ -130,41 +130,6 @@ private AuditLog findPostFilterAnnotation(Method method) {
 
     }
 
-    public static final class AuditLogExpressionAttribute extends ExpressionAttribute {
-
-        private static final AuditLogExpressionAttribute NULL_ATTRIBUTE = new AuditLogExpressionAttribute(null, null);
-
-        private final AuditLog auditLog;
-
-        private Object result;
-
-        private Boolean success;
-
-        private AuditLogExpressionAttribute(Expression expression, AuditLog auditLog) {
-            super(expression);
-            this.auditLog = auditLog;
-        }
-
-        public void setResult(Object result) {
-            this.result = result;
-        }
-
-        public void setSuccess(Boolean success) {
-            this.success = success;
-        }
-
-        public Object getResult() {
-            return result;
-        }
-
-        public Boolean getSuccess() {
-            return success;
-        }
-
-        public AuditLog getAuditLog() {
-            return auditLog;
-        }
-    }
 }
 
 

audit/src/main/java/io/github/simpleauth0/audit/expression/AuditLogExpressionAttribute.java

@@ -0,0 +1,60 @@
+package io.github.simpleauth0.audit.expression;
+
+import io.github.simpleauth0.audit.annotation.AuditLog;
+import lombok.Data;
+import org.springframework.expression.Expression;
+
+/**
+ * @author: ReLive27
+ * @date: 2025/4/28 20:53
+ */
+@Data
+public class AuditLogExpressionAttribute {
+
+    public static final AuditLogExpressionAttribute NULL_ATTRIBUTE = new AuditLogExpressionAttribute(null, null);
+
+    private final Expression expression;
+
+    private Object[] args;
+
+    private String[] params;
+
+    private final AuditLog auditLog;
+
+    private Object result;
+
+    private Boolean success;
+
+    public AuditLogExpressionAttribute(Expression expression, AuditLog auditLog) {
+        this.expression = expression;
+        this.auditLog = auditLog;
+    }
+
+    public void setResult(Object result) {
+        this.result = result;
+    }
+
+    public void setSuccess(Boolean success) {
+        this.success = success;
+    }
+
+    public Object getResult() {
+        return result;
+    }
+
+    public Boolean getSuccess() {
+        return success;
+    }
+
+    public AuditLog getAuditLog() {
+        return auditLog;
+    }
+
+    public void setArgs(Object[] args) {
+        this.args = args;
+    }
+
+    public void setParams(String[] params) {
+        this.params = params;
+    }
+}

audit/src/main/java/io/github/simpleauth0/audit/expression/ExpressionAttribute.java

@@ -1,12 +1,13 @@
 package io.github.simpleauth0.audit.handler;
 
-import io.github.simpleauth0.audit.expression.ExpressionAttribute;
+
+import io.github.simpleauth0.audit.expression.AuditLogExpressionAttribute;
 
 /**
  * @author: ReLive27
  * @date: 2025/5/23 22:06
  */
-public interface AuditLogStorageHandler<T extends ExpressionAttribute> {
+public interface AuditLogStorageHandler<T extends AuditLogExpressionAttribute> {
 
     void handler(T expressionAttribute);
 }

audit/src/main/java/io/github/simpleauth0/audit/handler/AuditLogStorageHandler.java

@@ -2,38 +2,54 @@
 
 import eu.bitwalker.useragentutils.OperatingSystem;
 import eu.bitwalker.useragentutils.UserAgent;
-import io.github.simpleauth0.audit.AuditLogMethodInterceptor;
 import io.github.simpleauth0.audit.annotation.AuditLog;
+import io.github.simpleauth0.audit.expression.AuditLogExpressionAttribute;
 import io.github.simpleauth0.audit.repository.AuditLogEvent;
 import io.github.simpleauth0.audit.repository.AuditLogEventRepository;
 import io.github.simpleauth0.audit.repository.Field;
 import io.github.simpleauth0.core.utils.IPUtils;
 import io.github.simpleauth0.core.utils.JsonUtils;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.expression.EvaluationContext;
 import org.springframework.expression.spel.support.StandardEvaluationContext;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 import org.springframework.web.context.request.RequestContextHolder;
 import org.springframework.web.context.request.ServletRequestAttributes;
+import reactor.core.publisher.Sinks;
 
 import javax.servlet.http.HttpServletRequest;
+import java.time.Duration;
 import java.util.*;
 
 /**
  * @author: ReLive27
  * @date: 2025/5/24 21:42
  */
-public class DefaultAuditLogStorageHandler implements AuditLogStorageHandler<AuditLogMethodInterceptor.AuditLogExpressionAttribute> {
+@Slf4j
+public class DefaultAuditLogStorageHandler implements AuditLogStorageHandler<AuditLogExpressionAttribute> {
+
+    private static final Sinks.Many<AuditLogEvent> sink = Sinks.many()
+            .unicast()
+            .onBackpressureBuffer();
 
     private final AuditLogEventRepository auditLogRepository;
 
     public DefaultAuditLogStorageHandler(AuditLogEventRepository auditLogRepository) {
         Assert.notNull(auditLogRepository, "auditLogRepository can not be null");
         this.auditLogRepository = auditLogRepository;
+        // 异步订阅处理，批量保存
+        sink.asFlux()
+                .bufferTimeout(10, Duration.ofSeconds(3))
+                .flatMap(this.auditLogRepository::save) // 异步保存
+                .doOnNext(batch -> log.debug("已保存日志数量：" + batch))
+                .doOnError(e -> log.error("日志流异常：" + e.getMessage()))
+                .onErrorContinue((e, o) -> log.error("单条日志保存失败: " + o))
+                .subscribe();
     }
 
     @Override
-    public void handler(AuditLogMethodInterceptor.AuditLogExpressionAttribute expressionAttribute) {
+    public void handler(AuditLogExpressionAttribute expressionAttribute) {
         AuditLogEvent auditEvent = new AuditLogEvent();
         AuditLog auditLog = expressionAttribute.getAuditLog();
         auditEvent.setEventType(auditLog.resourceType());
@@ -70,6 +86,21 @@ public void handler(AuditLogMethodInterceptor.AuditLogExpressionAttribute expres
         fieldList.add(new Field("request_param", JsonUtils.toJson(requestParams), null));
         fieldList.add(new Field("response_body", expressionAttribute.getResult() != null ? JsonUtils.toJson(expressionAttribute.getResult()) : "", null));
         auditEvent.setFields(fieldList);
-        this.auditLogRepository.save(auditEvent);
+
+        save(auditEvent);
+    }
+
+    public void save(AuditLogEvent logEvent) {
+        sink.emitNext(logEvent, (signalType, emitResult) -> {
+            if (emitResult == Sinks.EmitResult.FAIL_NON_SERIALIZED) {
+                // Reactor 自身会重试；返回 true 表示可重试
+                return true;
+            }
+            if (emitResult.isFailure()) {
+                log.warn("日志发送失败: " + emitResult);
+            }
+            // 返回 false 表示不重试
+            return false;
+        });
     }
 }

audit/src/main/java/io/github/simpleauth0/audit/handler/DefaultAuditLogStorageHandler.java

@@ -1,13 +1,15 @@
 package io.github.simpleauth0.audit.repository;
 
-import reactor.core.publisher.Mono;
+import reactor.core.publisher.Flux;
+
+import java.util.List;
 
 /**
  * @author: ReLive27
  * @date: 2025/6/30 22:56
  */
 public interface AuditLogEventRepository {
 
-    Mono<Void> save(AuditLogEvent event);
+    Flux<Integer> save(List<AuditLogEvent> auditLogEvents);
 
 }