小编lka*_*ris的帖子

我正在使用 Postgre SQL 来玩 R2DBC。我正在尝试的用例是通过 ID 以及语言、演员和类别获取电影。下面是架构

这是ServiceImpl中对应的一段代码

@Override
public Mono<FilmModel> getById(Long id) { 
    Mono<Film> filmMono = filmRepository.findById(id).switchIfEmpty(Mono.error(DataFormatException::new)).subscribeOn(Schedulers.boundedElastic());
    Flux<Actor> actorFlux = filmMono.flatMapMany(this::getByActorId).subscribeOn(Schedulers.boundedElastic());
    Mono<String> language = filmMono.flatMap(film -> languageRepository.findById(film.getLanguageId())).map(Language::getName).subscribeOn(Schedulers.boundedElastic());
    Mono<String> category = filmMono.flatMap(film -> filmCategoryRepository
                    .findFirstByFilmId(film.getFilmId()))
            .flatMap(filmCategory -> categoryRepository.findById(filmCategory.getCategoryId()))
            .map(Category::getName).subscribeOn(Schedulers.boundedElastic());

    return Mono.zip(filmMono, actorFlux.collectList(), language, category)
            .map(tuple -> {
                FilmModel filmModel = GenericMapper.INSTANCE.filmToFilmModel(tuple.getT1());
                List<ActorModel> actors = tuple
                        .getT2()
                        .stream()
                        .map(act -> GenericMapper.INSTANCE.actorToActorModel(act))
                        .collect(Collectors.toList());
                filmModel.setActorModelList(actors);
                filmModel.setLanguage(tuple.getT3());
                filmModel.setCategory(tuple.getT4());
                return filmModel;
            });
         }

日志显示 4 次调用拍摄

2021-12-16 21:21:20.026 DEBUG 32493 --- [ctor-tcp-nio-10] o.s.r2dbc.core.DefaultDatabaseClient     : Executing …

我有很多用户（150-2亿）。每个用户有 N(30-100) 个属性。该属性可以是整数、文本或时间戳类型。属性是未知的，所以我想动态地、即时地添加它们。

解决方案 1 - 通过更改表添加新列

   CREATE TABLE USER_PROFILE(
    UID uuid PRIMARY KEY,
    LAST_UPDATE_DATE TIMESTAMP,
    CREATION_DATE TIMESTAMP
   );

对于每个新属性：

    ALTER TABLE USER_PROFILE ADD AGE INT;
    INSERT INTO USER_PROFILE ( UID, LAST_UPDATE_DATE, CREATION_DATE, AGE) VALUES ('01f63e8b-db53-44ef-924e-7a3ccfaeec28', 2021-01-12 07:34:19.121, 2021-01-12 07:34:19.121, 27);

解决方案 2 - 固定架构：

CREATE TABLE USER_PROFILE(
    UID uuid,
    ATTRIBUTE_NAME TEXT,
    ATTRIBUTE_VALUE_TEXT TEXT,
    ATTRIBUTE_VALUE_TIMESTAMP TIMESTAMP,
    ATTRIBUTE_VALUE_INT INT,
    LAST_UPDATE_DATE TIMESTAMP,
    CREATION_DATE TIMESTAMP,
    PRIMARY KEY (UID, ATTRIBUTE_NAME)
);

对于每个新属性：

INSERT INTO USER_PROFILE ( UID, ATTRIBUTE_NAME, ATTRIBUTE_VALUE_INT, LAST_UPDATE_DATE, CREATION_DATE) VALUES ('01f63e8b-db53-44ef-924e-7a3ccfaeec28', 'age', …

我有微服务应用程序。为了进行协作，每个服务都使用异步消息传递。我知道，spring data jpa默认使用乐观锁。但是，如果这种锁定方法不是由用户调用，而是由另一个服务调用（在我的示例中，有验证服务，可以将对象的状态更新为有效或无效），我想处理异常并重试更新对象。我还必须在这个微服务上使用 webflux 堆栈。现在我有这样的代码：

public void updateStatus(String id, EventStatus status) {
    eventRepository.findById(id)
            .doOnNext(eventDocument -> {
                eventDocument.setStatus(status);
                eventRepository.save(eventDocument).subscribe();
            }).doOnError(OptimisticLockingFailureException.class, exception -> { //Retry in 2 sec if optimistic lock occurs on update
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException ex) {
                    ex.printStackTrace();
                }
                updateStatus(id, status);
            })
            .subscribe();
}

我不喜欢这里的递归。对此有更好的解决方案吗？

假设我有一个Flux包含许多（数十亿个字符串）的输入，如下所示：

• 苹果
• 应用
• 圣经
• 书

这样的字符串有数十亿个，它们无法放入内存中，这就是我想使用反应式方法的原因。

流已排序。现在我想要的是通过前 3 个字符创建一系列有序字符串组：

• 应用程序：苹果，应用程序
• 围兜：圣经
• 嘘：书

这Flux最终会出现 HTTP 响应，这意味着所有“app”项目必须在“bib”项目开始之前输出。

如果不使用，Flux我可以使用有序属性并将项目收集到准备好的存储桶中（每个存储桶的字符串数量将适合内存） - 每当前缀发生变化时，我将刷新存储桶并开始收集新的前缀。有序流的一大优点是我知道一旦遇到新的前缀，旧的就不会再出现了。

但使用Flux我不知道如何做到这一点。将会.groupBy()返回Flux，Flux但我认为在尝试将其序列化到 HTTP 响应输出流时这不会起作用。

我有一个如下所示的 Mono：

    private void getNodeDetail() {
        Mono<String> mono = webClient.get()
                .uri("/alfresco/api/-default-/public/alfresco/versions/1/nodes/f37b52a8-de40-414b-b64d-a958137e89e2")
                .retrieve().bodyToMono(String.class);

        System.out.println(mono.subscribe());
        System.out.println(mono.block());
    }

问题：第一个 sysout 向我展示了reactor.core.publisher.LambdaSubscriber@77114efe使用 block() 时的情况，它向我展示了我需要的内容（json 字符串）。但我想使用 Aysnc 方法。那么，如上所述，这是否意味着我的目标系统（在本例中为 Alfresco）不支持异步调用？subscribe()如果不是这种情况，如何使用，就像 一样以字符串格式在控制台上打印响应block()？