小编Bri*_*zel的帖子

我想使用Micrometer记录异步方法最终发生时的执行时间。有推荐的方法吗？

示例：Kafka回复模板。我想记录实际执行sendAndReceive调用（在请求主题上发送消息并在回复主题上接收响应）所花费的时间。

    public Mono<String> sendRequest(Mono<String> request) {
        return request
            .map(r -> new ProducerRecord<String, String>(requestsTopic, r))
            .map(pr -> {
                pr.headers()
                        .add(new RecordHeader(KafkaHeaders.REPLY_TOPIC,
                                "reply-topic".getBytes()));
                return pr;
            })
            .map(pr -> replyingKafkaTemplate.sendAndReceive(pr))
            ... // further maps, filters, etc.

就像是

responseGenerationTimer.record(() -> replyingKafkaTemplate.sendAndReceive(pr)))

在这里不会工作；它仅记录创建所需的时间Supplier，而不是实际的执行时间。

我在 Spring Boot MVC 2.1 项目中使用 WebClient，发现客户端发出的第一个请求最多需要 6 秒。后续请求速度更快（~30 毫秒）。

Spring 的 JIRA中有一个已解决的问题，建议使用 Jetty 作为 WebClient Http 连接器。我已经尝试过这种方法，以大约 800 毫秒的第一个请求改进了数字。这次是一个改进，但距离通常需要 < 200 毫秒的 RestTemplate 还很远。

Netty 方法（5s 第一个请求）：

会议：

@Bean
public WebClient webClient() {
    return WebClient.create();
}

用法：

private final WebClient webClient;

@GetMapping(value="/wc", produces = APPLICATION_JSON_UTF8_VALUE)
public Mono<String> findWc() throws URISyntaxException {
    URI uri = new URI("http://xxx");
    final Mono<String> response = webClient.get().uri(uri).retrieve().bodyToMono(String.class);
    return response;
}

码头方法（800ms 第一个请求）：

会议：

