Spring Boot RSocket中@MessageMapping返回响应元数据及RSocketRequester接收方法问询

有两种常用的实现方式可以在@MessageMapping注解的方法中返回响应元数据：

• 方式一：返回org.springframework.messaging.Message<T>类型，将元数据存放在消息头中，Spring会自动将消息头转换为RSocket响应的元数据部分
  代码示例：

  @MessageMapping("query.resource")
  public Message<ResourceDTO> queryResource(RequestDTO request) {
      // 业务逻辑构造响应载荷
      ResourceDTO payload = getResourceById(request.getResourceId());
      // 构造元数据
      Map<String, Object> metadata = Map.of(
          "request-trace-id", request.getTraceId(),
          "resource-expire-time", System.currentTimeMillis() + 3600 * 1000
      );
      return MessageBuilder.withPayload(payload)
          .copyHeaders(metadata)
          .build();
  }
  

• 方式二：直接返回RSocket原生的io.rsocket.Payload对象，手动拼接数据和元数据部分，适合需要底层自定义控制的场景，需注意自行处理编解码逻辑。

默认的RetrieveSpec检索方法仅返回解码后的载荷，可通过以下两种方式获取完整的响应元数据：

• 方式一：指定接收类型为Payload，手动解析元数据和载荷
  代码示例：

  @Autowired
  private RSocketRequester rSocketRequester;
  @Autowired
  private RSocketStrategies rsocketStrategies;
  
  public Mono<CustomResult> requestResource(RequestDTO req) {
      return rSocketRequester.route("query.resource")
          .data(req)
          .retrieveMono(Payload.class)
          .map(payload -> {
              // 解析复合元数据
              CompositeMetadata metadata = CompositeMetadata.from(payload.metadata(), false);
              // 解码载荷数据
              ResourceDTO data = rsocketStrategies.dataDecoder()
                  .decode(payload.data(), ResolvableType.forClass(ResourceDTO.class),
                      MimeTypeUtils.APPLICATION_JSON, Collections.emptyMap());
              // 封装包含元数据和载荷的自定义结果
              return new CustomResult(data, metadata);
          });
  }
  

• 方式二：使用exchangeToMono/exchangeToFlux方法获取完整响应上下文，Spring会自动完成元数据的解码，可直接从响应头中读取元数据
  代码示例：

  public Mono<CustomResult> requestResource(RequestDTO req) {
      return rSocketRequester.route("query.resource")
          .data(req)
          .exchangeToMono(response -> {
              // 直接读取解码后的元数据
              String traceId = response.headers().getFirst("request-trace-id");
              Long expireTime = response.headers().getFirst("resource-expire-time", Long.class);
              // 读取响应载荷
              return response.bodyToMono(ResourceDTO.class)
                  .map(data -> new CustomResult(data, traceId, expireTime));
          });
  }
  

注意事项 #

• 自定义格式的元数据需要提前在RSocketStrategies中注册对应的编解码器，保证两端序列化/反序列化逻辑一致
• 使用复合元数据时，需要确保服务端和客户端的元数据MIME类型配置匹配，避免解析异常

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