You need to enable JavaScript to run this app. 如何为不同Apache Camel路由设置专属WebFlux WebClient配置?
优惠活动
大模型
产品
解决方案
定价
更多

如何为不同Apache Camel路由设置专属WebFlux WebClient配置?

现有路由构建中的WebClient实现逻辑

在路由构建实现中,项目在路由声明前就完成了WebClient的构建逻辑,具体代码如下:

@Override
public void configure() {
  createSubscription(activeProfile.equalsIgnoreCase("RESTART"));
  from(String.format("reactive-streams:%s", streamName))
      .to("log:camel.proxy?level=INFO&groupInterval=500000")
      .to(String.format("kafka:%s?brokers=%s", kafkaTopic, kafkaBrokerUrls));
}

private void createSubscription(boolean restart) {
  WebClient.builder()
      .defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.TEXT_XML_VALUE)
      .build()
      .post()
      .uri(initialRequestUri)
      .body(BodyInserters.fromObject(restart ? String.format(restartRequestBody, ZonedDateTime.now(ZoneId.of("UTC")).toString().replace("[UTC]", "")) : initialRequestBody))
      .retrieve()
      .bodyToMono(String.class)
      .map(initResp ->
          new JSONObject(initResp)
              .getJSONObject("RESPONSE")
              .getJSONArray("RESULT")
              .getJSONObject(0)
              .getJSONObject("INFO")
              .getString("SSEURL")
      )
      .flatMapMany(url -> {
        log.info(url);
        return WebClient.create()
            .get()
            .uri(url)
            .retrieve()
            .bodyToFlux(new ParameterizedTypeReference<ServerSentEvent<String>>() {
            })
            .flatMap(sse -> {
                  val data = new JSONObject(sse.data())
                      .getJSONObject("RESPONSE")
                      .getJSONArray("RESULT")
                      .getJSONObject(0)
                      .getJSONArray(apiName);
                  val list = new ArrayList<String>();
                  for (int i = 0; i < data.length(); i++) {
                    list.add(data.getJSONObject(i).toString());
                  }
                  return Flux.fromIterable(list);
                }
            );
        }
      )
      .onBackpressureBuffer()
      .flatMap(msg -> camelReactiveStreamsService.toStream(streamName, msg, String.class))
      .doFirst(() -> log.info(String.format("Reactive stream %s was %s", streamName, restart ? "restarted" : "started")))
      .doOnError(err -> {
        log.error(String.format("Reactive stream %s has terminated with error, restarting", streamName), err);
        createSubscription(true);
      })
      .doOnComplete(() -> {
        log.warn(String.format("Reactive stream %s has completed, restarting", streamName));
        createSubscription(true);
      })
      .subscribe();
}
现有实现的缺陷

从现有实现逻辑来看,当前的WebClient配置是全局作用于整个Spring Boot应用的,并未和特定Apache Camel路由的ID、目标URL做绑定关联。这就导致后续新增的、对接不同URL接口、需要自定义请求头/初始请求参数的响应式流路由,也会被这套通用WebClient配置影响,并不符合实际业务需求。

核心咨询问题
  • 是否可以实现和特定Apache Camel路由关联的专属WebClient配置,而非全局生效,让配置仅作用于对应的绑定路由?
  • 这类配置是否支持通过Spring DSL实现?

内容的提问来源于stack exchange,提问作者Eljah

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.11.09 01:27:03