Вызов службы Java, отслеживание всего процесса, простая схема реализации

Java
Вызов службы Java, отслеживание всего процесса, простая схема реализации

предисловие

Некоторое время назад я помогал команде проекта в реконструкции системы, системное приложение было разбито на несколько сервисов, а некоторые сервисы были развернуты в кластерах. С развитием вышеупомянутой архитектуры, ELK + Filebeat, естественно, были введены для сбора журналов. Но при просмотре логов с Kibana из-за отсутствияTraceID, из-за чего разработчикам сложно отфильтровать соответствующие логи указанного запроса, а также сложно отследить процесс обращения приложения к нижестоящему сервису, что занимает много времени. После того, как я сам несколько раз проверил проблему, я не мог выдерживать каждый раз времени, поэтому я поспешно упомянул об этом ремонте супервайзеру.

Эта статья предназначена в основном для того, чтобы зафиксировать мои собственные взгляды на проект.TraceIDИсследование решения преобразования отслеживания ссылок, возникших проблем и конкретной реализации, в то же время, это преобразование также углубило мое собственное понимание отслеживания распределенных сервисов, о чем я также написал в нем.

Основное содержание этой статьи:

  • Первоначальная реализация
  • Проблема с потерей traceId асинхронного потока
  • Трассировка ссылок для Dubbo RPC
  • Отслеживание ссылок для службы HTTP
  • Подумайте о реализации SpringCloud Sleuth
  • резюме

1. Начальная реализация

Общая идея состоит в том, чтобы использовать функцию MDC slf4j + Spring Interceptor для создания traceId и помещения его в MDC при поступлении внешнего запроса.

MDC

Вот краткое введение в MDC.

MDC (сопоставленный диагностический контекст) — это функция, предоставляемая log4j и logback для облегчения ведения журнала в многопоточных условиях. MDC можно рассматривать как карту, привязанную к текущему потоку, к которой можно добавить пары ключ-значение. Доступ к содержимому, содержащемуся в MDC, можно получить с помощью кода, выполняющегося в том же потоке. Дочерние потоки текущего потока наследуют содержимое MDC в своем родительском потоке. Когда требуется ведение журнала, просто получите необходимую информацию от MDC. Содержимое MDC сохраняется программой при необходимости. Для веб-приложения эти данные обычно хранятся в самом начале обрабатываемого запроса.

Проще говоря, MDC — это структура журналов, предоставляемаяInheritableThreadLocal, в него можно в коде проекта поместить пары ключ-значение, а печатать изThreadLocalПолучите соответствующее значение и распечатайте его. Подробный принцип не будет повторяться в этой статье. Просто посмотрите на классы реализации в log4j и logback.

выполнить

  1. Пользовательские перехватчики SpringTraceInterceptor 
/**
 * @author Richard_yyf
 */
public class TraceInterceptor extends HandlerInterceptorAdapter {

    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
        // 清空
        MDC.clear();

        ThreadMdcUtil.setTraceIdIfAbsent();

        //后续逻辑... ...
        return true;
    }
}
  1. зарегистрировать перехватчик
/**
 * @author Richard_yyf
 */
@Configuration
public class WebMvcConfig implements WebMvcConfigurer {

    @Override
    public void addInterceptors(InterceptorRegistry registry) {
        registry.addInterceptor(traceInterceptor())
                .addPathPatterns("/**")
                .order(0);
    }

    @Bean
    public TraceInterceptor traceInterceptor() {
        return new TraceInterceptor();
    }

}

ThreadMdcUtilЭто инструментальный класс, который я сам инкапсулировал, который оборачивает некоторые операции с TraceId:

public class ThreadMdcUtil {
    
    public static String createTraceId() {
        String uuid = UUID.randomUUID().toString();
        return DigestUtils.md5Hex(uuid).substring(8, 24);
    }

    public static void setTraceIdIfAbsent() {
        if (MDC.get(TRACE_ID) == null) {
            MDC.put(TRACE_ID, createTraceId());
        }
    }
    // 省略了一些方法在后面会展示出来
}

DigestUtilsОт третьих лиц зависят:

