更新 content-service

1.利用 logback 的 mdc 自动在日志中添加 requestId 字段
2.利用 TaskDecorator 接口和 mdc 实现在异步线程的日志中添加 requestId 字段
3.在 logback-spring.xml 配置中通过 %X{request_id} 来引用 mdc 中的 requestId 字段
4.添加 RabbitAppender, 使用 mq 异步存储 runtime 日志

content/content-service/src/main/java/cn/reghao/tnb/content/app/config/MdcExecutor.java

@@ -0,0 +1,32 @@
+package cn.reghao.tnb.content.app.config;
+
+import org.slf4j.MDC;
+
+import java.util.concurrent.Executor;
+
+/**
+ * 解决线程异步场景下 RequestId 的在 logback 中的打印问题
+ *
+ * @author reghao
+ * @date 2025-07-17 16:57:41
+ */
+public class MdcExecutor implements Executor {
+    private final String key = "request_id";
+    private final Executor executor;
+
+    public MdcExecutor(Executor executor) {
+        this.executor = executor;
+    }
+
+    public void execute(Runnable command) {
+        final String requestId = MDC.get(key);
+        executor.execute(() -> {
+            MDC.put(key, requestId);
+            try {
+                command.run();
+            } finally {
+                MDC.remove(key);
+            }
+        });
+    }
+}

content/content-service/src/main/java/cn/reghao/tnb/content/app/config/MdcTaskDecorator.java

@@ -0,0 +1,32 @@
+package cn.reghao.tnb.content.app.config;
+
+import org.slf4j.MDC;
+import org.springframework.core.task.TaskDecorator;
+
+import java.util.Map;
+
+/**
+ * 解决线程异步场景下 RequestId 的在 logback 中的打印问题
+ *
+ * @author reghao
+ * @date 2025-07-18 09:18:16
+ */
+public class MdcTaskDecorator implements TaskDecorator {
+    @Override
+    public Runnable decorate(Runnable runnable) {
+        // Capture the context of the current thread (the main thread)
+        Map<String, String> contextMap = MDC.getCopyOfContextMap();
+        return () -> {
+            try {
+                // Set the context map for the child thread
+                if (contextMap != null) {
+                    MDC.setContextMap(contextMap);
+                }
+                runnable.run();
+            } finally {
+                // Clear the context map after the child thread has finished
+                MDC.clear();
+            }
+        };
+    }
+}

content/content-service/src/main/java/cn/reghao/tnb/content/app/config/SpringLifecycle.java

@@ -1,15 +1,11 @@
 package cn.reghao.tnb.content.app.config;
 
-import cn.reghao.jutil.jdk.machine.id.MachineId;
-import cn.reghao.jutil.jdk.machine.id.MachineIdLinux;
-import cn.reghao.tnb.common.prop.LogProperties;
-import cn.reghao.tnb.log.Appenders;
+import cn.reghao.tnb.content.app.util.rabbit.RabbitAppender;
 import cn.reghao.tnb.log.LoggerConfig;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.boot.ApplicationArguments;
 import org.springframework.boot.ApplicationRunner;
-import org.springframework.core.env.Environment;
 import org.springframework.stereotype.Component;
 
 import java.util.List;
