从 RequestContext 或类似的访问 StreamListener headers

Question

我有一个调用十几个其他服务的服务。这使用控制器 class 中的 @StreamListener 从 Kafka 主题读取。出于可追溯性目的，来自 Kafka 消息的相同 headers（原始请求 ID）也需要转发到所有其他服务

传统上，使用 @PostMapping("/path") 或 GetMapping，会生成一个请求上下文，并且可以使用 RequestContextHolder.currentRequestAttributes() 从任何地方访问 headers，我会通过每当我需要拨打外部电话时，HttpHeaders object 变成 RequestEntity

但是在 StreamListener 中，没有生成请求上下文并且尝试访问 RequestContextHolder 导致异常

这是我尝试执行的操作的示例，结果出现异常：

public class Controller {
  @Autowired Service1 service1
  @Autowired Service2 service2

  @StreamListener("stream")
  public void processMessage(Model model) {
    service1.execute(model);
    service2.execute(model);
  }
}

public class Service {
  RestTemplate restTemplate;

  public void execute(Model model){
    // Do some stuff

    HttpHeaders httpHeaders = RequestContextHolder.currentRequestAttributes().someCodeToGetHttpHeaders();
    HttpEntity<Model> request = new HttpEntity(model, httpHeaders);
    restTemplate.exchange(url, HttpMethod.POST, request, String.class);
  }
}

我目前的解决方法是将 StreamListener 更改为 PostMapping 并让另一个 PostMapping 调用它以便生成请求上下文。另一种选择是使用 ThreadLocal 但它看起来就像 janky

我知道 @Headers MessageHeaders 注释可以访问流 headers，但是，如果不将 headers 传递给每个服务和会影响很多单元测试

理想情况下，我需要一种方法来创建我自己的请求上下文（或任何适当的术语），以便有一个地方存储请求范围 objects（HttpHeader）或另一个线程让请求 headers 向下传递堆栈而不向 service.execute

添加请求参数的安全方法

Answer 1

我找到了一个解决方案，我把它留在这里给其他试图实现类似目标的人

如果您的目标是通过 REST 控制器和流侦听器转发一堆 headers end-to-end，您可能需要考虑使用 Spring Cloud Sleuth

通过你的maven或gradle配置将它添加到你的项目中：

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-sleuth</artifactId>
</dependency>

具体来说，在 Spring Cloud Sleuth 中有一项功能可以转发 headers 或 "baggage"，方法是在 [=35] 中设置属性 spring.sleuth.propagation-keys =].这些 key-value 对在整个跟踪过程中持续存在，包括也实现相同传播密钥的任何下游 http 或流调用

如果需要在代码级别访问这些字段，您可以使用 ExtraFieldPropagation 静态函数获取和设置它们：

ExtraFieldPropagation.set("country-code", "FO"); // Set
String countryCode = ExtraFieldPropagation.get("country-code"); // Get

请注意，ExtraFieldPropagation setter 无法设置属性不存在于已定义的 spring.sleuth.propagation-keys 中，因此不会接受任意键

您可以阅读 documentation 了解更多信息

从 RequestContext 或类似的访问 StreamListener headers

Accessing StreamListener headers from RequestContext or similar

java

spring

spring-boot

apache-kafka-streams

requestcontext