<dependency>
	<groupId>commons-codec</groupId>
	<artifactId>commons-codec</artifactId>
    <version>***</version>
</dependency>

TRACE_IDставитьConstantУдобная ссылка в классе:

public class Constant {
    ...
   public static final String TRACE_ID = "traceId";
    ...
}

  1. Измените формат вывода в файле конфигурации журнала и увеличьте печать поля TraceID.

    Метод значения:%X{traceid}

image.png

результат

После выполнения вышеуказанных шагов журнал, распечатанный после того, как ваше веб-приложение получит запрос, будет включен.TraceId.

image.png

2. Столкнулся с проблемой потери TraceID пула потоков.

Предыдущее решение просто реализует наши самые основные потребности. Но если вы действительно используете его, вы обнаружите, что поток асинхронной задачи не захвачен.TraceIDиз.

Зрелое приложение определенно будет использовать много пулов потоков. Общие являются@AsyncПулы потоков для асинхронных вызовов, некоторые пулы потоков, определяемые самим приложением, и так далее.

Как упоминалось чуть ранее, MDC проходит черезInheritableThreadLocalРеализовано, при создании дочернего потока копируется свойство inheritableThreadLocals родительского потока. Но в пуле потоков потоки используются повторно, а не создаются заново, поэтому содержимое MDC не может быть передано.

Так что нам нужна кривая, чтобы спасти страну.Поскольку потоки используются повторно, мы, естественно, можем подумать о выполнении некоторых «дерзких» операций, когда задачи отправляются в пул потоков, чтобы содержимое MDC могло быть передано.

Модернизация

Просто поместите код сюда:

/**
 * @author Richard_yyf
 */
public class ThreadMdcUtil {
    
    public static String createTraceId() {
        String uuid = UUID.randomUUID().toString();
        return DigestUtils.md5Hex(uuid).substring(8, 24);
    }

    public static void setTraceIdIfAbsent() {
        if (MDC.get(TRACE_ID) == null) {
            MDC.put(TRACE_ID, createTraceId());
        }
    }

    public static void setTraceId() {
        MDC.put(TRACE_ID, createTraceId());
    }

    public static void setTraceId(String traceId) {
        MDC.put(TRACE_ID, traceId);
    }

    public static <T> Callable<T> wrap(final Callable<T> callable, final Map<String, String> context) {
        return () -> {
            if (context == null) {
                MDC.clear();
            } else {
                MDC.setContextMap(context);
            }
            setTraceIdIfAbsent();
            try {
                return callable.call();
            } finally {
                MDC.clear();
            }
        };
    }

    public static Runnable wrap(final Runnable runnable, final Map<String, String> context) {
        return () -> {
            if (context == null) {
                MDC.clear();
            } else {
                MDC.setContextMap(context);
            }
            setTraceIdIfAbsent();
            try {
                runnable.run();
            } finally {
                MDC.clear();
            }
        };
    }
}

Оберните собственное расширениеThreadPoolExecutor

/**
 * @author Richard_yyf
 */
public class ThreadPoolExecutorMdcWrapper extends ThreadPoolExecutor {

    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);
    }

    @Override
    public void execute(Runnable task) {
        super.execute(ThreadMdcUtil.wrap(task, MDC.getCopyOfContextMap()));
    }

    @Override
    public <T> Future<T> submit(Runnable task, T result) {
        return super.submit(ThreadMdcUtil.wrap(task, MDC.getCopyOfContextMap()), result);
    }

    @Override
    public <T> Future<T> submit(Callable<T> task) {
        return super.submit(ThreadMdcUtil.wrap(task, MDC.getCopyOfContextMap()));
    }

    @Override
    public Future<?> submit(Runnable task) {
        return super.submit(ThreadMdcUtil.wrap(task, MDC.getCopyOfContextMap()));
    }
}

использовать

Конкретное использование заключается в том, чтобы поместить ваш оригинальныйexecutor = new ThreadPoolExecutor(...)изменить наexecutor = new ThreadPoolExecutorMdcWrapper(...)Вот и все.

Например, если вы используете Spring@AsyncДля асинхронных методов объявите это при настройке пула потоков:

