java异步删除密钥以特定字符串开头的文档

Question

我试图从我的桶中删除所有密钥以某个字符串开头的文档

我的观点是：

private void createAndPublishCacheView()
{
 List<View> views = new ArrayList<View>();
 views.add(DefaultView.create("cache_view_by_key", "function (doc, meta) { emit(meta.id, null) }"));

 DesignDocument designDoc   = DesignDocument.create("cache_doc", views);
 this.cacheBucket.bucketManager().upsertDesignDocument(designDoc);
}

我的removeAll方法是（如果文档的键以'callId'开头，则删除文档）：

public void removeAll(String callId) throws Exception
{
 final CountDownLatch latch = new CountDownLatch(1);
 ViewQuery query            = cacheView.startKeyDocId(callId);

 bucket.async()
    .query(query)
    .doOnNext(new Action1<AsyncViewResult>() 
              {
               @Override
               public void call(AsyncViewResult viewResult) 
               {
                if (!viewResult.success()) throw new RuntimeException();
               }
              }
             )
    .flatMap(new Func1<AsyncViewResult, Observable<AsyncViewRow>>() 
             {
              @Override
              public Observable<AsyncViewRow> call(AsyncViewResult viewResult) 
              {
               return viewResult.rows();
              }
             }
            )
    .subscribe(new Subscriber<AsyncViewRow>() 
               {
                @Override
                public void onCompleted() { latch.countDown(); }

                @Override
                public void onError(Throwable t) { throw new RuntimeException(t.getMessage()); }

                @Override
                public void onNext(AsyncViewRow viewRow) 
                { 
                 System.out.println("viewRow.id() = " + viewRow.id());
                 bucket.remove(viewRow.id());
                }
               }
              );

 latch.await();
 }

我是初学者。如果onNext方法为：

  public void onNext(AsyncViewRow viewRow) 
                { 
                 System.out.println("viewRow.id() = " + viewRow.id());
                 bucket.remove(viewRow.id());
                }

然后控制台上只打印并删除了几个文档，我收到一个：

java.lang.RuntimeException: java.util.concurrent.TimeoutException
at com.couchbase.client.java.util.Blocking.blockForSingle(Blocking.java:75)
at com.couchbase.client.java.CouchbaseBucket.remove(CouchbaseBucket.java:486)
at com.couchbase.client.java.CouchbaseBucket.remove(CouchbaseBucket.java:466)
at com.ventusproxy.tests.couchbase.TestCB$3.onNext(TestCB.java:167)
at com.ventusproxy.tests.couchbase.TestCB$3.onNext(TestCB.java:1)
at rx.observers.SafeSubscriber.onNext(SafeSubscriber.java:130)

如果onNext方法为：

  public void onNext(AsyncViewRow viewRow) 
                { 
                 System.out.println("viewRow.id() = " + viewRow.id());
                 bucket.async().remove(viewRow.id());
                }

然后，所有文档都会打印在控制台上，但不会被删除

有人能帮我吗

此外，我不确定这是否是删除一大组文档的最有效方法

谢谢

琼

Answer 1

你尝试做的事情有一些问题：

1. startKeyDocId方法实际上并不像你想象的那样。使用 startKey(callId)
2. 仅指定开始键将返回大于的所有结果 参数，而不仅仅是以该键开始的结果。看法 查询按升序按字典顺序比较键，因此您将得到大于callId的所有字符串。您应该同时指定这两个选项 查询中的开始键和结束键，以及inclusive end，可以对其进行适当的限制。所以cacheView.startKey(callId).endKey(calldId).inclusiveEnd(true)（从内存中键入，语法可能会有点不正确。）
3. 您试图一次检索所有匹配的密钥。如果有的话 对于许多结果，查询将超时，因为它需要 太长，无法返回所有结果。如果你期待很多结果（比如 超过1000）您应该使用分页来检索和处理它们 大块。下面是一个很好的例子，演示了如何在RxJava中使用可观察对象实现分页：How to handle paging with RxJava?
4. 异步调用视图查询，但使用阻塞移除调用。与其订阅异步查询调用的结果，不如将其flatMap放入bucket.async().remove(row.id())，然后订阅所有异步删除调用的结果，添加一些重试和错误处理逻辑，等等。请看这里关于实现重试和错误处理的信息：http://developer.couchbase.com/documentation/server/4.0/sdks/java-2.2/observables.html#story-h2-10（具体地滚动到retryWhen示例。）
5. 最后，如果您使用的是Couchbase 4.0，那么可以用以下N1QL查询替换整个内容：DELETE FROM bucket WHERE META(bucket).id like 'ABC%';