@@ -21,33 +17,21 @@ import java.util.List;
 @Slf4j
 @Component
 public class SpringLifecycle implements ApplicationRunner, DisposableBean {
-    private final Environment env;
-    private final MachineId machineId;
-    private final LogProperties logProperties;
+    private final RabbitAppender rabbitAppender;
 
-    public SpringLifecycle(Environment env, LogProperties logProperties) {
-        this.env = env;
-        this.machineId = new MachineIdLinux();
-        this.logProperties = logProperties;
+    public SpringLifecycle(RabbitAppender rabbitAppender) {
+        this.rabbitAppender = rabbitAppender;
     }
 
     @Override
     public void run(ApplicationArguments args) {
-        if (logProperties.getEnabled()) {
-            initLogConfig();
-        }
+        //initLogConfig(rabbitAppender);
         log.info("ContentService 启动...");
     }
 
-    private void initLogConfig() {
-        String wsUrl = logProperties.getWsUrl();
-        if (wsUrl != null) {
-            String token = logProperties.getToken();
-            String app = env.getProperty("spring.application.name");
-            String host = machineId.ipv4();
-            String url = String.format("%s?token=%s&app=%s&host=%s", wsUrl, token, app, host);
-            LoggerConfig.initLogger(List.of(Appenders.wsAppender(url, app, host)));
-        }
+    private void initLogConfig(RabbitAppender rabbitAppender) {
+        log.info("ContentService use RabbitAppender...");
+        LoggerConfig.initLogger(List.of(rabbitAppender));
     }
 
     @Override

content/content-service/src/main/java/cn/reghao/tnb/content/app/config/ThreadPoolConfig.java

@@ -0,0 +1,42 @@
+package cn.reghao.tnb.content.app.config;
+
+import cn.reghao.jutil.jdk.machine.id.MachineId;
+import cn.reghao.jutil.jdk.machine.id.MachineIdLinux;
+import cn.reghao.tnb.content.app.util.rabbit.RabbitAppender;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.env.Environment;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+
+import java.util.concurrent.Executor;
+
+/**
+ * @author reghao
+ * @date 2025-07-18 09:17:28
+ */
+@Configuration
+public class ThreadPoolConfig {
+    @Bean(name = "taskExecutor")
+    public Executor taskExecutor() {
+        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+        executor.setCorePoolSize(10);
+        executor.setMaxPoolSize(20);
+        executor.setQueueCapacity(10_000);
+        //executor.setRejectedExecutionHandler();
+        executor.setThreadNamePrefix("MyExecutor-");
+
+        // Set the TaskDecorator
+        executor.setTaskDecorator(new MdcTaskDecorator());
+        executor.initialize();
+        return executor;
+    }
+
+    @Bean
+    public RabbitAppender rabbitAppender(Environment env, RabbitTemplate rabbitTemplate) {
+        MachineId machineId = new MachineIdLinux();
+        String app = env.getProperty("spring.application.name");
+        String host = machineId.ipv4();
+        return new RabbitAppender(app, host, rabbitTemplate);
+    }
+}

content/content-service/src/main/java/cn/reghao/tnb/content/app/config/web/TokenFilter.java → content/content-service/src/main/java/cn/reghao/tnb/content/app/config/web/AccessFilter.java

@@ -3,17 +3,20 @@ package cn.reghao.tnb.content.app.config.web;
 import cn.reghao.jutil.web.ServletUtil;
 import cn.reghao.tnb.common.auth.LoginUser;
 import cn.reghao.tnb.common.auth.UserContext;
+import org.slf4j.MDC;
 import org.springframework.stereotype.Component;
 
 import javax.servlet.*;
 import java.io.IOException;
 
 /**
+ * HTTP 请求过滤器
+ *
  * @author reghao
- * @date 2023-08-25 21:14:23
+ * @date 2025-07-18 09:18:16
  */
 @Component
-public class TokenFilter implements Filter {
+public class AccessFilter implements Filter {
     @Override
     public void init(FilterConfig filterConfig) throws ServletException {
     }
@@ -21,6 +24,9 @@ public class TokenFilter implements Filter {
     @Override
     public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain)
             throws IOException, ServletException {
+        String requestId = ServletUtil.getHeader("x-request-id");
+        MDC.put("request_id", requestId);
+
         long userId = -1L;
         String userIdStr = ServletUtil.getHeader("x-user-id");
         if (userIdStr != null) {
@@ -31,6 +37,8 @@ public class TokenFilter implements Filter {
         LoginUser loginUser = new LoginUser(userId);
         try (UserContext context = new UserContext(loginUser)) {
             chain.doFilter(request, response);
+        } finally {
+            MDC.clear();
         }
     }
 

content/content-service/src/main/java/cn/reghao/tnb/content/app/config/web/AccessLogInterceptor.java → content/content-service/src/main/java/cn/reghao/tnb/content/app/config/web/AccessInterceptor.java

@@ -10,14 +10,14 @@ import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
 
 /**
- * 用户访问日志拦截器
+ * HTTP 请求拦截器
  *
  * @author reghao
- * @date 2021-11-14 14:19:07
+ * @date 2025-07-18 09:18:16
  */
 @Slf4j
 @Component
-public class AccessLogInterceptor implements HandlerInterceptor {
+public class AccessInterceptor implements HandlerInterceptor {
     @Override
     public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler)
             throws Exception {

content/content-service/src/main/java/cn/reghao/tnb/content/app/config/web/WebConfig.java

@@ -10,25 +10,25 @@ import javax.servlet.Filter;
 
 /**
  * @author reghao
- * @date 2023-08-25 21:14:23
+ * @date 2025-07-18 09:18:16
  */
 @Configuration
 public class WebConfig implements WebMvcConfigurer {
-    private final AccessLogInterceptor accessLogInterceptor;
+    private final AccessInterceptor accessInterceptor;
 
-    public WebConfig(AccessLogInterceptor accessLogInterceptor) {
-        this.accessLogInterceptor = accessLogInterceptor;
+    public WebConfig(AccessInterceptor accessInterceptor) {
+        this.accessInterceptor = accessInterceptor;
     }
 
     @Override
     public void addInterceptors(InterceptorRegistry registry) {
-        registry.addInterceptor(accessLogInterceptor);
+        registry.addInterceptor(accessInterceptor);
     }
 
     @Bean
-    public FilterRegistrationBean<Filter> filterRegistrationBean(TokenFilter tokenFilter) {
+    public FilterRegistrationBean<Filter> filterRegistrationBean(AccessFilter accessFilter) {
         FilterRegistrationBean<Filter> registrationBean = new FilterRegistrationBean<>();
-        registrationBean.setFilter(tokenFilter);
+        registrationBean.setFilter(accessFilter);
         registrationBean.addUrlPatterns("*");
         return registrationBean;
     }

content/content-service/src/main/java/cn/reghao/tnb/content/app/util/AppLog.java

@@ -0,0 +1,23 @@
+package cn.reghao.tnb.content.app.util;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+
+/**
+ * @author reghao
+ * @date 2025-07-18 11:33:46
+ */
+@AllArgsConstructor
+@NoArgsConstructor
+@Getter
+public class AppLog {
+    private String requestId;
+    private String app;
+    private String host;
+    private long timestamp;
+    private String level;
+    private String thread;
+    private String logger;
+    private String message;
+}

content/content-service/src/main/java/cn/reghao/tnb/content/app/util/rabbit/RabbitAppender.java

@@ -0,0 +1,53 @@
+package cn.reghao.tnb.content.app.util.rabbit;
+
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.UnsynchronizedAppenderBase;
+import cn.reghao.jutil.jdk.serializer.JsonConverter;
+import cn.reghao.tnb.content.app.util.AppLog;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+
+/**
+ * @author reghao
+ * @date 2025-07-18 09:54:51
+ */
+public class RabbitAppender extends UnsynchronizedAppenderBase<ILoggingEvent> {
+    private final String app;
+    private final String host;
+    private final RabbitTemplate rabbitTemplate;
+
+    public RabbitAppender(String app, String host, RabbitTemplate rabbitTemplate) {
+        setName("rabbitAppender");
+        this.app = app;
+        this.host = host;
+        this.rabbitTemplate = rabbitTemplate;
+    }
+
+    @Override
+    public void start() {
+        super.start();
+    }
+
+    @Override
+    public void stop() {
+        super.stop();
+    }
+
+    @Override
+    protected void append(ILoggingEvent event) {
+        AppLog appLog = getAppLog(event);
+        String jsonPayload = JsonConverter.objectToJson(appLog);
+
+        String routingKey = "tnb.log.runtime";
+        rabbitTemplate.convertAndSend(routingKey, jsonPayload);
+    }
+
+    private AppLog getAppLog(ILoggingEvent event) {
+        String requestId = event.getMDCPropertyMap().get("request_id");
+        long timestamp = event.getTimeStamp();
+        String level = event.getLevel().toString();
+        String thread = event.getThreadName();
+        String logger = event.getLoggerName();
+        String message = event.getFormattedMessage();
+        return new AppLog(requestId, app, host, timestamp, level, thread, logger, message);
+    }
+}

content/content-service/src/main/java/cn/reghao/tnb/content/app/vod/service/LogService.java

@@ -1,6 +1,5 @@
 package cn.reghao.tnb.content.app.vod.service;
 
-import cn.reghao.tnb.common.prop.LogProperties;
 import org.springframework.cloud.context.config.annotation.RefreshScope;
 import org.springframework.cloud.context.environment.EnvironmentChangeEvent;
 import org.springframework.context.ApplicationContext;
@@ -18,22 +17,18 @@ import java.util.Set;
  */
 @RefreshScope
 @Service
+@Deprecated
 public class LogService {
-    private final LogProperties logProperties;
+    //private final LogProperties logProperties;
 
-    public LogService(LogProperties logProperties) {
-        this.logProperties = logProperties;
+    public LogService() {
     }
 
     public Map<String, Object> getLogProperties() {
-        boolean enabled = logProperties.getEnabled();
-        String wsUrl = logProperties.getWsUrl();
-        String token = logProperties.getToken();
-
         Map<String, Object> map = new HashMap<>();
-        map.put("enabled", enabled);
+        /*map.put("enabled", enabled);
         map.put("wsUrl", wsUrl);
-        map.put("token", token);
+        map.put("token", token);*/
         return map;
     }
 

content/content-service/src/main/java/cn/reghao/tnb/content/app/vod/service/impl/RecommendServiceImpl.java

@@ -1,5 +1,6 @@
 package cn.reghao.tnb.content.app.vod.service.impl;
 
+import cn.reghao.jutil.web.ServletUtil;
 import cn.reghao.jutil.web.WebResult;
 import cn.reghao.tnb.common.db.PageScroll;
 import cn.reghao.tnb.content.api.dto.VideoCard;
@@ -17,6 +18,7 @@ import cn.reghao.tnb.content.app.vod.service.rcmd.task.RcmdProducer;
 import cn.reghao.tnb.content.app.vod.service.rcmd.RedisKeys;
 import cn.reghao.tnb.content.app.util.redis.ds.RedisSet;
 import cn.reghao.tnb.user.api.iface.UserService;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.dubbo.config.annotation.DubboReference;
 import org.springframework.stereotype.Service;
 
@@ -28,6 +30,7 @@ import java.util.stream.Collectors;
  * @author reghao
  * @date 2024-08-07 07:32:32
  */
+@Slf4j
 @Service
 public class RecommendServiceImpl implements RecommendService {
     @DubboReference(check = false, retries = 0, timeout = 3000)
@@ -73,10 +76,10 @@ public class RecommendServiceImpl implements RecommendService {
             return;
         }
 
-        RcmdData rcmdData = new RcmdData(loginUser, size);
         try {
-            rcmdProducer.put(rcmdData);
-        } catch (InterruptedException e) {
+            String requestId = ServletUtil.getHeader("x-request-id");
+            rcmdProducer.put(new RcmdData(requestId, loginUser, size));
+        } catch (Exception e) {
             e.printStackTrace();
             Thread.currentThread().interrupt();
         }

content/content-service/src/main/java/cn/reghao/tnb/content/app/vod/service/rcmd/task/RcmdConsumer.java

@@ -1,6 +1,5 @@
 package cn.reghao.tnb.content.app.vod.service.rcmd.task;
 
-import cn.reghao.jutil.jdk.thread.ThreadPoolWrapper;
 import cn.reghao.tnb.content.app.util.redis.ds.RedisList;
 import cn.reghao.tnb.content.app.util.redis.ds.RedisSet;
 import cn.reghao.tnb.content.app.vod.service.ContentPermission;
@@ -9,12 +8,13 @@ import cn.reghao.tnb.content.app.vod.service.rcmd.UserInterestBased;
 import cn.reghao.tnb.user.api.iface.UserService;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.dubbo.config.annotation.DubboReference;
+import org.slf4j.MDC;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.stereotype.Component;
 
 import javax.annotation.PostConstruct;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Future;
-import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.Executor;
 
 /**
  * @author reghao
@@ -26,40 +26,41 @@ public class RcmdConsumer {
     @DubboReference(check = false)
     private UserService userService;
 
-    private final int threads = 10;
-    private final ExecutorService threadPool = ThreadPoolWrapper.threadPool("data-consumer", threads);
+    private final Executor taskExecutor;
     private final Object monitor;
     private final RcmdProducer rcmdProducer;
     private final VideoPostQuery videoPostQuery;
-    private final UserInterestBased userInterestBased;
     private final RedisSet redisSet;
     private final RedisList redisList;
     private final ContentPermission contentPermission;
+    private final UserInterestBased userInterestBased;
 
-    public RcmdConsumer(VideoPostQuery videoPostQuery, UserInterestBased userInterestBased, RedisSet redisSet,
-                        RedisList redisList, ContentPermission contentPermission, RcmdProducer rcmdProducer) {
+    public RcmdConsumer(@Qualifier("taskExecutor") Executor taskExecutor, VideoPostQuery videoPostQuery, RedisSet redisSet, RedisList redisList,
+                        ContentPermission contentPermission, RcmdProducer rcmdProducer, UserInterestBased userInterestBased) {
+        this.taskExecutor = taskExecutor;
         this.videoPostQuery = videoPostQuery;
-        this.userInterestBased = userInterestBased;
         this.redisSet = redisSet;
         this.redisList = redisList;
         this.contentPermission = contentPermission;
         this.monitor = rcmdProducer.getMonitor();
         this.rcmdProducer = rcmdProducer;
+        this.userInterestBased = userInterestBased;
     }
 
     @PostConstruct
     public void run() {
-        threadPool.submit(new ConsumerThread());
+        taskExecutor.execute(new ConsumerThread());
         log.info("rcmd producer-consumer 模型启动...");
     }
 
     class ConsumerThread implements Runnable {
         @Override
         public void run() {
-            ThreadPoolExecutor tpe = (ThreadPoolExecutor) threadPool;
+            ThreadPoolTaskExecutor executor = (ThreadPoolTaskExecutor) taskExecutor;
+            int capacity = executor.getQueueCapacity();
             while (!Thread.interrupted()) {
                 try {
-                    if (tpe.getActiveCount() < threads) {
+                    if (executor.getQueueSize() < capacity) {
                         dispatch();
                     } else {
                         //log.info("当前有 {} 个活跃线程, 休眠 10s 等待线程池空闲...", tpe.getActiveCount());
@@ -78,10 +79,12 @@ public class RcmdConsumer {
             if (object != null) {
                 if (object instanceof RcmdData) {
                     RcmdData rcmdData = (RcmdData) object;
+                    String requestId = rcmdData.getRequestId();
+                    MDC.put("request_id", requestId);
+
                     int mode = userService.getRecommendMode(rcmdData.getUserId());
-                    RcmdTask rcmdTask = new RcmdTask(rcmdData, videoPostQuery, userInterestBased,
-                            redisSet, redisList, contentPermission, mode);
-                    Future<?> future = threadPool.submit(rcmdTask);
+                    RcmdTask rcmdTask = new RcmdTask(rcmdData, videoPostQuery, redisSet, redisList, contentPermission, mode);
+                    taskExecutor.execute(rcmdTask);
                 } else {
                     log.error("Object 类型未知");
                 }

content/content-service/src/main/java/cn/reghao/tnb/content/app/vod/service/rcmd/task/RcmdData.java

@@ -10,6 +10,7 @@ import lombok.Getter;
 @AllArgsConstructor
 @Getter
 public class RcmdData {
+    private String requestId;
     private long userId;
     private int size;
 }

content/content-service/src/main/java/cn/reghao/tnb/content/app/vod/service/rcmd/task/RcmdTask.java

@@ -10,6 +10,7 @@ import cn.reghao.tnb.content.app.vod.service.VideoPostQuery;
 import cn.reghao.tnb.content.app.vod.service.rcmd.RedisKeys;
 import cn.reghao.tnb.content.app.vod.service.rcmd.UserInterestBased;
 import lombok.extern.slf4j.Slf4j;
+import org.slf4j.MDC;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -25,16 +26,15 @@ public class RcmdTask implements Runnable {
     private final RcmdData rcmdData;
     private final RedisSet redisSet;
     private final RedisList redisList;
-    private final UserInterestBased userInterestBased;
     private final VideoPostQuery videoPostQuery;
     private final ContentPermission contentPermission;
     private final int mode;
+    private UserInterestBased userInterestBased;
 
-    public RcmdTask(RcmdData rcmdData, VideoPostQuery videoPostQuery, UserInterestBased userInterestBased,
-                    RedisSet redisSet, RedisList redisList, ContentPermission contentPermission, int mode) {
+    public RcmdTask(RcmdData rcmdData, VideoPostQuery videoPostQuery, RedisSet redisSet, RedisList redisList,
+                    ContentPermission contentPermission, int mode) {
         this.rcmdData = rcmdData;
         this.videoPostQuery = videoPostQuery;
-        this.userInterestBased = userInterestBased;
         this.redisSet = redisSet;
         this.redisList = redisList;
         this.contentPermission = contentPermission;
@@ -68,7 +68,8 @@ public class RcmdTask implements Runnable {
             e.printStackTrace();
         }
 
-        log.info("{}'s rcmd-task cost {}s", loginUser, (System.currentTimeMillis()-start)/1000);
+        String requestId = MDC.get("request_id");
+        log.info("{}: {}'s rcmd-task cost {}s", requestId, loginUser, (System.currentTimeMillis()-start)/1000);
     }
 
     List<String> getRedisVideoIds(long loginUser) {

content/content-service/src/main/resources/application-dev.yml

@@ -1,7 +1,3 @@
-log:
-  enabled: false
-  ws-url: ws://bnt.reghao.cn/ws/log/push
-  token: 012345678
 dubbo:
   registry:
     address: zookeeper://localhost:2181

content/content-service/src/main/resources/application-test.yml

@@ -1,7 +1,3 @@
-log:
-  enabled: false
-  ws-url: ws://bnt.reghao.cn/ws/log/push
-  token: 012345678
 dubbo:
   registry:
     address: zookeeper://192.168.0.209:2181

content/content-service/src/main/resources/logback-spring.xml

@@ -4,7 +4,7 @@
     <appender name="consoleLog" class="ch.qos.logback.core.ConsoleAppender">
         <layout class="ch.qos.logback.classic.PatternLayout">
             <pattern>
-                %d{HH:mm:ss.SSS} [%thread] %-5level %c %M %L - %msg%n
+                %d{HH:mm:ss.SSS} [%X{request_id}] [%thread] %-5level %c %M %L - %msg%n
             </pattern>
         </layout>
     </appender>
@@ -18,7 +18,7 @@
         </filter>
         <encoder>
             <pattern>
-                %d{HH:mm:ss.SSS} %-5level %c %M %L - %msg%n
+                %d{HH:mm:ss.SSS} [%X{request_id}] [%thread] %-5level %c %M %L - %msg%n
             </pattern>
             <charset>UTF-8</charset>
         </encoder>
@@ -37,7 +37,7 @@
         </filter>
         <encoder>
             <pattern>
-                %d{HH:mm:ss.SSS} %-5level %c %M %L - %msg%n
+                %d{HH:mm:ss.SSS} [%X{request_id}] [%thread] %-5level %c %M %L - %msg%n
             </pattern>
             <charset>UTF-8</charset>
         </encoder>