@SpringBootApplication
public class Application {

    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }

    @EnableAsync
    @Configuration
    class TaskPoolConfig {

        @Bean("taskExecutor")
        public Executor taskExecutor() {
            ThreadPoolTaskExecutor executor = new ThreadPoolExecutorMdcWrapper();
            executor.setCorePoolSize(10);
            executor.setMaxPoolSize(20);
            executor.setQueueCapacity(200);
            executor.setKeepAliveSeconds(60);
            return executor;
        }
    }

}

результат

В соответствии с приведенными выше шагами, когда ваша асинхронная задача печатает журнал, она выводит первоначально запрошенный TraceID.

image.png

3. Отслеживание ссылок Dubbo RPC

Наша проектная группа в основном использует Dubbo для разработки микросервисной инфраструктуры. Мы хотим передать вышестоящему сервисуTraceID, чтобы добиться эффекта отслеживания ссылок.

Dubbo предоставляет такой механизм, который может бытьDubbo RPC + Dubbo Filterнастроить и передать потребителюTraceID.

Пожалуйста, обратитесь к официальному сайту для описания этих двух концепций.

Dubbo RPC
Dubbo Filter

Здесь я непосредственно привожу код и конфигурацию точки расширения.

Dubbo Filter for Consumer

Сторона потребительского приложения:

/**
 * @author Richard_yyf
 */
@Activate(group = {Constants.CONSUMER})
public class ConsumerRpcTraceFilter implements Filter {

    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        //如果MDC上下文有追踪ID，则原样传递给provider端
        String traceId = MDC.get(TRACE_ID);
        if (StringUtils.isNotEmpty(traceId)) {
            RpcContext.getContext().setAttachment(TRACE_ID, traceId);
        }
        return invoker.invoke(invocation);
    }

}

Конфигурация SPI:

существуетresourcesкаталог, создать/META-INF/dubbo/com.alibaba.dubbo.rpc.Filterдокумент.

consumerRpcTraceFilter=com.xxx.xxx.filter.ConsumerRpcTraceFilter

image.png

Dubbo Filter for Provider

Сторона приложения поставщика услуг:

/**
 * @author Richard_yyf
 */
@Activate(group = {Constants.PROVIDER})
public class ProviderRpcTraceFilter implements Filter {
    
    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        // 接收消费端的traceId
        String traceId = RpcContext.getContext().getAttachment(TRACE_ID);
        if (StringUtils.isBlank(traceId)) {
            traceId = ThreadMdcUtil.createTraceId();
        }

        // 设置日志traceId
        ThreadMdcUtil.setTraceId(traceId);

        // TODO 如果这个服务还会调用下一个服务，需要再次设置下游参数
        // RpcContext.getContext().setAttachment("trace_id", traceId);

        try {
            return invoker.invoke(invocation);
        } finally {
            // 调用完成后移除MDC属性
            MDC.remove(TRACE_ID);
        }
    }

}

Конфигурация SPI:

providerRpcTraceFilter=com.xxx.xxx.filter.ProviderRpcTraceFilter

4. Отслеживание ссылок для службы HTTP

В дополнение к методу Dubbo RPC вызовы между общими микросервисами также выполняются через HTTP REST. В этом сценарии вышестоящая служба должна автоматически отправлять HTTP-вызов вышестоящей службе.TraceIDДобавлено в заголовок HTTP.

Возьмем в качестве примера обычно используемый Spring RestTemplate, используйте перехватчики для переноса HTTP-заголовка.

        RestTemplate restTemplate = new RestTemplate();

        List<ClientHttpRequestInterceptor> list = new ArrayList<>();
        list.add((request, body, execution) -> {
            String traceId = MDC.get(TRACE_ID);
            if (StringUtils.isNotEmpty(traceId)) {
                request.getHeaders().add(TRACE_ID, traceId);
            }
            return execution.execute(request, body);
        });

        restTemplate.setInterceptors(list);

Поскольку нисходящие службы доступны через интерфейс HTTP, для их получения рекомендуется добавить перехватчик.

