在 couchbase java sdk 2.7.15 上使用异步存储桶更新文档答案

【问题标题】：Upserting documents with async bucket on couchbase java sdk 2.7.15在 couchbase java sdk 2.7.15 上使用异步存储桶更新文档
【发布时间】：2021-03-01 10:14:22
【问题描述】：

我有以下代码使用 Observables 将一组 pojos（要转换为 json 文档）插入到 couchbase 中：

    public <T> long  batchUpsert(Iterable<T> items, Function<T, JsonDocument> docCreator, Bucket couchbaseBucket) {

    AtomicLong counter = new AtomicLong();

    AsyncBucket asyncBucket = couchbaseBucket.async();
    Observable<JsonDocument> observableFromDocs =
            Observable
                    .from(items)
                    .map(elem -> docCreator.apply(elem))
                    .filter(elem -> elem!=null)//skip creating problematic docs. logging their info for troubleshooting
                    .flatMap(elem -> upsertDocument(elem, asyncBucket))
                    .retryWhen(
                            RetryBuilder.anyOf(BackpressureException.class, Exception.class)
                                    .doOnRetry((Integer integer, Throwable throwable, Long aLong, TimeUnit timeUnit) ->
                                            log.error("Retrying load. Attempt {} For exception {}", integer,throwable.toString())
                                    )
                                    .delay(Delay.exponential(TimeUnit.MILLISECONDS, RETRY_DELAY_CEILING))
                                    .max(MAX_RETRIES)
                                    .build()
                    );

    observableFromDocs.subscribe(
            (elem)-> {},
            elem -> log.error("Document insertion failure", elem),
            () -> {counter.incrementAndGet();log.debug("Completed ASYNC load ");});

    return counter.get();
}

这段代码运行良好，创建了文档，并且对 upsert 进行了异步调用，但没有文档被 upsert 到 couchbase，它静默失败，并且没有记录异常，几乎就像线程在内部死亡一样，有人可以指出我究竟做错了什么？我要拔毛了... :)

我现在验证它对于 1 个项目的集合失败，有人可以告诉我为什么会这样吗？

【问题讨论】：

标签： rx-java couchbase couchbase-java-api

【解决方案1】：

我认为您的代码不能等待所有异步操作完成，所以这种情况导致了问题官方sdkdock有一个批处理操作的例子，例子说等待最后一个操作并调用块

【讨论】：