@Bean
public JettyResourceFactory resourceFactory() {
    return …

在我使用Spring 4.1.x版本并拥有自己的Cors Filter和Interceptor之前.我也使用Tomcat 8.0.8一切都很好.

因为我已将相同的应用程序移动到Tomcat 8.0.23,CORS停止工作.

因此,我将应用程序更新为我的POM中的以下版本的Spring依赖项:

org.springframework: 4.2.0.RC2
org.springframework.security: 4.0.1.RELEASE
org.springframework.security.oauth2: 2.0.7.RELEASE
org.springframework.ws.version: 2.2.0.RELEASE
spring.data.commons.core.version: 1.4.1.RELEASE

现在,我还更新了我的securityConfiguration.xml,以获得安全4.0.1.RELEASE的正确模式.

应用程序运行正常,除CORS外,所有更改都很完美.

所以我按照说明操作并将以下内容添加到我的WebMvcConfigurerAdapter实现类中:

@Override
    public void addCorsMappings(CorsRegistry registry) {
        Properties prop = null;

        try {

            InputStream in = getClass().getResourceAsStream(
                    "/com/nando/config/cors.properties");
            prop = new Properties();
            prop.load(in);
            in.close();            
        } catch (IOException e) {
            prop = null;
        }

        if (prop != null) {
            String domains = prop.getProperty("allowed.origins");

            List<String> allowedOrigins = new ArrayList<String>(
                    Arrays.asList(domains.split(",")));

            if (allowedOrigins.size() > 0) {
                String[] arrayOrigins = null;
                arrayOrigins = new String[allowedOrigins.size()]; …

春季启动1.5.3.RELEASE有几类org.springframework.boot.context.embed,并在子包jetty,tomcat和undertow.但是,https://github.com/spring-projects/spring-boot/tree/v2.0.0.M1/spring-boot/src/main/java/org/springframework/boot/context不再包含子目录embed.

课程在哪里TomcatEmbeddedServletContainerFactory消失了？

在 tomcat 中将 Spring WebFlux 应用程序部署为 war 文件时遇到以下异常。使用 AbstractAnnotationConfigDispatcherHandlerInitializer 扩展并实现 getConfigClasses 方法，如下所示。

应用

@SpringBootApplication
public class TestSpringApplication extends AbstractAnnotationConfigDispatcherHandlerInitializer {

    public static void main(String[] args) {
        SpringApplication.run(TestSpringApplication.class, args);
    }
    @Override
    protected Class<?>[] getConfigClasses() {
        return new Class[]{
            WebConfig.class
        };
    }

控制器

@Controller
public class TestSpringController {

    @RequestMapping("/test")
    public Mono<String> hello(Model model){

         return Mono.just("account");
    }
}

注意：account.html 位于 resources/templates 下

例外

java.lang.IllegalStateException: Could not resolve view with name 'account'.
    at org.springframework.web.reactive.result.view.ViewResolutionResultHandler.lambda$resolveViews$3(ViewResolutionResultHandler.java:276) ~[spring-webflux-5.0.1.RELEASE.jar:5.0.1.RELEASE]
    at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:107) [reactor-core-3.1.1.RELEASE.jar:3.1.1.RELEASE]

我正在学习Flux和Mono，并在尝试检查Redis中是否不存在键时碰壁。我需要过滤通量，并且如果键存在，则要删除该元素。我在文档中找不到执行此操作的方法，我只能过滤是否存在该密钥。我需要相反。

client.request(MWS_EU, Orders, ordersRequest, ListOrdersResponse.class)
    .flatMapIterable(listOrdersResponse -> listOrdersResponse.getOrders())
    .filterWhen(order -> isNewOrder(order.getOrderId()))
    .flatMap(...)

和检查redis：

private Mono<Boolean> isNewOrder(String orderId) {
    return redisOrders.opsForValue().get(orderId).hasElement();
} 

有什么干净的方法可以过滤我的Flux以​​仅在键不存在时保留元素？

我不想用block()。

我可能错过了一些东西,但我无法弄清楚它是什么.

以下代码什么都不做:

webClient.get().uri("/some/path/here").retrieve()
     .bodyToMono(GetLocationsResponse.class)
     .doOnNext(System.out::println)
     .subscribe();

如果我试图阻止呼叫它工作正常:

webClient.get().uri("/some/path/here").retrieve()
      .bodyToMono(GetLocationsResponse.class)
      .doOnNext(System.out::println)
      .block();

奇怪的是,如果我"手动"创建一个Flux(即不是来自spring webClient),这很好用:

Flux.just("1", "2", "3")
    .filter(s -> !s.equals("2"))
    .doOnNext(System.out::println)
    .subscribe();

有人可以解释一下我做错了什么吗？是不是.subscribe()应该在第一种情况下执行操作,就像它在最后一样？

谢谢!

当 Flux 为空时，我试图返回 404，类似于此处：WebFlux 功能：如何检测空 Flux 并返回 404？

我主要担心的是，当您检查通量是否包含元素时，它会发出该值，而您会丢失它。当我尝试在服务器响应上使用 switch if empty 时，它永远不会被调用（我偷偷认为这是因为 Mono 不是空的，只有主体是空的）。

我正在做的一些代码（我的路由器类上有一个过滤器，检查 DataNotFoundException 以返回 notFound）：

Flux<Location> response = this.locationService.searchLocations(searchFields, pageToken);
return ok()
        .contentType(APPLICATION_STREAM_JSON)
        .body(response, Location.class)
        .switchIfEmpty(Mono.error(new DataNotFoundException("The data you seek is not here.")));

^这从不调用 switchIfEmpty

Flux<Location> response = this.locationService.searchLocations(searchFields, pageToken);

return response.hasElements().flatMap(l ->{
   if(l){
       return ok()
               .contentType(APPLICATION_STREAM_JSON)
               .body(response, Location.class);
   } 
   else{
       return Mono.error(new DataNotFoundException("The data you seek is not here."));
   }
});

^这会丢失 hasElements 上的发射元素。

有没有办法在 hasElements 中恢复发出的元素，或者让 switchIfEmpty 只检查主体的内容？

代码

package com.example.BLModel;

import org.hibernate.annotations.Type;

import javax.persistence.Column;
import javax.persistence.EmbeddedId;
import javax.persistence.Entity;
import javax.persistence.FetchType;
import javax.persistence.JoinColumn;
import javax.persistence.ManyToOne;
import javax.persistence.MapsId;
import javax.persistence.Table;

@Entity
@Table(name = "email_template_custom")
public class EmailTemplateCustom {
    @EmbeddedId
    private EmailTemplateCustomId id;

    @MapsId("tenantId")
    @ManyToOne(fetch = FetchType.LAZY, optional = false)
    @JoinColumn(name = "tenant_id", nullable = false)
    private Tenant tenant;

    @Column(name = "email_subject", nullable = false, length = 1000)
    private String emailSubject;

    @Column(name = "email_content")
    @Type(type = "org.hibernate.type.TextType")
    private String emailContent;

    public EmailTemplateCustomId getId() {
        return id;
    }

    public void setId(EmailTemplateCustomId …

我想尝试在 Spring Boot 3 Kafka 生产者和消费者之间进行跟踪。我按照https://spring.io/blog/2022/10/12/observability-with-spring-boot-3中的示例进行操作 ，当我在生产者和消费者之间使用resttemplate进行API调用时，会自动添加traceID，但是卡夫卡消息不携带它们。我使用了以下依赖项

 <dependency>
            <groupId>io.micrometer</groupId>
            <artifactId>micrometer-tracing-bridge-otel</artifactId>
        </dependency>

        <dependency>
            <groupId>io.opentelemetry</groupId>
            <artifactId>opentelemetry-exporter-zipkin</artifactId>
        </dependency>
     <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <version>3.0.0</version>
        </dependency>

卡夫卡生产者配置

@Bean
public ProducerFactory<String, String> producerFactory() {
    Map<String, Object> configProps = new HashMap<>();
    //typical properties 
    return new DefaultKafkaProducerFactory<>(configProps);
}

@Bean
public KafkaTemplate<String, String> kafkaTemplate(ProducerFactory<String, String> producerFactory) {
    KafkaTemplate<String, String> stringStringKafkaTemplate = new KafkaTemplate<>(producerFactory);
    stringStringKafkaTemplate.setObservationEnabled(true);//trying out with
    return stringStringKafkaTemplate;
}

发送消息

    kafkaTemplate.send("topic-1" , "message");

在消费者方面：

    @KafkaListener(topics = "topic-1", groupId = "group1")
    public void listenGroupFoo(String message) {
        logger.info("Received Message in …