public class TraceInterceptor extends HandlerInterceptorAdapter {

    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) {
        MDC.clear();
        String traceId = request.getHeader(TRACE_ID);
        if (StringUtils.isEmpty(traceId)) {
            ThreadMdcUtil.setTraceId();
        } else {
            MDC.put(TRACE_ID, traceId);
        }
        return true;
    }
}

5. Думаем о реализации Spring Cloud Sleuth

После вышеперечисленных шагов мы сами сформировали относительно простое решение для отслеживания услуг.

В качестве универсальной среды разработки микросервисов Spring Cloud предоставляет Spring Cloud Sleuth в качестве решения для распределенной трассировки в рамках этой технологической системы. Здесь я хочу взять его и рассказать об этом.

Sleuth — это зрелое техническое решение, основанное на Google Dapper в качестве теоретической основы, и некоторые термины в нем взяты из этой статьи. заTraceIDЧто касается проблемы передачи, то некоторые идеи решения проблемы простой версии решения, о которых мы упоминали выше, на самом деле также отражены в Sleuth.

Первый — распределенная трассировка, Сыщик будетSpanIDиTraceIDДобавьте в Slf4J MDC, чтобы на распечатанном логе был соответствующий логотип.

При столкновении с проблемой сбоя передачи TraceID пула потоков мы достаточно заворачиваем операцию отправки задач, а в Slueth это достигается реализациейHystrixConcurrencyStrategyинтерфейс для решенияTraceIDПроблема с асинхронной доставкой. Когда Hystrix действительно вызывается, он вызываетHystrixConcurrencyStrategyизwrapCallableметод. Реализуя этот интерфейс, вwrapCallableгенерал-лейтенантTraceIDхранится (см.SleuthHystrixConcurrencyStrategy).

При вызове Dubbo RPC и Http Service мы передаем Dubbo RpcContext + Filter и Http Header + Interceptor через протокол или точку расширения и механизм контекста, предоставленный самой инфраструктурой, для передачиTraceID. при отслеживании в Spring Cloud Sleuth@Async,RestTemplate,Zuul,FeignПри ожидании комплектующих тоже аналогичное решение. такие как отслеживаниеRestTemplateОн заимствует механизм Interceptor Spring Client, как указано выше (@SeeTraceRestTemplateInterceptor).

Выше приведено сравнение нашего простого решения со Spring Cloud Sleuth, чтобы показать, что идея отслеживания логов и некоторые технические решения схожи.

Конечно, Spring Cloud Sleuth реализован на основе Dapper и обеспечивает относительно зрелую архитектуру отслеживания вызовов распределенной системы.После интеграции зависимостей ZipKin + spring-cloud-sleuth-zipkin можно построить полный сбор данных, хранение данных и функцию отображения данных. система отслеживания услуг.

С помощью Sleuth вы можете четко понять, через какие сервисы прошел запрос на обслуживание и сколько времени занимает обработка каждого сервиса. Это позволяет нам легко разобраться в связи вызовов между микросервисами. Дополнительно Сыщик может помочь нам:

  • Анализ, занимающий много времени: с помощью Sleuth вы можете легко понять, сколько времени занимает каждый запрос на выборку, чтобы проанализировать, какие вызовы службы требуют больше времени;
  • Визуальные ошибки: Исключения, которые не перехвачены программой, можно увидеть в интерфейсе встроенного сервиса Zipkin;
  • Оптимизация ссылок: для часто вызываемых служб можно реализовать некоторые меры по оптимизации этих служб.

PS: Начиная с spring-cloud-sleth 2.0Официальная поддержка Dubbo, идея состоит в том, чтобы расширить механизм через фильтр Даббо.

резюме

Давайте поговорим о том, почему бы не представить решение Sleuth + ZipKin? Поскольку цепочка вызовов в нашей системе несложная, обычно существует только один уровень взаимосвязи вызовов, поэтому мы не хотим добавлять сторонние компоненты и предпочитаем использовать простые решения.

Эта статья заканчивается здесь. Реализовать простое лог-решение для коллтрекинга микросервисов не так уж сложно, главное — решить задачу и изучить несколько отличных технических решений на рынке.

Если эта статья была вам полезна, я надеюсь поставить лайк, это самая большая мотивация для меня.

Категории