springcloud trace SDK 自研方案

逝兮诚

发布于 2021-12-08 16:20:09

6300

发布于 2021-12-08 16:20:09

文章被收录于专栏：代码人生代码人生

SDK 形式，利用 threadlocal 实现 trace。http, grpc, rabbitMQ, springcloud-gateway, 异步线程池这类常见场景。

客户端在协议 header 中增加 x-request-id，服务端解析 header 中 x-request-id，存储在 threadlocal，日志框架格式化输出。

java 服务只用依赖 common-tracing.jar lib。

日志格式化输出 traceId 可以使用 MDC，各种协议支持可以考虑 springBoot 的自动注入机制。

grpc

grpc 使用 client 拦截器和 server拦截器解析 x-request-id。利用 io.grpc.ClientInterceptor，org.lognet.springboot.grpc.GRpcGlobalInterceptor 实现无侵入自动注入。

RabbitMQ

rabbitMQ 使用 messageConvert 解析 x-request-id，使用 SmartInitializingSingleton 获取 SimpleMessageListenerContainer bean，再注入 messagePostProcessors。

@Slf4j
@Component
@ConditionalOnClass(SimpleMessageListenerContainer.class)
public class FromProcessorConfig implements SmartInitializingSingleton {

    @Autowired(required = false)
    SimpleMessageListenerContainer listenerContainer;


    public void setAfterReceivePostProcessors(SimpleMessageListenerContainer listenerContainer) {
        try {
            Class<?> obj = Class.forName("org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer");
            Field field = obj.getSuperclass().getDeclaredField("afterReceivePostProcessors");
            field.setAccessible(true);
            Collection<MessagePostProcessor> messagePostProcessors = (Collection<MessagePostProcessor>)field.get(listenerContainer);
            if (CollectionUtils.isEmpty(messagePostProcessors)) {
                listenerContainer.setAfterReceivePostProcessors(new TracingFromMessagePostProcessor());
            }else {
                messagePostProcessors.add(new TracingFromMessagePostProcessor());
            }
            log.info("Initial tracingFromMessagePostProcessor success.");
        } catch (Exception e) {
            log.error("Initial tracingFromMessagePostProcessor failed.", e);
        }
    }

    @Override
    public void afterSingletonsInstantiated() {
        if (listenerContainer != null) {
            setAfterReceivePostProcessors(listenerContainer);
        }
    }
}

@Slf4j
@Component
@ConditionalOnClass(RabbitTemplate.class)
public class ToProcessorConfig implements SmartInitializingSingleton {

    @Autowired(required = false)
    RabbitTemplate rabbitTemplate;

    public void setBeforePublishPostProcessors(RabbitTemplate template) {
        try {
            Class<?> obj = Class.forName("org.springframework.amqp.rabbit.core.RabbitTemplate");
            Field field = obj.getDeclaredField("beforePublishPostProcessors");
            field.setAccessible(true);
            Collection<MessagePostProcessor> messagePostProcessors = (Collection<MessagePostProcessor>)field.get(template);
            if (CollectionUtils.isEmpty(messagePostProcessors)) {
                template.setBeforePublishPostProcessors(new TracingToMessagePostProcessor());
            }else {
                messagePostProcessors.add(new TracingToMessagePostProcessor());
            }
            log.info("Initial tracingToMessagePostProcessor success.");
        } catch (Exception e) {
            log.error("Initial tracingToMessagePostProcessor failed.", e);
        }
    }

    @Override
    public void afterSingletonsInstantiated() {
        if (rabbitTemplate != null) {
            setBeforePublishPostProcessors(rabbitTemplate);
        }
    }
}