Spring Integration+Micrometer：DirectChannel上下文传播失效求助

问题分析 #

你在JPA Repository添加@Observed注解后，每次调用都会生成新Trace，说明链路上下文没有从Spring Integration的消息处理流程正确传播到JPA操作中。虽然配置了@GlobalChannelInterceptor的ObservationPropagationChannelInterceptor，但自定义的DirectChannel可能没有正确应用该拦截器。

解决方案 #

1. 手动为DirectChannel添加拦截器 #

全局拦截器有时无法自动应用到手动创建的DirectChannel，直接在创建Channel时显式注入并添加拦截器：

@Bean(name = CHANNEL_NAME)
public MessageChannel myInputChannel(ObservationPropagationChannelInterceptor observationPropagationChannelInterceptor) {
    DirectChannel directChannel = new DirectChannel();
    directChannel.setComponentName(CHANNEL);
    directChannel.setDatatypes(ConsumerRecord.class);
    // 手动添加观察上下文传播拦截器
    directChannel.addInterceptor(observationPropagationChannelInterceptor);
    return directChannel;
}

2. 验证拦截器是否生效 #

在ObservationPropagationChannelInterceptor的方法中添加日志，确认消息处理时拦截器被调用：

// 在拦截器Bean中扩展日志逻辑
@Override
public Message<?> preSend(Message<?> message, MessageChannel channel) {
    log.debug("Propagating observation context for channel: {}", channel.getComponentName());
    return super.preSend(message, channel);
}

3. 确保JPA调用在消息处理的同步上下文内 #

如果JPA操作是异步执行（比如用@Async、手动开线程），会丢失当前Observation上下文。需要：

• 尽量在消息处理的主线程中调用JPA方法
• 若必须异步，手动传递Observation上下文：

  // 在消息处理器中获取当前Observation
  Observation currentObservation = Observation.getCurrentObservation();
  // 异步执行时绑定上下文
  CompletableFuture.runAsync(() -> {
      try (Observation.Scope scope = currentObservation.openScope()) {
          // 此处调用JPA Repository方法
          myRepository.save(entity);
      }
  });
  

4. 确认ObservationRegistry的一致性 #

确保整个应用中只有一个ObservationRegistry Bean，避免因多实例导致上下文无法共享。可以在消息处理器和JPA Repository类中注入ObservationRegistry，打印其哈希值验证是否为同一实例。

5. 验证Trace ID一致性 #

在消息监听方法和JPA Repository方法中分别打印当前Trace ID，确认是否一致：

// 消息监听方法
@ServiceActivator(inputChannel = CHANNEL_NAME)
public void handleMessage(ConsumerRecord<String, String> record) {
    String traceId = Observation.getCurrentObservation()
        .map(obs -> obs.getContext().getTraceId())
        .orElse("NO_TRACE_ID");
    log.info("Message handling trace ID: {}", traceId);
    // 调用JPA方法
    myRepository.findBySomeField(record.value());
}

// JPA Repository方法
@Observed
public MyEntity findBySomeField(String field) {
    String traceId = Observation.getCurrentObservation()
        .map(obs -> obs.getContext().getTraceId())
        .orElse("NO_TRACE_ID");
    log.info("JPA query trace ID: {}", traceId);
    // 查询逻辑
}

如果两个Trace ID一致，说明上下文传播成功；若不一致，检查拦截器是否生效、是否存在异步执行导致上下文丢失的情况。

内容的提问来源于stack exchange，提问作者Suyash Soni