带有 couchbase 的 rxjava 是否为非批量操作提供价值答案

【问题标题】：Does rxjava with couchbase offer value for non-bulk opertions带有 couchbase 的 rxjava 是否为非批量操作提供价值
【发布时间】：2015-02-25 12:45:11
【问题描述】：

新的 Couchbase SDK 使批量操作更容易使用，使用 rx-java 的性能更高。但是使用 rx 对单个值进行操作有什么价值吗？

如果我们看一个简单的 CAS / 插入操作，即如果值存在，则执行 CAS，否则执行插入并返回文档值

    final String id = "id";
    final String modified = "modified";
    final int numCasRetries = 3;

    Observable
        .defer(() -> bucket.async().get(id))
        .flatMap(document -> {
            try {
                if (document == null) {
                    JsonObject content = JsonObject.create();
                    content.put(modified,  new Date().getTime());
                    document = bucket.insert(JsonDocument.create(id, content));
                } else {
                    document.content().put(modified, new Date().getTime());
                    document = bucket.replace(document);
                }
                return Observable.just(document);
            } catch (CASMismatchException e) {
                return Observable.error(e);
            }
        })
        .retry((count, error) -> {
            // Only retry on CASMismatchException
            return ((error instanceof CASMismatchException)
                    && (count < numCASRetries));
        })
         .onErrorResumeNext(error -> {
            return Observable.error(new Exception(error));
        })
        .toBlocking()
        .single();

所以toBlocking 将阻塞调用线程，直到有结果可用。一次只能从 Couchbase 写入和读取一个值。所以我不明白为什么或者即使这段代码会比

    final String id = "id";
    final String modified = "modified";
    final int numCasRetries = 3;

    JsonDocument document = null;
    for (int i = 1; i <= numCasRetries; i++) {
        document = bucket.get(id);
        try {
            if (document == null) {
                JsonObject content = JsonObject.create();
                content.put(modified,  new Date().getTime());
                document = bucket.insert(JsonDocument.create(id, content));
            } else {
                document.content().put(modified, new Date().getTime());
                document = bucket.replace(document);
            } 
            return document;
        } catch (CASMismatchException e) {
            if (i == numCasRetries) {
                throw e;
            }
        }
    }

如果我认为在这种情况下，rx 方法的可读性较差。

【问题讨论】：

标签： java couchbase reactive-programming rx-java

【解决方案1】：

对于最终需要阻止的单个文档的操作，我倾向于同意您的第二个示例更清晰。

当您大量使用异步处理时，RxJava 会大放异彩，尤其是当您需要高级错误处理、重试场景、异步流程组合时...

上一代的 Couchbase Java SDK (1.4.x) 只是为此提供了 Future，它没有提供我们在 RxJava 中发现的优雅、强大和表达能力。

【讨论】：