分布式系统日志追踪实现

系统管理员 2022-10-15 01:57 53阅读 0赞

### 文章目录 ###

*   *   *  1. 设计分析
         *  2. 代码实现
         *  3. common公共模块
         *   *  （1）前置过滤器
             *  （2）改造线程池
             *  （3）HTTP工具类扩展
         *  4. 创建用户服务
         *   *  （1）用户模块
             *  （2）构建线程测试类
             *  （3）测试接口
         *  5. 订单服务
         *  6. 测试效果

### 1. 设计分析 ###

全服务唯一的traceId，可以使用uuid生成，正常来说不会出现重复的；关于服务间传递，对于调用者，在协议头加上traceId，对于被调用者，通过前置拦截器或者过滤器统一拦截；关于服务内部传递，可以使用ThreadLocal传递traceId，一处放置，随处可用；关于多线程传递，分为两种情况：子线程，可以使用InheritableThreadLocal。线程池，需要改造线程池对提交的任务进行包装，把提交者的traceId包装到任务中  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70]  
上面的整个的流程是：  
用户请求经过网关，网关这里省略了其实也就是在用户前多了一个服务而已，不影响本次的实现目标，请求经过网关后到达用户服务，用户服务在前置过滤器阶段就需要有一个traceId，并将其保存在上下文中，这个需要是线程相关的，用户服务需要调用订单服务，此时需要将traceId携带到Header中传输给订单服务中，订单服务同样也保存此次traceId并进行数据整合返回给用户服务。用户服务此时还有一个特别的任务需要处理，而这个任务为了提高处理效率，开启了异步线程又或者通过线程池进行了处理，那么此时这个traceId就不适用了，那么我们就需要通过其他的方式来解决这个问题，可以通过下面几种方式解决  
1：如果是开启子线程去处理，那么可以利用InheritableThreadLocal传递父线程值给子线程  
2：通过自定义线程配置，将当前线程对象的上下文传递给新的线程

### 2. 代码实现 ###

这里使用SpringBoot，它默认使用的日志框架是logback，而且Slf4j提供了一个包装了InheritableThreadLocal的类叫MDC，我们只要把traceId放在MDC中，打印日志的时候统一打印就可以了，不用显式地打印traceId。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 1]  
模块1：公共模块，封装拦截器，traceId的生成，服务内传递，请求头的传递等；  
模块2：用户服务，依赖于公共模块  
模块3：订单服务，依赖于公共模块

### 3. common公共模块 ###

#### （1）前置过滤器 ####

可以使用过滤器也可以使用拦截器实现

@Slf4j
    @WebFilter("/**")
    @Component
    public class TraceFilter implements Filter { 
    
        @Override
        public void init(FilterConfig filterConfig) throws ServletException { 
             log.info("init TraceFilter");
        }
    
        @Override
        public void doFilter(ServletRequest servletRequest, ServletResponse servletResponse, FilterChain chain) throws IOException, ServletException, IOException { 
            HttpServletRequest request = (HttpServletRequest) servletRequest;
            // 从请求头中获取traceId
            String traceId = request.getHeader("traceId");
            //判断traceId是否存在，不存在则创建
            if (StringUtils.isEmpty(traceId)) { 
                traceId = UUID.randomUUID().toString();
            }
            MDC.put("traceId", traceId);
            chain.doFilter(servletRequest, servletResponse);
        }
    
        @Override
        public void destroy() { 
            log.info("destroy TraceFilter");
        }
    }

#### （2）改造线程池 ####

线程间任务执行上下文是独立的，我们需要在公共模块中构建一个扩展线程处理，使得创建的新的线程保存当前线程的上下文信息，这样我们才可以在日志中记录完整的请求流程数据

public class TraceThreadPoolExecutor extends ThreadPoolExecutor { 
    
        @Override
        public void execute(Runnable command) { 
            // 提交者的本地变量
            Map<String, String> contextMap = MDC.getCopyOfContextMap();
            super.execute(()->{ 
                if (contextMap != null) { 
                    // 如果提交者有本地变量，任务执行之前放入当前任务所在的线程的本地变量中
                    MDC.setContextMap(contextMap);
                }
                try { 
                    command.run();
                } finally { 
                    // 任务执行完，清除本地变量，以防对后续任务有影响
                    MDC.clear();
                }
            });
        }
        public TraceThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) { 
            super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
        }
    
        public TraceThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) { 
            super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
        }
    
        public TraceThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) { 
            super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
        }
    
    
        public TraceThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) { 
            super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
        }
    
    
    }

改造Spring的异步线程池，包装提交的任务。

