1

我正在向 CouchBase 批量插入数据，并在几分钟后收到此异常：

com.couchbase.client.java.error.TemporaryFailureException 在 com.couchbase.client.java.CouchbaseAsyncBucket$13.call(CouchbaseAsyncBucket.java:445) 在 com.couchbase.client.java.CouchbaseAsyncBucket$13.call(CouchbaseAsyncBucket.java:426 ) 在 rx.internal.operators.OperatorMap$1.onNext(OperatorMap.java:54) 在 rx.observers.Subscribers$5.onNext(Subscribers.java:234) 在 rx.subjects.SubjectSubscriptionManager$SubjectObserver.onNext(SubjectSubscriptionManager.java: 222) 在 rx.subjects.AsyncSubject.onCompleted(AsyncSubject.java:101) 在 com.couchbase.client.core.endpoint.AbstractGenericHandler$1.call(AbstractGenericHandler.java:265) 在 rx.internal.schedulers.ScheduledAction.run( ScheduledAction.java:55) 在 java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) 在 java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) 在 java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask 的 java.util.concurrent.FutureTask.run(FutureTask.java:266)。在 java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) 在 java.lang.Thread 的 java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) 运行(ScheduledThreadPoolExecutor.java:293) .run(Thread.java:745)在 java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) 在 java.lang.Thread 的 java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) 运行(ScheduledThreadPoolExecutor.java:293) .run(Thread.java:745)在 java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) 在 java.lang.Thread 的 java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) 运行(ScheduledThreadPoolExecutor.java:293) .run(Thread.java:745)

这是代码：

    cluster = CouchbaseCluster.create();
    Bucket bucket = cluster.openBucket();

    int numBatchs = 10000;
    int numDocsInBatch = 1000;

    for (int j = 0; j < numBatchs; j++) {
        long start = System.currentTimeMillis();

        List<JsonDocument> documents = new ArrayList<>(numDocsInBatch);
        for (int i = 0; i < numDocsInBatch; i++) {
            String uniqueID = UUID.randomUUID().toString() + "_" + System.currentTimeMillis();

            JsonObject user = JsonObject.
                    empty();

            documents.add(JsonDocument.create(uniqueID, user));
        }

        Observable
                .from(documents)
                .flatMap(new Func1<JsonDocument, Observable<JsonDocument>>() {
                    @Override
                    public Observable<JsonDocument> call(final JsonDocument docToInsert) {
                        return bucket.async().insert(docToInsert);
                    }
                })
                .last()
                .toBlocking()
                .single();

有什么我做错了吗？

4

1 回答 1

3

TemporaryFailureException通常是由于服务器可能太忙而引起的，通常可以通过退避重试该操作。

供参考：http: //developer.couchbase.com/documentation/server/4.0/sdks/java-2.2/documents-bulk.html

于 2016-01-14T14:28:34.890 回答