java - react 器 Mono.defer 调用上的时间戳未更新

Question

我有一项服务，可以提供数据库中所有现有股票代码的汇率列表。调用非常简单，当调用服务的 rates() 方法时，会以以下类的形式给出一个随机费率:

public class Rate {
  private final long timestamp = Instant.now().toEpochMilli();
  private String symbol;
  private final double rate = (Math.random() * 100).toBigDecimal().setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue();

  // getters and setters
}

服务方法实现是:

public Flux<Rate> rates() {
  return service.findAll();
}

我有一个响应式(Reactive) WebSocketHandler，如下所示:

@Service
public class MyHandler implements WebSocketHandler {

    private final ObjectMapper objectMapper;
    private final RateService service;
    private final ApplicationProperties applicationProperties;

    public MyHandler(ObjectMapper objectMapper, RateService service, ApplicationProperties applicationProperties) {
        this.objectMapper = objectMapper;
        this.service = service;
        this.applicationProperties = applicationProperties;
    }

    @Override
    public Mono<Void> handle(WebSocketSession session) {
        return session.send(
                Mono.defer(() -> service.rates().collectList())
                        .repeat()
                        .delayElements(Duration.ofSeconds(applicationProperties.getInterval()))
                        .map(value -> {
                            try {
                                return objectMapper.writeValueAsString(value);
                            } catch (JsonProcessingException e) {
                                return null;
                            }
                        })
                        .map(session::textMessage)
        );
    }
}

当我通过客户端连接到 WebSocket 时，我每 5 秒就会收到一个费率列表，这很好。我遇到的唯一问题是 timestamp 属性是每 5 秒完全相同的时间戳。所以我通过 WebSocket 的第一个响应是这样的: [{"timestamp":"2018-10-08T09:52:51.387Z","symbol":"GOOG","rate":33.91}] ，第二个响应是 [{"timestamp":"2018-10-08T09:52:51.387Z","symbol":"GOOG","rate":51.43}]。所以速率已经改变，但时间戳没有改变。

我可以通过覆盖 rate 字段的 getter 来解决它，但我希望不要在该层中解决它。我只是希望 getter 返回值而不进行操作。

Answer 1

几乎正确，要理解，请运行下面的命令，有一些内容出现在错误的位置，我建议您首先查看使用“Creating Rate”发出的时间戳。

在您创建列表并重复、重复、重复之前，因此在将其发送到客户端之前已创建并存储该列表，因此如果您延迟创建过程而不是发送过程，则将每 5 创建一次费率秒并发射。

public class MyHandler  {

    private final ObjectMapper objectMapper = new ObjectMapper();

    public Flux<String> handle() {
        return Mono.defer(() -> rates().delayElements(Duration.ofSeconds(5)).collectList())
                .repeat()
                .map(value -> {
                    try {
                        return objectMapper.writeValueAsString(value);
                    } catch (JsonProcessingException e) {
                        return null;
                    }
                });
    }

    private Flux<Rate> rates() {
        return Flux.just(new Rate("a"), new Rate("b"), new Rate("c"));
    }

    public static void main(String[] args) {
        new MyHandler().handle()
                .subscribe(str -> System.out.println(str + ", time now - " + Instant.now().toEpochMilli()));
        try {
            Thread.sleep(60_000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    static class Rate {
        private final long timestamp;
        private String symbol;
        private final double rate;

        public Rate(String symbol) {
            this.symbol = symbol;
            timestamp = Instant.now().toEpochMilli();
            rate  = new BigDecimal(Math.random() * 100).setScale(2, RoundingMode.HALF_UP).doubleValue();
            System.out.println("Creating Rate " + timestamp);
        }
        // getters and setters

        public long getTimestamp() {
            return timestamp;
        }
        public String getSymbol() {
            return symbol;
        }
        public void setSymbol(String symbol) {
            this.symbol = symbol;
        }
        public double getRate() {
            return rate;
        }
    }
}