@Slf4j
    @Component
    public class TraceAsyncConfigurer implements AsyncConfigurer { 
    
        @Override
        public Executor getAsyncExecutor() { 
            ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
            executor.setCorePoolSize(8);
            executor.setMaxPoolSize(16);
            executor.setQueueCapacity(100);
            executor.setThreadNamePrefix("async-pool-");
            executor.setTaskDecorator(new MdcTaskDecorator());
            executor.setWaitForTasksToCompleteOnShutdown(true);
            executor.initialize();
            return executor;
        }
    
        @Override
        public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() { 
            return (throwable, method, params) -> log.error("asyc execute error, method={}, params={}", method.getName(), Arrays.toString(params));
        }
    
        public static class MdcTaskDecorator implements TaskDecorator { 
            @Override
            public Runnable decorate(Runnable runnable) { 
                Map<String, String> contextMap = MDC.getCopyOfContextMap();
                return () -> { 
                    if (contextMap != null) { 
                        MDC.setContextMap(contextMap);
                    }
                    try { 
                        runnable.run();
                    } finally { 
                        MDC.clear();
                    }
                };
            }
        }

#### （3）HTTP工具类扩展 ####

@Slf4j
    public class HttpUtils { 
    
        public static String get(String url) throws URISyntaxException { 
            RestTemplate restTemplate = new RestTemplate();
            MultiValueMap<String, String> headers = new HttpHeaders();
            //将traceId放在请求头中，保证服务间调用能够获取记录
            headers.add("traceId", MDC.get("traceId"));
            URI uri = new URI(url);
            RequestEntity<?> requestEntity = new RequestEntity<>(headers, HttpMethod.GET, uri);
            ResponseEntity<String> exchange = restTemplate.exchange(requestEntity, String.class);
            if (exchange.getStatusCode().equals(HttpStatus.OK)) { 
                log.info("send http request success");
            }
            return exchange.getBody();
        }
    
    }

### 4. 创建用户服务 ###

#### （1）用户模块 ####

构建用户服务，暴露端口8081，创建一个测试接口，该接口内调用了订单服务，用户服务需要引用上面创建的公共模块，注意@ComponentScan扫描公共模块

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 2]

#### （2）构建线程测试类 ####

创建UserService测试通过异步线程保存MDC日志记录

public interface UserService { 
    
        void sendMsgByAsy();
    
        void sendMsgByThreadPool();
    }

@Service
    @Slf4j
    public class UserServiceImpl   implements  UserService{ 
    
        public static final TraceThreadPoolExecutor threadPool = new TraceThreadPoolExecutor(5, 5, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100));
    
        @Async
        @Override
        public void sendMsgByAsy() { 
            log.info("通过Async线程执行异步任务");
        }
        @Override
        public void sendMsgByThreadPool() { 
            threadPool.execute(()->log.info("线程池执行异步任务"));
        }
    }

#### （3）测试接口 ####

@Slf4j
    @RestController
    public class UserController { 
    
        @Autowired
        private  UserService userService;
    
        @GetMapping("/user")
        public String a(String name) { 
            log.info("Hello, " + name);
            try { 
                //调用订单服务
                String result = HttpUtils.get("http://localhost:8082/order");
                //执行异步任务
                //测试通过 @Async注解异步线程
                userService.sendMsgByAsy();
                //测试线程池开启异步线程
                userService.sendMsgByThreadPool();
                return  result;
            } catch (Exception e) { 
                log.error("远程调用异常", e);
            }
            return "fail";
        }
    }

### 5. 订单服务 ###

订单服也需要引用公共模块，注意@ComponentScan扫描公共模块，创建如下测试接口

@Slf4j
    @RestController
    public class OrderController { 
    
    
        @GetMapping("/order")
        public String order() { 
            log.info("订单服务...");
            return "success";
        }
    }

### 6. 测试效果 ###

分别开启用户服务和订单服务，访问用户服务接口  
![在这里插入图片描述][20210523105243232.png]

![在这里插入图片描述][2021052310524846.png]  
上面左侧的唯一标识字符串，就是MDC保存的TraceID,可以看到不同服务，不同线程保存的结果是一样的，那么只要按照这样的结果，我们就可以利用它分析日志，并可以通过这个构建分布式日志追踪

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70]: /images/20221014/9e1bd8c57a4d410894e48c84aea59f40.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 1]: /images/20221014/7dbf61e45da140099e500ac89ddfb1fd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L09jdG9wdXMyMQ_size_16_color_FFFFFF_t_70 2]: /images/20221014/9b61b4b46e1c4d37870dbf0b19fd6cf1.png
[20210523105243232.png]: /images/20221014/bdf71d41cb1347909fc6a2c3edf02c1f.png
[2021052310524846.png]: /images/20221014/2add82033ea24f419683fe5269f69fb5.png