CKQueryOperation处理大批处理

Question

我在创建包含大量数据的CKQuery操作时遇到了问题。我的查询可以处理100个结果，但在更多的结果查询失败后，因为一个线程调度错误或其他原因(libdispatch.dylib`dispatch_group_leave：)，我迷路了……有什么想法吗？

+ (void) fetchAnswersWithRecordId:(CKRecordID *)recordId completionHandler:(CloudKitCompletionHandler)handler {
NSPredicate *predicate = [NSPredicate predicateWithFormat:@"ANSrecordID == %@", recordId];
CKQuery *query = [[CKQuery alloc] initWithRecordType:ckAnswers predicate:predicate];

CKQueryOperation *operation = [[CKQueryOperation alloc] initWithQuery:query];
CKQueryOperation * __weak weakSelf = operation;
operation.resultsLimit = 300;
NSMutableArray *tmp = [[NSMutableArray alloc] init];


operation.recordFetchedBlock = ^(CKRecord *record) {
    [tmp addObject:record];
};

operation.queryCompletionBlock = ^(CKQueryCursor *cursor, NSError *error) {
    if (!handler)
        return;

    NSArray *array = [NSArray arrayWithArray:tmp];
    if(cursor != nil) {
        CKQueryOperation *newOperation = [[CKQueryOperation alloc] initWithCursor:cursor];
        newOperation.recordFetchedBlock = weakSelf.recordFetchedBlock;
        newOperation.completionBlock = weakSelf.completionBlock;
        [[self publicCloudDatabase] addOperation:newOperation];
    } else {
        NSLog(@"Results: %lu", [array count]);
        dispatch_async(dispatch_get_main_queue(), ^{
            handler(array, error);
        });
    }
};

[[self publicCloudDatabase] addOperation:operation];}

Answer 1

我认为您的问题出在__weak操作和在另一个操作中创建操作的方式上。下面是一个示例(用swift)，展示了我如何做类似的事情，即获取额外的结果，但是是在fetch中而不是在查询中。注意第一次使用实例变量进行初始化，并通过GCD dispatch_aync使用半递归：

private func _fetchRecordChangesFromCloud() {
    if !_fetching {
        // this is the first and only time this code is called in GCD recusion

        // we clean the caches we use to collect the results of the fetch
        // so we can then save the record in the correct order so references can be created
        _fetchedModifiedRecords = []
        _fetchedDeletedRecordIDs = []

        // mark fetching has started
        _fetching = true
    }

    let operation = CKFetchRecordChangesOperation(recordZoneID: _customRecordZoneID, previousServerChangeToken: _serverChangeToken)

    operation.recordChangedBlock = { (record: CKRecord?) in
        if let record = record {
            println("Received record to save: \(record)")
            self._fetchedModifiedRecords.append(record)
        }
    }

    operation.recordWithIDWasDeletedBlock = { (recordID: CKRecordID?) in
        if let recordID = recordID {
            println("Received recordID to delete: \(recordID)")
            self._fetchedDeletedRecordIDs.append(recordID)
        }
    }

    operation.fetchRecordChangesCompletionBlock = {
        (serverChangeToken: CKServerChangeToken?, clientChangeToken: NSData?, error: NSError?) -> Void in

        if let error = error {
            println("Error in fetching record changes: \(error)")
            // try again next sync
            self._fetchAfterNextSuccessfullSync = true
            self._fetching = false
            return
        }

        // fetched records successfuly
        println("fetched records successfuly")

        if let serverChangeToken = serverChangeToken {
            self._serverChangeToken = serverChangeToken
        }

        if operation.moreComing {
            // we need to create another operation object and do it again
            dispatch_async(dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0)) {
                self._fetchRecordChangesFromCloud()
            }
        } else {
            // we are finally done

            // process the fetched records
            self._processFetchedRecords()

            // save all changes back to persistent store
            self._saveBackgroundContext()

            // we are done
            self._fetching = false
        }
    }

    self._privateDatabase.addOperation(operation)
}

Answer 2

仅供参考，dispatch_async()不是必需的，这是一个内存管理问题。以下内容适用于多批抓取：

//
__block CKQueryOperation* enumerateOperationActive = nil;

//
NSPredicate* predicate = [NSPredicate predicateWithValue:TRUE];
CKQuery* query = [[[CKQuery alloc] initWithRecordType:@"myType" predicate:predicate] autorelease];

CKQueryOperation* enumerateOperation = [[[CKQueryOperation alloc] initWithQuery:query] autorelease];

// DEBUG: fetch only 1 record in order to "force" a nested CKQueryOperation cycle
enumerateOperation.resultsLimit = 1;

enumerateOperation.recordFetchedBlock = ^(CKRecord* recordFetched)
    {
        // ...
    };

enumerateOperation.queryCompletionBlock = ^(CKQueryCursor* cursor, NSError* error)
    {
        if (error)
        {
            // ...
        }
        else
        {
            if (cursor)
            {
                CKQueryOperation* enumerateOperationNested = [[[CKQueryOperation alloc] initWithCursor:cursor] autorelease];

                // DEBUG: fetch only 1 record in order to "force" a doubly-nested CKQueryOperation cycle
                enumerateOperationNested.resultsLimit = 1;

                enumerateOperationNested.recordFetchedBlock = /* re-used */ enumerateOperationActive.recordFetchedBlock;
                enumerateOperationNested.queryCompletionBlock = /* re-used */ enumerateOperationActive.queryCompletionBlock;

                // CRITICAL: keep track of the very last (active) operation
                enumerateOperationActive = enumerateOperationNested;
                [database addOperation:enumerateOperationNested];
            }
        }
    };

//

// CRITICAL: keep track of the very last (active) operation
enumerateOperationActive = enumerateOperation;
[database addOperation:enumerateOperation];

注意:如果您尝试访问(原始) enumerateOperation.queryCompletionBlock而不是(最后一个) enumerateOperationActive.queryCompletionBlock，操作将永远不会完成。

Answer 3

我喜欢主线程上的递归解决方案。下面是我在Objective C中的解决方案。我使用了一个类级变量：_recordArray，提前分配。

- (void) readRecords_Resurs: (CKDatabase *) database query: (CKQuery *) query cursor: (CKQueryCursor *) cursor {

    // Send either query or cursor

    CKQueryOperation *operation;
    if (query != nil) operation = [[CKQueryOperation alloc] initWithQuery: query];
    else operation = [[CKQueryOperation alloc] initWithCursor: cursor];
    operation.recordFetchedBlock = ^(CKRecord *record) {
        [_recordArray addObject:record];
    };
    operation.queryCompletionBlock = ^(CKQueryCursor *cursor, NSError *error) {
        if (cursor == nil || error != nil) {
            // Done
            dispatch_async(dispatch_get_main_queue(), ^{ [self readRecordsDone: error == nil ? nil : [error localizedDescription]]; });
        }
        else {
            // Get next batch using cursor
            dispatch_async(dispatch_get_main_queue(), ^{ [self readRecords_Resurs: database query: nil cursor: cursor]; });
        }
    };

    [database addOperation: operation]; // start
}

- (void) readRecordsDone: (NSString *) errorMsg {
}