分布式微服务架构日志调用链路跟踪-traceId

2023-12-16 05:12

本文主要是介绍分布式微服务架构日志调用链路跟踪-traceId,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

分布式微服务架构日志调用链路跟踪-traceId

在ELK日志集成平台里(日志的写入,收集,跟踪,搜索,分析)

背景知识

在xxx(博主之前的公司),每个前端请求里面,都会在request的header区携带一个traceId 随机数值,用来跟踪在后端的调用链路栈打印.通过ES收集的日志数据,在ELK日志集成平台里,用traceId就能得到用户请求的完整链路,排查问题的效率非常高.那么具体是怎么设计的?
同样的,每个请求的用户会话信息该怎么传递?
traceId,UserId,facilityId,tenantId.

遇到的场景

有几个难题需要思考: 1.客户端调用方法中,遇到了rpc该怎么传递traceId? 2.遇到了feign接口该怎么传递traceId? 3.遇到了mq该怎么传递tranceId? 4.遇到了线程池里的新线程该怎么传递tranceId? 下面针对各个场景进行逐一的分析

xxx的实现方案:

在gateway中对sid进行校验和初步存储 gateway-GlobalFilter-SIDCheckedFilter.java gateway的原理:一系列的router和filter集合.

常见:

  • IP白名单和黑名单控制
  • SID检测
  • 请求签名检测,防止恶意构建请求
  • 用户会话检测,防止非系统用户请求
  • private static String getSID(ServerHttpRequest request) {String sid = request.getHeaders().getFirst(Constants.HEADER_SID);// 将sid放到日志的变量中MDC.put(Constants.HEADER_SID,sid);return sid;
    }
    @Override
    public Mono<Void> execute(ServerHttpRequest request, ServerWebExchange exchange, GatewayFilterChain chain) {String sid = getSID(request);if(logger.isInfoEnabled()) {String url = StringUtils.defaultIfBlank(request.getPath().value(), "");String ip = IpUtils.getIp(request);MDC.put(MDC_SID,sid);MDC.put(MDC_URI,url);MDC.put(MDC_IP,ip);MDC.put(MDC_USER,null);logger.info("网关请求..");}if (validSid) {if (StringUtils.isBlank(sid)) {logger.info("{} SID为空", this.getRequestUri(request));// 构造返回错误信息ApiResponse<String> responseMap = ApiResponse.failResp(RequestCode.SID_ISNULL);return returnError(exchange, HttpStatus.BAD_REQUEST, responseMap);}// 解决高并发状态的多次重复请求问题String redisSidKey = GatewayRedisKeyConstants.GATEWAY_SID_KEY_PREFIX + sid;boolean isSuccess = redisTemplate.opsForValue().setIfAbsent(redisSidKey, GATEWAY_SID_KEY_DEF_VAL, GATEWAY_SID_KEY_EXPIRE_TIME, TimeUnit.MINUTES);if (!isSuccess) {logger.info("SID已存在,重复请求:{}", sid);ApiResponse<String> responseMap = ApiResponse.failResp(RequestCode.SID_REPEATED_REQUESTS);return returnError(exchange, HttpStatus.BAD_REQUEST, responseMap);}}exchange = afterValidHandler(exchange);//验证过后处理器return chain.filter(exchange);
    }/*** 验证过后处理器** @param exchange* @return*/
    @Override
    public ServerWebExchange afterValidHandler(ServerWebExchange exchange) {ServerHttpRequest request = exchange.getRequest();String sid = request.getHeaders().getFirst(Constants.HEADER_SID);//不要验证SID的情况下,如果SI为空,默认生成一个SID,供下游系统使用if (StringUtils.isBlank(sid)) {//生成sidString gatewaySid = "gateway_" + DateTimeUtils.format(LocalDateTime.now(), DateTimeUtils.FORMAT_1) + RandomUtil.randomNumbers(6);//构造新的ServerHttpRequestServerHttpRequest.Builder builder = request.mutate()//往header中添加网关生成的SID.header(com.ess.framework.commons.constant.Constants.HEADER_SID, gatewaySid);// 将sid放到日志的变量中MDC.put(Constants.HEADER_SID,gatewaySid);exchange = exchange.mutate().request(builder.build()).build();}return exchange;
    }
    

    定义个framework-boot的依赖包

    这个依赖包内使用interceptor拦截器对web请求做拦截,拿到traceId,通过threadLocal对象放入一个Context对象中.每个请求对应一个ThreadLocal对象. 每个服务的全局拦截器
    @RefreshScope
    public abstract class BootWebConfigurer implements WebMvcConfigurer {protected final Logger logger = LoggerFactory.getLogger(this.getClass());@Autowiredprivate RequestMappingHandlerAdapter requestMappingHandlerAdapter;@Bean@RefreshScopepublic GatewayInterceptor getGatewayInterceptor() {return new GatewayInterceptor();}@Bean@RefreshScopepublic SIDInterceptor getSIDInterceptor() {return new SIDInterceptor();}@Bean@RefreshScopepublic InnerInterceptor getInnerInterceptor() {return new InnerInterceptor();}@Beanpublic TokenInterceptor getTokenInterceptor() {return new TokenInterceptor();}@Beanpublic LogMdcInterceptor getLogMdcInterceptor() {return new LogMdcInterceptor();}@Beanpublic UnionInterceptor getUnionInterceptor() {return new UnionInterceptor();}@Overridepublic void addInterceptors(InterceptorRegistry registry) {registry.addInterceptor(getSIDInterceptor()).addPathPatterns("/**").excludePathPatterns(ResouceConstants.EXCLUDE_STATIC_RESOURCE);registry.addInterceptor(getGatewayInterceptor()).addPathPatterns("/**").excludePathPatterns(ResouceConstants.EXCLUDE_STATIC_RESOURCE);registry.addInterceptor(getInnerInterceptor()).addPathPatterns("/**").excludePathPatterns(ResouceConstants.EXCLUDE_STATIC_RESOURCE);registry.addInterceptor(getTokenInterceptor()).addPathPatterns("/**").excludePathPatterns(ResouceConstants.EXCLUDE_STATIC_RESOURCE);registry.addInterceptor(getLogMdcInterceptor()).addPathPatterns("/**").excludePathPatterns(ResouceConstants.EXCLUDE_STATIC_RESOURCE);// 添加联盟拦截器registry.addInterceptor(getUnionInterceptor()).addPathPatterns("/**").excludePathPatterns(ResouceConstants.EXCLUDE_STATIC_RESOURCE);}/*** 初始化SIDReturnValueHandler  对所有的响应返回SID字段*/@PostConstructpublic void initSidHandler() {final List<HandlerMethodReturnValueHandler> originalHandlers = new ArrayList<>(requestMappingHandlerAdapter.getReturnValueHandlers());final int deferredPos = obtainValueHandlerPosition(originalHandlers, DeferredResultMethodReturnValueHandler.class);SIDReturnValueHandler decorator = null;for (HandlerMethodReturnValueHandler handler : originalHandlers) {if (handler instanceof RequestResponseBodyMethodProcessor) {decorator = new SIDReturnValueHandler((RequestResponseBodyMethodProcessor) handler);break;}}originalHandlers.add(deferredPos + 1, decorator);requestMappingHandlerAdapter.setReturnValueHandlers(originalHandlers);}private int obtainValueHandlerPosition(final List<HandlerMethodReturnValueHandler> originalHandlers, Class<?> handlerClass) {for (int i = 0; i < originalHandlers.size(); i++) {final HandlerMethodReturnValueHandler valueHandler = originalHandlers.get(i);if (handlerClass.isAssignableFrom(valueHandler.getClass())) {return i;}}return -1;}@Overridepublic void addResourceHandlers(ResourceHandlerRegistry registry) {registry.addResourceHandler("/");registry.addResourceHandler("/favicon.ico").addResourceLocations("classpath:/favicon.ico");registry.addResourceHandler("app.html").addResourceLocations("classpath:/");registry.addResourceHandler("/error");// Swagger2配置registry.addResourceHandler("/v2/**");registry.addResourceHandler("/swagger-resources/**");registry.addResourceHandler("/csrf");registry.addResourceHandler("/webjars/**").addResourceLocations("classpath:/META-INF/resources/webjars/");registry.addResourceHandler("/swagger-ui.html").addResourceLocations("classpath:/META-INF/resources/");registry.addResourceHandler("/doc.html").addResourceLocations("classpath:/META-INF/resources/");}/*** 自定义扩展消息转换器** @param converters*/@Overridepublic void extendMessageConverters(List<HttpMessageConverter<?>> converters) {//定义fastjson转换消息对象FastJsonHttpMessageConverter fastJsonConverter = new FastJsonHttpMessageConverter();//添加fastjson全局配置FastJsonConfig fastJsonConfig = new FastJsonConfig();fastJsonConfig.setSerializerFeatures(// 排序配置SerializerFeature.SortField,SerializerFeature.MapSortField,// 避免对象重复引用SerializerFeature.DisableCircularReferenceDetect,// 格式化输出SerializerFeature.PrettyFormat);fastJsonConfig.setCharset(Charset.forName("UTF-8"));fastJsonConfig.setDateFormat("yyyy-MM-dd HH:mm:ss");//默认json会对属性里的json字符串值进行排序,加了这个Feature.OrderedField则会禁止排序fastJsonConfig.setFeatures(Feature.OrderedField);fastJsonConverter.setFastJsonConfig(fastJsonConfig);//添加支持的MediaType类型fastJsonConverter.setSupportedMediaTypes(Lists.newArrayList(MediaType.APPLICATION_JSON));converters.set(0, fastJsonConverter);//BigDecimal格式化SerializeConfig serializeConfig = SerializeConfig.getGlobalInstance();serializeConfig.put(BigDecimal.class, BigDecimalConfigure.instance);fastJsonConfig.setSerializeConfig(serializeConfig);//字节数组消息转换器(供文件下载使用)converters.add(new ByteArrayHttpMessageConverter());Map<String, HttpMessageConverter<?>> convertersMap = new LinkedHashMap<>();//转换器去重,并设置所有转换器的默认编码for (HttpMessageConverter converter : converters) {String name = converter.getClass().getSimpleName();if (converter instanceof StringHttpMessageConverter) {//设置StringHttpMessageConverter的默认编码为:UTF-8((StringHttpMessageConverter) converter).setDefaultCharset(Charset.forName("UTF-8"));}if (!convertersMap.containsKey(name)) {convertersMap.put(name, converter);}}converters.clear();converters.addAll(Lists.newArrayList(convertersMap.values()));}@Beanpublic Converter<String, LocalDate> localDateConverter() {return new Converter<String, LocalDate>() {@Overridepublic LocalDate convert(String source) {return LocalDate.parse(source, Constants.DATE_FORMATTER);}};}@Beanpublic Converter<String, LocalDateTime> localDateTimeConverter() {return new Converter<String, LocalDateTime>() {@Overridepublic LocalDateTime convert(String source) {return LocalDateTime.parse(source, Constants.DATE_TIME_FORMATTER);}};}/*** Java常用时间类型序列化和反序列化* LocalDateTime、LocalDate、LocalTime、java.util.Date、java.sql.Date、Calendar、Timestamp* 解决通过Feign调用时,客户端是以上类型会报错问题。*/@Beanpublic ObjectMapper objectMapper() {ObjectMapper objectMapper = new ObjectMapper();JavaTimeModule javaTimeModule = new JavaTimeModule();//定义序列化日期时间类型转换器javaTimeModule.addSerializer(LocalDateTime.class, new LocalDateTimeSerializer(DateTimeUtils.DEFAULT_FORMATTER));javaTimeModule.addSerializer(LocalDate.class, new LocalDateSerializer(DateTimeUtils.YEAR_MONTH_DAY_FORMATTER));javaTimeModule.addSerializer(LocalTime.class, new LocalTimeSerializer(DateTimeUtils.HOUR_MINUTE_SECOND_FORMATTER));javaTimeModule.addSerializer(Date.class, new DateSerializer(true, new SimpleDateFormat(DateTimeUtils.DEFAULT_FORMAT)));javaTimeModule.addSerializer(java.sql.Date.class, new SqlDateSerializer().withFormat(true, new SimpleDateFormat(DateTimeUtils.DEFAULT_FORMAT)));javaTimeModule.addSerializer(Calendar.class, new CalendarSerializer(true, new SimpleDateFormat(DateTimeUtils.DEFAULT_FORMAT)));//定义反序列化日期时间类型转换器javaTimeModule.addDeserializer(LocalDateTime.class, new LocalDateTimeDeserializer(DateTimeUtils.DEFAULT_FORMATTER));javaTimeModule.addDeserializer(LocalDate.class, new LocalDateDeserializer(DateTimeUtils.YEAR_MONTH_DAY_FORMATTER));javaTimeModule.addDeserializer(LocalTime.class, new LocalTimeDeserializer(DateTimeUtils.HOUR_MINUTE_SECOND_FORMATTER));javaTimeModule.addDeserializer(Date.class, new DateDeserializers.DateDeserializer(DateDeserializers.DateDeserializer.instance,new SimpleDateFormat(DateTimeUtils.DEFAULT_FORMAT), DateTimeUtils.DEFAULT_FORMAT));javaTimeModule.addDeserializer(java.sql.Date.class, new DateDeserializers.SqlDateDeserializer(new DateDeserializers.SqlDateDeserializer(),new SimpleDateFormat(DateTimeUtils.DEFAULT_FORMAT), DateTimeUtils.DEFAULT_FORMAT));javaTimeModule.addDeserializer(Calendar.class, new DateDeserializers.CalendarDeserializer(new DateDeserializers.CalendarDeserializer(),new SimpleDateFormat(DateTimeUtils.DEFAULT_FORMAT), DateTimeUtils.DEFAULT_FORMAT));javaTimeModule.addDeserializer(Timestamp.class, new DateDeserializers.TimestampDeserializer(new DateDeserializers.TimestampDeserializer(),new SimpleDateFormat(DateTimeUtils.DEFAULT_FORMAT), DateTimeUtils.DEFAULT_FORMAT));objectMapper.registerModule(javaTimeModule);objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);return objectMapper;}
    }

    sid拦截器

    
    package com.ess.framework.boot.interceptor;import com.ess.framework.commons.constant.Constants;
    import com.ess.framework.commons.response.RequestCode;
    import com.ess.framework.commons.utils.EssContextHolder;
    import com.ess.framework.commons.utils.ExceptionUtils;
    import org.apache.commons.lang3.StringUtils;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.cloud.context.config.annotation.RefreshScope;
    import org.springframework.web.servlet.HandlerInterceptor;import javax.servlet.http.HttpServletRequest;
    import javax.servlet.http.HttpServletResponse;/*** SID 拦截器 ,验证SID是否为空*/
    @RefreshScope
    public class SIDInterceptor implements HandlerInterceptor {protected final Logger logger = LoggerFactory.getLogger(this.getClass());/*** 是否验证SID*/@Value("${valid.sid:false}")private boolean validSid = false;@Overridepublic boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {String sid = request.getHeader(Constants.HEADER_SID);// 验证sid是否为空if (validSid && StringUtils.isBlank(sid)) {logger.warn("{} {}", request.getRequestURI(), RequestCode.SID_ISNULL.message());ExceptionUtils.throwBusiness(RequestCode.SID_ISNULL);}// 设置SID到线程变量 这是传输的关键if (StringUtils.isNotBlank(sid)) {EssContextHolder.setSID(sid);} else {EssContextHolder.setSID(null);}return true;}
    }
    

    关于feign接口的传递tranceId

    
    package com.ess.framework.boot.interceptor;import com.ess.framework.commons.constant.Constants;
    import com.ess.framework.commons.utils.EssContextHolder;
    import com.ess.framework.commons.utils.IpUtils;
    import feign.RequestInterceptor;
    import feign.RequestTemplate;
    import lombok.extern.slf4j.Slf4j;
    import org.apache.commons.lang3.StringUtils;
    import org.slf4j.MDC;
    import org.springframework.web.context.request.RequestContextHolder;
    import org.springframework.web.context.request.ServletRequestAttributes;import javax.servlet.http.HttpServletRequest;/*** @ClassName FeignRequestInterceptor* @Description feign拦截器* @Date 2021/5/27 21:56* @Version*/
    @Slf4j
    public class FeignRequestInterceptor implements RequestInterceptor {private final static String HEADER_ESS_FEIGN_IP = "ess-feign-ip";private final static String HEADER_IP = "ip";@Overridepublic void apply(RequestTemplate requestTemplate) {try {String ip = null;String sid = null;String token = null;String unionId = null;/* ExecuteTaskUtils 线程池处理方案**/String threadName = Thread.currentThread().getName();if (StringUtils.startsWith(threadName, "ess-task-pool") || StringUtils.startsWith(threadName, "ess-async-pool")) {sid = MDC.get(Constants.HEADER_SID);ip = MDC.get(HEADER_IP);unionId  = EssContextHolder.getUnionId();token = EssContextHolder.getToken();} else {ServletRequestAttributes attributes = (ServletRequestAttributes) RequestContextHolder.getRequestAttributes();if (attributes == null) {return;}HttpServletRequest request = attributes.getRequest();token = request.getHeader(Constants.HEADER_TOKEN);sid = request.getHeader(Constants.HEADER_SID);ip = IpUtils.getIPAddress(request);unionId  = request.getHeader(Constants.HEADER_UNIONID);}requestTemplate.header(Constants.HEADER_TOKEN, token);requestTemplate.header(Constants.HEADER_SID, sid);requestTemplate.header(Constants.HEADER_UNIONID, unionId);requestTemplate.header(HEADER_ESS_FEIGN_IP, ip);// feign调用都设置网关标识设置为true,防止被拦截requestTemplate.header(Constants.PASS_GATEWAY, "true");} catch (Exception e) {log.error("拦截器异常", e);}}
    }

    由此可见,feign请求的本质就是requestTemplate,而原理上,就是把sid重新塞到request的header里面.

    线程池sid的传递

    那么怎么传递到另外的线程里呢? ==答案就是threadLocal作为管道存储对象.
    
    package com.ess.framework.commons.utils;/*** 线程变量上下文*/
    public class EssContextHolder {private EssContextHolder() {}/*** sid*/private final static ThreadLocal<String> SID = new ThreadLocal<>();/*** token*/private final static ThreadLocal<String> TOKEN = new ThreadLocal<>();/*** 联盟code*/private final static ThreadLocal<String> UNION_CODE = new ThreadLocal<>();/*** 联盟unionId*/private final static ThreadLocal<String> UNION_ID = new ThreadLocal<>();/*** 设置SID** @param sid*/public static void setSID(String sid) {EssContextHolder.SID.set(sid);}/*** 获取SID*/public static String getSID() {return EssContextHolder.SID.get();}/*** 设置TOKEN** @param token*/public static void setToken(String token) {EssContextHolder.TOKEN.set(token);}/*** 获取TOKEN*/public static String getToken() {return EssContextHolder.TOKEN.get();}/*** 设置unionCode*/public static void setUnionCode(String unionCode) {EssContextHolder.UNION_CODE.set(unionCode);}/*** 获取unionCode*/public static String getUnionCode() {return EssContextHolder.UNION_CODE.get();}/*** 设置unionId*/public static void setUnionId(String unionId) {EssContextHolder.UNION_ID.set(unionId);}/*** 获取联盟unionId*/public static String getUnionId() {return EssContextHolder.UNION_ID.get();}
    }

    重载线程池实现细节,这样springboot在使用自定义线程池时,就会初始化个性化的实现细节,把sid等会话状态传递到线程的ThreadLocal里.

    package com.ess.framework.boot.asynctask;import com.ess.framework.commons.utils.EssContextHolder;
    import org.slf4j.MDC;import java.util.concurrent.*;/*** @ClassName ttt* @Description TODO* @Author shengfq* @Date 2021/5/28 0028 上午 10:53* @Version*/
    public class ThreadPoolExecutorMdcWrapper extends ThreadPoolExecutor {public ThreadPoolExecutorMdcWrapper(AsyncTaskThreadPoolConfig config,ThreadFactory threadFactory,RejectedExecutionHandler handler ) {this(config.getCorePoolSize(), config.getMaxPoolSize(), config.getKeepAliveSecond(), TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(config.getQueueCapacity()),threadFactory,handler);}public ThreadPoolExecutorMdcWrapper(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,BlockingQueue<Runnable> workQueue) {super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);}public ThreadPoolExecutorMdcWrapper(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);}public ThreadPoolExecutorMdcWrapper(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);}public ThreadPoolExecutorMdcWrapper(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,RejectedExecutionHandler handler) {super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);}@Overridepublic void execute(Runnable task) {String sid =   EssContextHolder.getSID();String token = EssContextHolder.getToken();String unionId = EssContextHolder.getUnionId();super.execute(ThreadMdcUtil.wrap(sid,token,unionId,task, MDC.getCopyOfContextMap()));}@Overridepublic <T> Future<T> submit(Runnable task, T result) {String sid =   EssContextHolder.getSID();String token = EssContextHolder.getToken();String unionId = EssContextHolder.getUnionId();return super.submit(ThreadMdcUtil.wrap(sid,token,unionId,task, MDC.getCopyOfContextMap()), result);}@Overridepublic <T> Future<T> submit(Callable<T> task) {String sid =   EssContextHolder.getSID();String token = EssContextHolder.getToken();String unionId = EssContextHolder.getUnionId();return super.submit(ThreadMdcUtil.wrap(sid,token,unionId,task, MDC.getCopyOfContextMap()));}@Overridepublic Future<?> submit(Runnable task) {String sid =   EssContextHolder.getSID();String token = EssContextHolder.getToken();String unionId = EssContextHolder.getUnionId();return super.submit(ThreadMdcUtil.wrap(sid,token,unionId,task, MDC.getCopyOfContextMap()));}
    }

    mq消息的sid传递

    如何传递给mq的消费者呢?实际上有了EssContextHolder对象,在Thread里获取sid很方便,如何通过mq传递sid,可以把sid放入message的header区,在消费端取出来,存入消费端线程的ThreadLocal对象中,就能起到传递的作用.
    消息发送

    消息接收

    import java.io.IOException;
    import java.nio.charset.StandardCharsets;
    import java.util.Map;
    import java.util.function.Consumer;import org.hzero.core.exception.OptimisticLockException;
    import org.hzero.core.message.MessageAccessor;
    import org.springframework.amqp.core.Message;import com.rabbitmq.client.Channel;
    import com..mom.me.common.constants.MeBaseConstants;
    import com..mom.me.common.mq.MessageHandler;import cn.hutool.core.util.ObjectUtil;
    import cn.hutool.json.JSONObject;
    import cn.hutool.json.JSONUtil;import io.choerodon.core.exception.CommonException;
    import io.choerodon.core.oauth.CustomUserDetails;
    import io.choerodon.core.oauth.DetailsHelper;import lombok.extern.slf4j.Slf4j;/*** <p>* 消息处理类* </p>*/
    @Slf4j
    public abstract class AbstractMessageHandlers implements MessageHandler {private AbstractMessageHandlers() {}public static void handleMessage(Message msg, Map<String, Object> headers, Channel channel, Consumer<String> fn) {String consumerQueue = msg.getMessageProperties().getConsumerQueue();try {log.trace("消息队列[{}]处理, headers={}", consumerQueue, headers);fn.accept(new String(msg.getBody(), StandardCharsets.UTF_8));} catch (CommonException e) {log.error("消息队列[{}], 消息处理失败业务异常->{}", consumerQueue,MessageAccessor.getMessage(e.getCode(), e.getParameters(), e.getMessage()).desc());} catch (IllegalArgumentException e) {log.error("消息队列[{}], 消息处理失败业务异常: {}", consumerQueue, e.getMessage());} catch (OptimisticLockException e) {// 如果乐观锁异常 则触发重试log.warn("消息队列[{}], 数据处理乐观锁异常触发重试", consumerQueue);throw e;} catch (Exception e) {log.error("消息队列[{}], 消息处理失败系统异常", consumerQueue, e);}}/*** 携带用户会话信息 -> 消息头获取用户会话信息* * 经AbstractMessageSender 发送MQ消息会在消息头附加用户会话信息* * JSON示例如下* * @param msg* @param headers* @param channel* @param fn*/public static void handleWithUser(Message msg, Map<String, Object> headers, Channel channel, Consumer<String> fn) {try {// 生产执行发出的mq消息头带有用户会话信息 切勿用于其他系统、模块的mq消息处理Object userDetails = headers.get(Constants.HEADER_SID);if (ObjectUtil.isNull(userDetails)) {throw new CommonException("MQ消息头获取用户会话信息失败!!!");}JSONObject jsonObject = JSONUtil.parseObj(userDetails);log.trace("设置用户会话信息,传入JSONObject信息: {}", jsonObject);CustomUserDetails details = new CustomUserDetails(jsonObject.getStr("username"), MeBaseConstants.DEFAULT);details.setTenantId(jsonObject.getLong("tenantId"));details.setUserId(jsonObject.getLong("userId"));details.setOrganizationId(jsonObject.getLong("organizationId"));details.setRealName(jsonObject.getStr("realName"));details.setLanguage(jsonObject.getStr("language"));details.setAdditionInfo(jsonObject.getJSONObject("additionInfo"));DetailsHelper.setCustomUserDetails(details);// 设置用户会话异常} catch (Exception e) {log.error("消息队列[{}], 消息处理失败", msg.getMessageProperties().getConsumerQueue(), e);return;}handleMessage(msg, headers, channel, fn);}public static void handleWithNack(Message msg, Map<String, Object> headers, Channel channel, Consumer<String> fn) {String consumerQueue = msg.getMessageProperties().getConsumerQueue();try {log.trace("消息队列[{}]处理, headers={}", consumerQueue, headers);fn.accept(new String(msg.getBody(), StandardCharsets.UTF_8));ack(msg, channel);} catch (CommonException e) {log.error("消息队列[{}], 消息处理失败业务异常->{}", consumerQueue,MessageAccessor.getMessage(e.getCode(), e.getParameters(), e.getMessage()).desc());nack(msg, channel);} catch (IllegalArgumentException e) {log.error("消息队列[{}], 消息处理失败业务异常: {}", consumerQueue, e.getMessage());nack(msg, channel);} catch (Exception e) {log.error("消息队列[{}], 消息处理失败系统异常:{}", consumerQueue, e);nack(msg, channel);}}public static void handleWithUserThrowEx(Message msg, Map<String, Object> headers, Channel channel,Consumer<String> fn) {String consumerQueue = msg.getMessageProperties().getConsumerQueue();try {// 生产执行发出的mq消息头带有用户会话信息 切勿用于其他系统、模块的mq消息处理Object userDetails = headers.get(MeBaseConstants.USER_DETAILS_KEY);if (ObjectUtil.isNull(userDetails)) {throw new CommonException("MQ消息头获取用户会话信息失败!!!");}JSONObject jsonObject = JSONUtil.parseObj(userDetails);log.trace("设置用户会话信息,传入JSONObject信息: {}", jsonObject);CustomUserDetails details = new CustomUserDetails(jsonObject.getStr("username"), MeBaseConstants.DEFAULT);details.setTenantId(jsonObject.getLong("tenantId"));details.setUserId(jsonObject.getLong("userId"));details.setOrganizationId(jsonObject.getLong("organizationId"));details.setRealName(jsonObject.getStr("realName"));details.setLanguage(jsonObject.getStr("language"));details.setAdditionInfo(jsonObject.getJSONObject("additionInfo"));DetailsHelper.setCustomUserDetails(details);log.trace("消息队列[{}]处理, headers={}", consumerQueue, headers);fn.accept(new String(msg.getBody(), StandardCharsets.UTF_8));} catch (CommonException e) {log.error("消息队列[{}], 消息处理失败业务异常->{}, 详细信息:", consumerQueue,MessageAccessor.getMessage(e.getCode(), e.getParameters(), e.getMessage()).desc(), e);throw e;} catch (Exception e) {log.error("消息队列[{}], 消息处理失败系统异常:{}", consumerQueue, e);throw e;}}private static void ack(Message msg, Channel channel) {try {long deliverTag = msg.getMessageProperties().getDeliveryTag();channel.basicAck(deliverTag, false);} catch (IOException ioe) {log.error("消息处理ack异常{}", ioe.getMessage(), ioe);}}private static void nack(Message msg, Channel channel) {try {long deliverTag = msg.getMessageProperties().getDeliveryTag();channel.basicNack(deliverTag, false, true);} catch (IOException ioe) {log.error("消息处理nack异常{}", ioe.getMessage(), ioe);}}}

    总结

    综上所述,sid的设置,拦截,传递,存储都有对应的机制.那么在日志系统中的存储则是通过org.slf4j.MDC;这个对象来实现的.

    这个对象的实现机制不在本文中展开细节,后续日志的入库,检索,都是通过slf4j这个日志框架和ELK日志系统进行.本文着重讲解了在应用服务中一个跟踪请求的链路id是如何传递,存储的.

    属于抛砖引玉的内容,内容较为主观,希望能对自己和其他开发者在分布式日志调用链路上有点启发.谢谢

    参考

    SpringBoot + MDC 实现全链路调用日志跟踪
    微服务-网关Spring Gateway进阶-日志跟踪唯一ID
    日志全链路追踪之MDC
    秒杀 : 做一个完善的全链路日志实现方案有多简单

这篇关于分布式微服务架构日志调用链路跟踪-traceId的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/499223

相关文章

mybatis的整体架构

mybatis的整体架构分为三层: 1.基础支持层 该层包括:数据源模块、事务管理模块、缓存模块、Binding模块、反射模块、类型转换模块、日志模块、资源加载模块、解析器模块 2.核心处理层 该层包括:配置解析、参数映射、SQL解析、SQL执行、结果集映射、插件 3.接口层 该层包括:SqlSession 基础支持层 该层保护mybatis的基础模块,它们为核心处理层提供了良好的支撑。

百度/小米/滴滴/京东,中台架构比较

小米中台建设实践 01 小米的三大中台建设:业务+数据+技术 业务中台--从业务说起 在中台建设中,需要规范化的服务接口、一致整合化的数据、容器化的技术组件以及弹性的基础设施。并结合业务情况,判定是否真的需要中台。 小米参考了业界优秀的案例包括移动中台、数据中台、业务中台、技术中台等,再结合其业务发展历程及业务现状,整理了中台架构的核心方法论,一是企业如何共享服务,二是如何为业务提供便利。

如何在页面调用utility bar并传递参数至lwc组件

1.在app的utility item中添加lwc组件: 2.调用utility bar api的方式有两种: 方法一,通过lwc调用: import {LightningElement,api ,wire } from 'lwc';import { publish, MessageContext } from 'lightning/messageService';import Ca

【区块链 + 人才服务】可信教育区块链治理系统 | FISCO BCOS应用案例

伴随着区块链技术的不断完善,其在教育信息化中的应用也在持续发展。利用区块链数据共识、不可篡改的特性, 将与教育相关的数据要素在区块链上进行存证确权,在确保数据可信的前提下,促进教育的公平、透明、开放,为教育教学质量提升赋能,实现教育数据的安全共享、高等教育体系的智慧治理。 可信教育区块链治理系统的顶层治理架构由教育部、高校、企业、学生等多方角色共同参与建设、维护,支撑教育资源共享、教学质量评估、

【区块链 + 人才服务】区块链集成开发平台 | FISCO BCOS应用案例

随着区块链技术的快速发展,越来越多的企业开始将其应用于实际业务中。然而,区块链技术的专业性使得其集成开发成为一项挑战。针对此,广东中创智慧科技有限公司基于国产开源联盟链 FISCO BCOS 推出了区块链集成开发平台。该平台基于区块链技术,提供一套全面的区块链开发工具和开发环境,支持开发者快速开发和部署区块链应用。此外,该平台还可以提供一套全面的区块链开发教程和文档,帮助开发者快速上手区块链开发。

flume系列之:查看flume系统日志、查看统计flume日志类型、查看flume日志

遍历指定目录下多个文件查找指定内容 服务器系统日志会记录flume相关日志 cat /var/log/messages |grep -i oom 查找系统日志中关于flume的指定日志 import osdef search_string_in_files(directory, search_string):count = 0

我在移动打工的日志

客户:给我搞一下录音 我:不会。不在服务范围。 客户:是不想吧 我:笑嘻嘻(气笑) 客户:小姑娘明明会,却欺负老人 我:笑嘻嘻 客户:那我交话费 我:手机号 客户:给我搞录音 我:不会。不懂。没搞过。 客户:那我交话费 我:手机号。这是电信的啊!!我这是中国移动!! 客户:我不管,我要充话费,充话费是你们的 我:可是这是移动!!中国移动!! 客户:我这是手机号 我:那又如何,这是移动!你是电信!!

系统架构设计师: 信息安全技术

简简单单 Online zuozuo: 简简单单 Online zuozuo 简简单单 Online zuozuo 简简单单 Online zuozuo 简简单单 Online zuozuo :本心、输入输出、结果 简简单单 Online zuozuo : 文章目录 系统架构设计师: 信息安全技术前言信息安全的基本要素:信息安全的范围:安全措施的目标:访问控制技术要素:访问控制包括:等保

【LabVIEW学习篇 - 21】:DLL与API的调用

文章目录 DLL与API调用DLLAPIDLL的调用 DLL与API调用 LabVIEW虽然已经足够强大,但不同的语言在不同领域都有着自己的优势,为了强强联合,LabVIEW提供了强大的外部程序接口能力,包括DLL、CIN(C语言接口)、ActiveX、.NET、MATLAB等等。通过DLL可以使用户很方便地调用C、C++、C#、VB等编程语言写的程序以及windows自带的大

利用命令模式构建高效的手游后端架构

在现代手游开发中,后端架构的设计对于支持高并发、快速迭代和复杂游戏逻辑至关重要。命令模式作为一种行为设计模式,可以有效地解耦请求的发起者与接收者,提升系统的可维护性和扩展性。本文将深入探讨如何利用命令模式构建一个强大且灵活的手游后端架构。 1. 命令模式的概念与优势 命令模式通过将请求封装为对象,使得请求的发起者和接收者之间的耦合度降低。这种模式的主要优势包括: 解耦请求发起者与处理者